package catalogsnapshot

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"sort"
	"strconv"
	"strings"

	"github.com/nexuscrm/automation/triggerprojection"
	kernelconnector "github.com/nexuscrm/kernel/pkg/connector"
	"github.com/nexuscrm/kernel/pkg/flowactions"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

const (
	unsupportedFlowAcceptCommandActionType = "acceptCommand"
	legacyMutationScopeValidationRule      = "validationRule"
	legacyMutationScopeRecordType          = "recordType"
	legacyMutationScopeTriggerSubscription = "triggerSubscription"
	legacyMutationScopeFlowDefinition      = "flowDefinition"
	legacyMutationScopeFlowVersion         = "flowVersion"
	legacyMutationScopeAutoNumberRule      = "autoNumberRule"
	legacyMutationScopeFieldMappingRule    = "fieldMappingRule"
	legacyMutationScopeCreateSourceGuard   = "createSourceGuard"
	legacyMutationScopeOAuthPublicScope    = "oauthPublicScope"
)

func profileUsesMutationClosure(profile ProjectionProfile) bool {
	return profile == ProjectionRecordMutationV1 || profile == ProjectionApprovalExecutionV1
}

// SupportsMutationClosure reports whether this immutable view carries the
// compiler-validated record-mutation facts consumed by the existing mutation
// engine. Approval execution is a distinct capability profile, but its v16
// artifact deliberately contains the same closed mutation authority; Approval
// read does not.
func (v *RuntimeMetadataView) SupportsMutationClosure() bool {
	return v != nil && profileUsesMutationClosure(v.doc.Profile) && v.doc.Mutation != nil
}

func inputHasMutationProjection(values []RuntimeProjectionInput) bool {
	for _, value := range values {
		if profileUsesMutationClosure(value.Profile) {
			return true
		}
	}
	return false
}

func inputHasConnectorRuntimeFacts(values []MutationDefinitionInput) bool {
	for _, value := range values {
		if value.ConnectorInboundFinalization != nil {
			return true
		}
		if value.ConnectorInboundAuthorship != nil {
			return true
		}
		if value.ConnectorInboundFieldMappingRule != nil {
			return true
		}
	}
	return false
}

func inputHasConnectorFinalizationAuthority(values []MutationDefinitionInput) bool {
	for _, value := range values {
		if value.ConnectorInboundFinalization != nil {
			return true
		}
	}
	return false
}

func inputHasConnectorRecoveryEligibility(values []MutationDefinitionInput) bool {
	for _, value := range values {
		if value.ConnectorInboundFinalization != nil &&
			value.ConnectorInboundFinalization.RecoverySourceEligibility != nil {
			return true
		}
	}
	return false
}

func mutationHasConnectorRuntimeFacts(doc *mutationDefinitionDocument) bool {
	if doc == nil {
		return false
	}
	if len(doc.ConnectorInboundFinalization) != 0 {
		return true
	}
	if len(doc.ConnectorInboundAuthorship) != 0 {
		return true
	}
	return len(doc.ConnectorInboundFieldMappingRules) != 0
}

func mutationHasConnectorFinalizationAuthority(doc *mutationDefinitionDocument) bool {
	return doc != nil && len(doc.ConnectorInboundFinalization) != 0
}

func mutationHasConnectorRecoveryEligibility(doc *mutationDefinitionDocument) bool {
	if doc == nil {
		return false
	}
	for _, value := range doc.ConnectorInboundFinalization {
		if value.RecoverySourceEligibility != nil {
			return true
		}
	}
	return false
}

func inputHasFieldMappingValuePolicies(values []MutationDefinitionInput) bool {
	for _, value := range values {
		if value.FieldMappingRule != nil &&
			(value.FieldMappingRule.ValueMap != "" || value.FieldMappingRule.OnUnmappedValue != "") {
			return true
		}
	}
	return false
}

func mutationHasFieldMappingValuePolicies(doc *mutationDefinitionDocument) bool {
	if doc == nil {
		return false
	}
	for _, value := range doc.FieldMappingRules {
		if value.ValueMap != "" || value.OnUnmappedValue != "" {
			return true
		}
	}
	return false
}

func runtimeViewSchemaFor(snapshotSchema string, profile ProjectionProfile) string {
	if profile == ProjectionLayoutAuthoringV1 {
		if snapshotSchema == CatalogSnapshotSchemaV21 {
			return RuntimeMetadataViewSchemaV28
		}
		return RuntimeMetadataViewSchemaV27
	}
	if profile == ProjectionPageLayoutResolutionV1 {
		return RuntimeMetadataViewSchemaV23
	}
	if profile == ProjectionEffectivePermissionsV1 {
		return RuntimeMetadataViewSchemaV21
	}
	if profileUsesApprovalAuthority(profile) {
		return RuntimeMetadataViewSchemaV16
	}
	if profile == ProjectionCommandAcceptanceV1 {
		return RuntimeMetadataViewSchemaV11
	}
	if profile == ProjectionReportExecutionV1 {
		return RuntimeMetadataViewSchemaV10
	}
	if profile == ProjectionRecordAccessV1 {
		return RuntimeMetadataViewSchemaV8
	}
	if profile == ProjectionRecordViewV1 {
		if snapshotSchema == CatalogSnapshotSchemaV6 || snapshotSchema == CatalogSnapshotSchemaV8 || snapshotSchema == CatalogSnapshotSchemaV9 || snapshotSchema == CatalogSnapshotSchemaV10 || snapshotSchema == CatalogSnapshotSchemaV11 || snapshotSchema == CatalogSnapshotSchemaV12 || snapshotSchema == CatalogSnapshotSchemaV13 || snapshotSchema == CatalogSnapshotSchemaV14 || snapshotSchema == CatalogSnapshotSchemaV15 || snapshotSchema == CatalogSnapshotSchemaV16 || snapshotSchema == CatalogSnapshotSchemaV17 || snapshotSchema == CatalogSnapshotSchemaV18 || snapshotSchema == CatalogSnapshotSchemaV19 || (snapshotSchema == CatalogSnapshotSchemaV20 || snapshotSchema == CatalogSnapshotSchemaV21) {
			// V24 carries both the Catalog-declared display-field bit and the
			// complete object facts needed to compute the access envelope that
			// record-view already returns. This is intentionally not a Catalog
			// snapshot bump: V6 already owns all of these source facts.
			return RuntimeMetadataViewSchemaV24
		}
		return RuntimeMetadataViewSchemaV5
	}
	// Later snapshots add only profile-specific report/command or record-view
	// facts. Other profiles still project exactly the v4 grammar and do not
	// inherit that authority.
	if snapshotSchema == CatalogSnapshotSchemaV5 || snapshotSchema == CatalogSnapshotSchemaV6 || snapshotSchema == CatalogSnapshotSchemaV8 || snapshotSchema == CatalogSnapshotSchemaV9 || snapshotSchema == CatalogSnapshotSchemaV10 || snapshotSchema == CatalogSnapshotSchemaV11 || snapshotSchema == CatalogSnapshotSchemaV12 || snapshotSchema == CatalogSnapshotSchemaV13 || snapshotSchema == CatalogSnapshotSchemaV14 || snapshotSchema == CatalogSnapshotSchemaV15 || snapshotSchema == CatalogSnapshotSchemaV16 || snapshotSchema == CatalogSnapshotSchemaV17 || snapshotSchema == CatalogSnapshotSchemaV18 || snapshotSchema == CatalogSnapshotSchemaV19 || (snapshotSchema == CatalogSnapshotSchemaV20 || snapshotSchema == CatalogSnapshotSchemaV21) {
		return RuntimeMetadataViewSchemaV4
	}
	if snapshotSchema == CatalogSnapshotSchemaV4 || profile == ProjectionRecordMutationV1 {
		return RuntimeMetadataViewSchemaV4
	}
	if snapshotSchema == CatalogSnapshotSchemaV3 {
		return RuntimeMetadataViewSchemaV3
	}
	return RuntimeMetadataViewSchemaV2
}

func runtimeMutationViewSchemaFor(snapshotSchema string, projected *mutationDefinitionDocument) string {
	if (snapshotSchema == CatalogSnapshotSchemaV18 || snapshotSchema == CatalogSnapshotSchemaV19 || (snapshotSchema == CatalogSnapshotSchemaV20 || snapshotSchema == CatalogSnapshotSchemaV21)) && mutationHasFieldMappingValuePolicies(projected) {
		return RuntimeMetadataViewSchemaV25
	}
	// Runtime view schemas describe the projected artifact, not unrelated facts
	// present elsewhere in the source Catalog snapshot. Current mutation
	// projections require the full object closure, but deriving V13 here keeps
	// that invariant local and prevents a future projection change from emitting
	// a V13 artifact after filtering all connector execution facts.
	if (snapshotSchema == CatalogSnapshotSchemaV13 || snapshotSchema == CatalogSnapshotSchemaV14 || snapshotSchema == CatalogSnapshotSchemaV15 || snapshotSchema == CatalogSnapshotSchemaV16 || snapshotSchema == CatalogSnapshotSchemaV17 || snapshotSchema == CatalogSnapshotSchemaV18 || snapshotSchema == CatalogSnapshotSchemaV19 || (snapshotSchema == CatalogSnapshotSchemaV20 || snapshotSchema == CatalogSnapshotSchemaV21)) && mutationHasConnectorRecoveryEligibility(projected) {
		return RuntimeMetadataViewSchemaV15
	}
	if (snapshotSchema == CatalogSnapshotSchemaV12 || snapshotSchema == CatalogSnapshotSchemaV13 || snapshotSchema == CatalogSnapshotSchemaV14 || snapshotSchema == CatalogSnapshotSchemaV15 || snapshotSchema == CatalogSnapshotSchemaV16 || snapshotSchema == CatalogSnapshotSchemaV17 || snapshotSchema == CatalogSnapshotSchemaV18 || snapshotSchema == CatalogSnapshotSchemaV19 || (snapshotSchema == CatalogSnapshotSchemaV20 || snapshotSchema == CatalogSnapshotSchemaV21)) && mutationHasConnectorFinalizationAuthority(projected) {
		return RuntimeMetadataViewSchemaV14
	}
	if (snapshotSchema == CatalogSnapshotSchemaV11 || snapshotSchema == CatalogSnapshotSchemaV12 || snapshotSchema == CatalogSnapshotSchemaV13 || snapshotSchema == CatalogSnapshotSchemaV14 || snapshotSchema == CatalogSnapshotSchemaV15 || snapshotSchema == CatalogSnapshotSchemaV16 || snapshotSchema == CatalogSnapshotSchemaV17 || snapshotSchema == CatalogSnapshotSchemaV18 || snapshotSchema == CatalogSnapshotSchemaV19 || (snapshotSchema == CatalogSnapshotSchemaV20 || snapshotSchema == CatalogSnapshotSchemaV21)) && mutationHasConnectorRuntimeFacts(projected) {
		return RuntimeMetadataViewSchemaV13
	}
	return runtimeViewSchemaFor(snapshotSchema, ProjectionRecordMutationV1)
}

func catalogSchemaForRuntimeViewSchema(schema string) string {
	switch schema {
	case RuntimeMetadataViewSchemaV28:
		return CatalogSnapshotSchemaV21
	case RuntimeMetadataViewSchemaV27:
		return CatalogSnapshotSchemaV20
	case RuntimeMetadataViewSchemaV26:
		return CatalogSnapshotSchemaV19
	case RuntimeMetadataViewSchemaV25:
		return CatalogSnapshotSchemaV18
	case RuntimeMetadataViewSchemaV24:
		return CatalogSnapshotSchemaV6
	case RuntimeMetadataViewSchemaV23:
		return CatalogSnapshotSchemaV17
	case RuntimeMetadataViewSchemaV22:
		return CatalogSnapshotSchemaV16
	case RuntimeMetadataViewSchemaV21:
		return CatalogSnapshotSchemaV15
	case RuntimeMetadataViewSchemaV19:
		return CatalogSnapshotSchemaV15
	case RuntimeMetadataViewSchemaV18:
		return CatalogSnapshotSchemaV4
	case RuntimeMetadataViewSchemaV17:
		return CatalogSnapshotSchemaV15
	case RuntimeMetadataViewSchemaV16:
		return CatalogSnapshotSchemaV14
	case RuntimeMetadataViewSchemaV15:
		return CatalogSnapshotSchemaV13
	case RuntimeMetadataViewSchemaV14:
		return CatalogSnapshotSchemaV12
	case RuntimeMetadataViewSchemaV13:
		return CatalogSnapshotSchemaV11
	case RuntimeMetadataViewSchemaV12:
		return CatalogSnapshotSchemaV10
	case RuntimeMetadataViewSchemaV11:
		return CatalogSnapshotSchemaV9
	case RuntimeMetadataViewSchemaV10:
		return CatalogSnapshotSchemaV8
	case RuntimeMetadataViewSchemaV8:
		// V8 is profile-specific and can be projected from every supported
		// Catalog snapshot schema because object KeyPrefix predates V2.
		return CatalogSnapshotSchemaV2
	case RuntimeMetadataViewSchemaV7:
		return CatalogSnapshotSchemaV6
	case RuntimeMetadataViewSchemaV6:
		return CatalogSnapshotSchemaV6
	case RuntimeMetadataViewSchemaV5:
		return CatalogSnapshotSchemaV5
	case RuntimeMetadataViewSchemaV4:
		return CatalogSnapshotSchemaV4
	case RuntimeMetadataViewSchemaV3:
		return CatalogSnapshotSchemaV3
	default:
		return CatalogSnapshotSchemaV2
	}
}

func emptyMutationDefinitionDocument() *mutationDefinitionDocument {
	return &mutationDefinitionDocument{
		ValidationRules: []models.ValidationRule{}, RecordTypes: []models.RecordType{},
		TriggerSubscriptions: []TriggerSubscription{}, FlowDefinitions: []runtimeFlowDefinitionRecord{},
		FlowVersions: []runtimeFlowVersionRecord{}, AutoNumberRules: []AutoNumberRule{},
		FieldMappingRules: []FieldMappingRule{}, CreateSourceGuards: []CreateFromSourceGuardRule{},
		ConnectorInboundFieldMappingRules: []ConnectorInboundFieldMappingRule{},
		ConnectorInboundAuthorship:        []ConnectorInboundAuthorshipBinding{},
		ConnectorInboundFinalization:      []ConnectorInboundFinalizationAuthority{},
		OAuthPublicScopes:                 []OAuthPublicScopeRule{},
	}
}

func buildMutationDefinitionDocument(
	revision CatalogRevision,
	inputs []MutationDefinitionInput,
	objects map[string]*models.ObjectMetadata,
	formulaEngine *formula.Engine,
) (*mutationDefinitionDocument, error) {
	if len(inputs) == 0 {
		return nil, nil
	}
	doc := emptyMutationDefinitionDocument()
	seen := map[string]struct{}{}
	for i, input := range inputs {
		path := fmt.Sprintf("mutationDefinitions[%d]", i)
		if err := requireSourceRevision(path, revision, input.CatalogRevision); err != nil {
			return nil, err
		}
		arms := 0
		for _, present := range []bool{
			input.ValidationRule != nil, input.RecordType != nil, input.TriggerSubscription != nil,
			input.FlowDefinition != nil, input.FlowVersion != nil, input.AutoNumberRule != nil,
			input.FieldMappingRule != nil, input.ConnectorInboundFieldMappingRule != nil,
			input.ConnectorInboundAuthorship != nil,
			input.ConnectorInboundFinalization != nil,
			input.CreateSourceGuard != nil, input.OAuthPublicScope != nil,
		} {
			if present {
				arms++
			}
		}
		if arms != 1 {
			return nil, conflict(ConflictInvalidDefinition, path, "mutation definition must declare exactly one typed fact")
		}

		switch {
		case input.ValidationRule != nil:
			rule, err := cloneJSON(*input.ValidationRule)
			if err != nil {
				return nil, conflict(ConflictEncoding, path, "clone validation rule: %v", err)
			}
			rule.Description = nil
			if err := requireMutationObject(path+".validationRule.objectApiName", rule.ObjectAPIName, objects); err != nil {
				return nil, err
			}
			if err := appendUnique(seen, "validationRule", rule.ID, path); err != nil {
				return nil, err
			}
			doc.ValidationRules = append(doc.ValidationRules, rule)
		case input.RecordType != nil:
			recordType := *input.RecordType
			// Name and DeveloperName are execution identities consumed by the
			// neutral record-type access surface and validation engine. They are
			// not optional presentation metadata. Label and Description remain
			// outside the least-privilege Runtime mutation projection.
			recordType.Label, recordType.Description = "", nil
			recordType.CreatedDate, recordType.LastModifiedDate = recordType.CreatedDate.UTC(), recordType.LastModifiedDate.UTC()
			if err := requireMutationObject(path+".recordType.objectApiName", recordType.ObjectAPIName, objects); err != nil {
				return nil, err
			}
			if err := appendUnique(seen, "recordType", recordType.ID, path); err != nil {
				return nil, err
			}
			doc.RecordTypes = append(doc.RecordTypes, recordType)
		case input.TriggerSubscription != nil:
			subscription := *input.TriggerSubscription
			for name, value := range map[string]string{
				"id": subscription.ID, "keyPrefix": subscription.KeyPrefix,
				"eventType": subscription.EventType, "recordTriggerType": subscription.RecordTriggerType,
				"flowDefinitionId": subscription.FlowDefinitionID, "flowVersionId": subscription.FlowVersionID,
			} {
				if err := requireCanonicalIdentity(path+".triggerSubscription."+name, value); err != nil {
					return nil, err
				}
			}
			if !subscription.IsActive {
				return nil, conflict(ConflictInvalidDefinition, path+".triggerSubscription.isActive", "projected trigger subscription must be active")
			}
			if err := appendUnique(seen, "triggerSubscription", subscription.ID, path); err != nil {
				return nil, err
			}
			doc.TriggerSubscriptions = append(doc.TriggerSubscriptions, subscription)
		case input.FlowDefinition != nil:
			definition := runtimeFlowDefinitionRecord{
				ID: input.FlowDefinition.Id, DeveloperName: input.FlowDefinition.DeveloperName,
				ProcessType: input.FlowDefinition.ProcessType, TriggerType: input.FlowDefinition.TriggerType,
				TriggerObjectOrEvent: input.FlowDefinition.TriggerObjectOrEvent,
				ActiveVersionID:      input.FlowDefinition.ActiveVersionId, IsDeleted: input.FlowDefinition.IsDeleted,
				IsCustomized: input.FlowDefinition.IsCustomized, OwnerPack: input.FlowDefinition.OwnerPack,
			}
			if err := appendUnique(seen, "flowDefinition", definition.ID, path); err != nil {
				return nil, err
			}
			doc.FlowDefinitions = append(doc.FlowDefinitions, definition)
		case input.FlowVersion != nil:
			metadata, err := canonicalFlowMetadata(input.FlowVersion)
			if err != nil {
				return nil, conflict(ConflictInvalidDefinition, path+".flowVersion.metadata", "%v", err)
			}
			version := runtimeFlowVersionRecord{
				ID: input.FlowVersion.Id, DefinitionID: input.FlowVersion.DefinitionId,
				VersionNumber: input.FlowVersion.VersionNumber, Status: input.FlowVersion.Status,
				ProcessType: input.FlowVersion.ProcessType, Source: input.FlowVersion.Source,
				DefinitionIsCustomized: input.FlowVersion.DefinitionIsCustomized,
				DefinitionOwnerPack:    input.FlowVersion.DefinitionOwnerPack,
				IsDeleted:              input.FlowVersion.IsDeleted, Metadata: metadata,
			}
			if err := appendUnique(seen, "flowVersion", version.ID, path); err != nil {
				return nil, err
			}
			doc.FlowVersions = append(doc.FlowVersions, version)
		case input.AutoNumberRule != nil:
			rule, err := cloneJSON(*input.AutoNumberRule)
			if err != nil {
				return nil, conflict(ConflictEncoding, path, "clone autonumber rule: %v", err)
			}
			if err := validateAutoNumberRuleBinding(rule, objects); err != nil {
				return nil, conflict(ConflictInvalidDefinition, path+".autoNumberRule", "%v", err)
			}
			if err := appendUnique(seen, "autoNumberRule", rule.FieldID, path); err != nil {
				return nil, err
			}
			doc.AutoNumberRules = append(doc.AutoNumberRules, rule)
		case input.FieldMappingRule != nil:
			rule := *input.FieldMappingRule
			for _, item := range []struct{ name, value string }{
				{"sourceObject", rule.SourceObject}, {"targetObject", rule.TargetObject},
				{"actionType", rule.ActionType}, {"targetField", rule.TargetField},
			} {
				if err := requireCanonicalIdentity(path+".fieldMappingRule."+item.name, item.value); err != nil {
					return nil, err
				}
			}
			for _, item := range []struct{ name, value string }{
				{"sourceField", rule.SourceField}, {"sourceLookupField", rule.SourceLookupField},
			} {
				if item.value != "" {
					if err := requireCanonicalIdentity(path+".fieldMappingRule."+item.name, item.value); err != nil {
						return nil, err
					}
				}
			}
			if err := requireMutationObject(path+".fieldMappingRule.sourceObject", rule.SourceObject, objects); err != nil {
				return nil, err
			}
			if err := requireMutationObject(path+".fieldMappingRule.targetObject", rule.TargetObject, objects); err != nil {
				return nil, err
			}
			if err := validateFieldMappingRule(rule, objects); err != nil {
				return nil, conflict(ConflictInvalidDefinition, path+".fieldMappingRule", "%v", err)
			}
			key := fieldMappingRuleKey(rule)
			if _, exists := seen["fieldMappingRule\x00"+key]; exists {
				return nil, conflict(ConflictInvalidDefinition, path, "field mapping rule %q is duplicated", key)
			}
			seen["fieldMappingRule\x00"+key] = struct{}{}
			doc.FieldMappingRules = append(doc.FieldMappingRules, rule)
		case input.ConnectorInboundFieldMappingRule != nil:
			rule := *input.ConnectorInboundFieldMappingRule
			if err := validateConnectorInboundFieldMappingRule(rule, objects, formulaEngine); err != nil {
				return nil, conflict(ConflictInvalidDefinition, path+".connectorInboundFieldMappingRule", "%v", err)
			}
			key := connectorInboundFieldMappingRuleKey(rule)
			if _, exists := seen["connectorInboundFieldMappingRule\x00"+key]; exists {
				return nil, conflict(ConflictInvalidDefinition, path, "connector inbound field mapping rule %q is duplicated", key)
			}
			seen["connectorInboundFieldMappingRule\x00"+key] = struct{}{}
			doc.ConnectorInboundFieldMappingRules = append(doc.ConnectorInboundFieldMappingRules, rule)
		case input.ConnectorInboundAuthorship != nil:
			binding := *input.ConnectorInboundAuthorship
			if err := validateConnectorInboundAuthorshipBinding(binding, objects); err != nil {
				return nil, conflict(ConflictInvalidDefinition, path+".connectorInboundAuthorship", "%v", err)
			}
			key := connectorInboundAuthorshipBindingKey(binding)
			if _, exists := seen["connectorInboundAuthorship\x00"+key]; exists {
				return nil, conflict(ConflictInvalidDefinition, path, "connector inbound authorship binding %q is duplicated", key)
			}
			seen["connectorInboundAuthorship\x00"+key] = struct{}{}
			doc.ConnectorInboundAuthorship = append(doc.ConnectorInboundAuthorship, binding)
		case input.ConnectorInboundFinalization != nil:
			authority, err := cloneJSON(*input.ConnectorInboundFinalization)
			if err != nil {
				return nil, conflict(ConflictEncoding, path, "clone connector inbound finalization authority: %v", err)
			}
			normalizeConnectorInboundFinalizationAuthority(&authority, objects)
			if err := validateConnectorInboundFinalizationAuthority(authority, objects, formulaEngine); err != nil {
				return nil, conflict(ConflictInvalidDefinition, path+".connectorInboundFinalization", "%v", err)
			}
			key := connectorInboundFinalizationAuthorityKey(authority)
			if _, exists := seen["connectorInboundFinalization\x00"+key]; exists {
				return nil, conflict(ConflictInvalidDefinition, path, "connector inbound finalization authority %q is duplicated", key)
			}
			seen["connectorInboundFinalization\x00"+key] = struct{}{}
			doc.ConnectorInboundFinalization = append(doc.ConnectorInboundFinalization, authority)
		case input.CreateSourceGuard != nil:
			rule := *input.CreateSourceGuard
			for _, item := range []struct{ name, value string }{
				{"targetObject", rule.TargetObject}, {"sourceFkField", rule.SourceFKField},
				{"owningStandardActionId", rule.OwningStandardActionID}, {"actionName", rule.ActionName},
			} {
				if err := requireCanonicalIdentity(path+".createSourceGuard."+item.name, item.value); err != nil {
					return nil, err
				}
			}
			if err := requireMutationObject(path+".createSourceGuard.targetObject", rule.TargetObject, objects); err != nil {
				return nil, err
			}
			field := models.FindField(objects[foldIdentity(rule.TargetObject)], rule.SourceFKField)
			if !isCreateSourceGuardField(field) {
				return nil, conflict(ConflictInvalidDefinition, path+".createSourceGuard.sourceFkField", "field %q is not a reference-like field on object %q", rule.SourceFKField, rule.TargetObject)
			}
			key := createSourceGuardRuleKey(rule)
			if _, exists := seen["createSourceGuard\x00"+key]; exists {
				return nil, conflict(ConflictInvalidDefinition, path, "create-from-source guard %q is duplicated", key)
			}
			seen["createSourceGuard\x00"+key] = struct{}{}
			doc.CreateSourceGuards = append(doc.CreateSourceGuards, rule)
		case input.OAuthPublicScope != nil:
			rule := *input.OAuthPublicScope
			rule.ObjectAPIName = strings.TrimSpace(rule.ObjectAPIName)
			rule.Name = strings.ToLower(strings.TrimSpace(rule.Name))
			rule.OwnerPack = strings.TrimSpace(rule.OwnerPack)
			if err := requireMutationObject(path+".oauthPublicScope.objectApiName", rule.ObjectAPIName, objects); err != nil {
				return nil, err
			}
			if err := requireCanonicalIdentity(path+".oauthPublicScope.name", rule.Name); err != nil {
				return nil, err
			}
			if rule.OwnerPack != "" {
				if err := requireCanonicalIdentity(path+".oauthPublicScope.ownerPack", rule.OwnerPack); err != nil {
					return nil, err
				}
			}
			key := oauthPublicScopeRuleKey(rule)
			if _, exists := seen["oauthPublicScope\x00"+key]; exists {
				return nil, conflict(ConflictInvalidDefinition, path, "OAuth public scope rule %q is duplicated", key)
			}
			seen["oauthPublicScope\x00"+key] = struct{}{}
			doc.OAuthPublicScopes = append(doc.OAuthPublicScopes, rule)
		}
	}
	sortMutationDefinitions(doc)
	if err := validateMutationDefinitions(doc, objects, formulaEngine); err != nil {
		return nil, conflict(ConflictInvalidDefinition, "mutationDefinitions", "%v", err)
	}
	return doc, nil
}

// applyRuntimeMutationDerivedObjectFacts closes facts that are authored once
// but consumed through the effective object shape after the exact object
// closure has been selected. An autonumber rule means the named field is
// populated by the server during create, so DefaultedOnCreate is a
// deterministic consequence of that rule rather than a second authored bit.
// Applying it to the immutable projection copy avoids turning the derived bit
// into a second global Catalog authority.
func applyRuntimeMutationDerivedObjectFacts(source, selected *mutationDefinitionDocument, objects []runtimeObjectRecord) error {
	if selected == nil {
		return nil
	}
	objectsByName := make(map[string]*runtimeObjectRecord, len(objects))
	for i := range objects {
		objectsByName[foldIdentity(objects[i].APIName)] = &objects[i]
	}
	findField := func(rule AutoNumberRule) (*runtimeMutationFieldRecord, error) {
		object := objectsByName[foldIdentity(rule.ObjectAPIName)]
		if object == nil {
			return nil, fmt.Errorf("autonumber rule %q references absent object %q", rule.FieldID, rule.ObjectAPIName)
		}
		for i := range object.Fields {
			candidate := &object.Fields[i]
			if candidate.Mutation != nil && foldIdentity(candidate.APIName) == foldIdentity(rule.FieldAPIName) &&
				foldIdentity(candidate.Mutation.ID) == foldIdentity(rule.FieldID) {
				return candidate.Mutation, nil
			}
		}
		return nil, fmt.Errorf("autonumber rule %q does not identify %s.%s", rule.FieldID, rule.ObjectAPIName, rule.FieldAPIName)
	}
	return applyAutoNumberDefaultedFacts(source, selected, func(name string) bool {
		return objectsByName[foldIdentity(name)] != nil
	}, func(rule AutoNumberRule, defaulted bool) error {
		field, err := findField(rule)
		if err != nil {
			return err
		}
		field.DefaultedOnCreate = defaulted
		return nil
	})
}

// applyAutoNumberDefaultedFacts is the single owner of the derived default bit.
// The callbacks only adapt the immutable Catalog and Runtime field carriers.
func applyAutoNumberDefaultedFacts(
	source, selected *mutationDefinitionDocument,
	containsObject func(string) bool,
	setDefaulted func(AutoNumberRule, bool) error,
) error {
	if selected == nil {
		return nil
	}
	// Clear source-derived bits before applying the selected same-revision
	// object closure. Rules apply regardless of the App initiating the write.
	if source != nil {
		for _, rule := range source.AutoNumberRules {
			if !containsObject(rule.ObjectAPIName) {
				continue
			}
			if err := setDefaulted(rule, false); err != nil {
				return err
			}
		}
	}
	for _, rule := range selected.AutoNumberRules {
		if err := setDefaulted(rule, true); err != nil {
			return err
		}
	}
	return nil
}

func validateAutoNumberRuleBinding(rule AutoNumberRule, objects map[string]*models.ObjectMetadata) error {
	for _, item := range []struct{ name, identity string }{
		{name: "objectApiName", identity: rule.ObjectAPIName},
		{name: "fieldApiName", identity: rule.FieldAPIName},
		{name: "fieldId", identity: rule.FieldID},
	} {
		if err := requireCanonicalIdentity("autonumberRule."+item.name, item.identity); err != nil {
			return err
		}
	}
	if objects == nil {
		return nil
	}
	object := objects[foldIdentity(rule.ObjectAPIName)]
	if object == nil {
		return fmt.Errorf("object %q is absent from this revision", rule.ObjectAPIName)
	}
	field := models.FindField(object, rule.FieldAPIName)
	if field == nil {
		return fmt.Errorf("field %q is absent from object %q", rule.FieldAPIName, rule.ObjectAPIName)
	}
	if foldIdentity(field.ID) != foldIdentity(rule.FieldID) {
		return fmt.Errorf("field id %q does not identify %s.%s", rule.FieldID, rule.ObjectAPIName, rule.FieldAPIName)
	}
	return nil
}

func requireMutationObject(path, objectAPIName string, objects map[string]*models.ObjectMetadata) error {
	if err := requireCanonicalIdentity(path, objectAPIName); err != nil {
		return err
	}
	if objects[foldIdentity(objectAPIName)] == nil {
		return conflict(ConflictInvalidDefinition, path, "object %q is absent from this revision", objectAPIName)
	}
	return nil
}

func appendUnique(seen map[string]struct{}, kind, identity, path string) error {
	if err := requireCanonicalIdentity(path+"."+kind+".id", identity); err != nil {
		return err
	}
	key := kind + "\x00" + foldIdentity(identity)
	if _, exists := seen[key]; exists {
		return conflict(ConflictInvalidDefinition, path, "%s %q is duplicated", kind, identity)
	}
	seen[key] = struct{}{}
	return nil
}

func canonicalFlowMetadata(version *models.FlowVersion) (json.RawMessage, error) {
	var source []byte
	switch {
	case len(version.MetadataRaw) != 0:
		source = version.MetadataRaw
	case strings.TrimSpace(version.MetadataJSON) != "":
		source = []byte(version.MetadataJSON)
	default:
		encoded, err := json.Marshal(version.Metadata)
		if err != nil {
			return nil, err
		}
		source = encoded
	}
	decoder := json.NewDecoder(bytes.NewReader(source))
	decoder.UseNumber()
	var value any
	if err := decoder.Decode(&value); err != nil {
		return nil, err
	}
	if err := decoder.Decode(&struct{}{}); err != io.EOF {
		return nil, fmt.Errorf("trailing flow metadata")
	}
	canonical, err := json.Marshal(value)
	if err != nil {
		return nil, err
	}
	if len(canonical) == 0 || bytes.Equal(canonical, []byte("null")) {
		return nil, fmt.Errorf("flow metadata must be an object")
	}
	return canonical, nil
}

func sortMutationDefinitions(doc *mutationDefinitionDocument) {
	sort.Slice(doc.ValidationRules, func(i, j int) bool { return identityLess(doc.ValidationRules[i].ID, doc.ValidationRules[j].ID) })
	sort.Slice(doc.RecordTypes, func(i, j int) bool { return identityLess(doc.RecordTypes[i].ID, doc.RecordTypes[j].ID) })
	sort.Slice(doc.TriggerSubscriptions, func(i, j int) bool {
		return identityLess(doc.TriggerSubscriptions[i].ID, doc.TriggerSubscriptions[j].ID)
	})
	sort.Slice(doc.FlowDefinitions, func(i, j int) bool { return identityLess(doc.FlowDefinitions[i].ID, doc.FlowDefinitions[j].ID) })
	sort.Slice(doc.FlowVersions, func(i, j int) bool { return identityLess(doc.FlowVersions[i].ID, doc.FlowVersions[j].ID) })
	sort.Slice(doc.AutoNumberRules, func(i, j int) bool {
		return identityLess(doc.AutoNumberRules[i].FieldID, doc.AutoNumberRules[j].FieldID)
	})
	sort.Slice(doc.FieldMappingRules, func(i, j int) bool {
		return fieldMappingRuleKey(doc.FieldMappingRules[i]) < fieldMappingRuleKey(doc.FieldMappingRules[j])
	})
	sort.Slice(doc.ConnectorInboundFieldMappingRules, func(i, j int) bool {
		return connectorInboundFieldMappingRuleKey(doc.ConnectorInboundFieldMappingRules[i]) <
			connectorInboundFieldMappingRuleKey(doc.ConnectorInboundFieldMappingRules[j])
	})
	sort.Slice(doc.ConnectorInboundAuthorship, func(i, j int) bool {
		return connectorInboundAuthorshipBindingKey(doc.ConnectorInboundAuthorship[i]) <
			connectorInboundAuthorshipBindingKey(doc.ConnectorInboundAuthorship[j])
	})
	sort.Slice(doc.ConnectorInboundFinalization, func(i, j int) bool {
		return connectorInboundFinalizationAuthorityKey(doc.ConnectorInboundFinalization[i]) <
			connectorInboundFinalizationAuthorityKey(doc.ConnectorInboundFinalization[j])
	})
	sort.Slice(doc.CreateSourceGuards, func(i, j int) bool {
		return createSourceGuardRuleKey(doc.CreateSourceGuards[i]) < createSourceGuardRuleKey(doc.CreateSourceGuards[j])
	})
	sort.Slice(doc.OAuthPublicScopes, func(i, j int) bool {
		return oauthPublicScopeRuleKey(doc.OAuthPublicScopes[i]) < oauthPublicScopeRuleKey(doc.OAuthPublicScopes[j])
	})
}

func fieldMappingRuleKey(value FieldMappingRule) string {
	return foldIdentity(value.SourceObject) + "\x00" + foldIdentity(value.TargetObject) + "\x00" +
		foldIdentity(value.ActionType) + "\x00" + foldIdentity(value.TargetField)
}

func connectorInboundFieldMappingRuleKey(value ConnectorInboundFieldMappingRule) string {
	key := foldIdentity(value.Source) + "\x00" + foldIdentity(value.SourceObject) + "\x00" +
		foldIdentity(value.TargetObject) + "\x00" + foldIdentity(value.ActionType) + "\x00" +
		foldIdentity(value.TargetField)
	if value.PackAuthored {
		return key + "\x00pack-authored"
	}
	return key
}

func connectorInboundAuthorshipBindingKey(value ConnectorInboundAuthorshipBinding) string {
	return foldIdentity(value.ConnectorOwnerKey) + "\x00" +
		foldIdentity(value.SourceObject) + "\x00" +
		foldIdentity(value.RemoteIDField)
}

func connectorInboundFinalizationAuthorityKey(value ConnectorInboundFinalizationAuthority) string {
	return foldIdentity(value.Contract.ConnectorOwnerKey) + "\x00" +
		foldIdentity(value.Contract.FlowDeveloperName) + "\x00" +
		foldIdentity(value.Contract.FlowActionName)
}

func createSourceGuardRuleKey(value CreateFromSourceGuardRule) string {
	return foldIdentity(value.TargetObject) + "\x00" + foldIdentity(value.SourceFKField) + "\x00" +
		foldIdentity(value.OwningStandardActionID)
}

func oauthPublicScopeRuleKey(value OAuthPublicScopeRule) string {
	return foldIdentity(value.ObjectAPIName) + "\x00" + foldIdentity(value.Name) + "\x00" + value.OwnerPack
}

func legacyMutationScopeToken(kind, factKey string) string {
	return kind + "\x00" + foldIdentity(factKey)
}

func legacyMutationFactTokens(doc *mutationDefinitionDocument) map[string]struct{} {
	tokens := map[string]struct{}{}
	if doc == nil {
		return tokens
	}
	for _, value := range doc.ValidationRules {
		tokens[legacyMutationScopeToken(legacyMutationScopeValidationRule, value.ID)] = struct{}{}
	}
	for _, value := range doc.RecordTypes {
		tokens[legacyMutationScopeToken(legacyMutationScopeRecordType, value.ID)] = struct{}{}
	}
	for _, value := range doc.TriggerSubscriptions {
		tokens[legacyMutationScopeToken(legacyMutationScopeTriggerSubscription, value.ID)] = struct{}{}
	}
	for _, value := range doc.FlowDefinitions {
		tokens[legacyMutationScopeToken(legacyMutationScopeFlowDefinition, value.ID)] = struct{}{}
	}
	for _, value := range doc.FlowVersions {
		tokens[legacyMutationScopeToken(legacyMutationScopeFlowVersion, value.ID)] = struct{}{}
	}
	for _, value := range doc.AutoNumberRules {
		tokens[legacyMutationScopeToken(legacyMutationScopeAutoNumberRule, value.FieldID)] = struct{}{}
	}
	for _, value := range doc.FieldMappingRules {
		tokens[legacyMutationScopeToken(legacyMutationScopeFieldMappingRule, fieldMappingRuleKey(value))] = struct{}{}
	}
	for _, value := range doc.CreateSourceGuards {
		tokens[legacyMutationScopeToken(legacyMutationScopeCreateSourceGuard, createSourceGuardRuleKey(value))] = struct{}{}
	}
	for _, value := range doc.OAuthPublicScopes {
		tokens[legacyMutationScopeToken(legacyMutationScopeOAuthPublicScope, oauthPublicScopeRuleKey(value))] = struct{}{}
	}
	return tokens
}

// validateLegacyMutationScopes validates the closed v3 serialization grammar
// without restoring its scope index as serving authority. Projection ignores
// this association and emits a v4 view from the complete effective closure.
func validateLegacyMutationScopes(doc *mutationDefinitionDocument, definitions map[string]struct{}) error {
	if doc == nil {
		return nil
	}
	expected := legacyMutationFactTokens(doc)
	seen := make(map[string]struct{}, len(doc.LegacyScopes))
	for i, scope := range doc.LegacyScopes {
		if err := requireCanonicalIdentity(fmt.Sprintf("mutation.scopes[%d].componentId", i), scope.ComponentID); err != nil {
			return err
		}
		if _, ok := definitions[foldIdentity(scope.ComponentID)]; !ok {
			return fmt.Errorf("mutation scope %d references absent definition component %q", i, scope.ComponentID)
		}
		token := legacyMutationScopeToken(scope.Kind, scope.FactKey)
		if _, ok := expected[token]; !ok {
			return fmt.Errorf("mutation scope %d references absent %q fact %q", i, scope.Kind, scope.FactKey)
		}
		if _, duplicate := seen[token]; duplicate {
			return fmt.Errorf("mutation fact %q has more than one legacy component association", token)
		}
		seen[token] = struct{}{}
	}
	if len(seen) != len(expected) {
		for token := range expected {
			if _, ok := seen[token]; !ok {
				return fmt.Errorf("mutation fact %q has no legacy component association", token)
			}
		}
	}
	return nil
}

func buildRuntimeMutationObject(source *models.ObjectMetadata) (*runtimeMutationObjectRecord, error) {
	copy, err := cloneJSON(*source)
	if err != nil {
		return nil, err
	}
	return &runtimeMutationObjectRecord{
		KeyPrefix: copy.KeyPrefix, IsSystem: copy.IsSystem, IsCustom: copy.IsCustom,
		CustomSetting: copy.CustomSetting, TableType: copy.TableType, DeleteMode: copy.DeleteMode,
		SharingParentObject: copy.SharingParentObject, SharingParentField: copy.SharingParentField,
		ChildCloneRules: copy.ChildCloneRules, ExclusiveSiblingFields: copy.ExclusiveSiblingFields,
		DefaultChildRows: copy.DefaultChildRows,
	}, nil
}

func projectMutationDefinitions(
	source *mutationDefinitionDocument,
	included map[string]struct{},
	profile ProjectionProfile,
	capabilityID string,
	rootFlowNames []string,
	outboundIntentRoutes []OutboundIntentRoute,
	formulaEngine *formula.Engine,
) (*mutationDefinitionDocument, error) {
	if source == nil {
		return nil, conflict(ConflictInvalidProjection, "mutation", "record-mutation projection requires mutation definitions")
	}
	if !profileUsesMutationClosure(profile) {
		return nil, conflict(ConflictInvalidProjection, "mutation", "profile %q has no mutation closure", profile)
	}
	out := emptyMutationDefinitionDocument()
	for _, value := range source.ValidationRules {
		if _, ok := included[foldIdentity(value.ObjectAPIName)]; ok {
			out.ValidationRules = append(out.ValidationRules, value)
		}
	}
	for _, value := range source.RecordTypes {
		if _, ok := included[foldIdentity(value.ObjectAPIName)]; ok {
			out.RecordTypes = append(out.RecordTypes, value)
		}
	}
	for _, value := range source.AutoNumberRules {
		if _, ok := included[foldIdentity(value.ObjectAPIName)]; ok {
			out.AutoNumberRules = append(out.AutoNumberRules, value)
		}
	}
	for _, value := range source.FieldMappingRules {
		if _, ok := included[foldIdentity(value.TargetObject)]; !ok {
			continue
		}
		if _, ok := included[foldIdentity(value.SourceObject)]; !ok {
			return nil, conflict(ConflictInvalidProjection, "mutation.fieldMappingRules", "mapping source object %q is outside the projection", value.SourceObject)
		}
		out.FieldMappingRules = append(out.FieldMappingRules, value)
	}
	if profile == ProjectionRecordMutationV1 {
		// Connector inbound facts and OAuth disclosure are owned by the generic
		// mutation/connector execution profile. Approval execution needs the
		// neutral mutation engine, but must not inherit unrelated connector
		// authority merely because both profiles write records.
		for _, value := range source.ConnectorInboundFieldMappingRules {
			if _, ok := included[foldIdentity(value.TargetObject)]; ok {
				out.ConnectorInboundFieldMappingRules = append(out.ConnectorInboundFieldMappingRules, value)
			}
		}
		for _, value := range source.ConnectorInboundAuthorship {
			if _, ok := included[foldIdentity(value.SourceObject)]; ok {
				out.ConnectorInboundAuthorship = append(out.ConnectorInboundAuthorship, value)
			}
		}
		for _, value := range source.ConnectorInboundFinalization {
			if _, ok := included[foldIdentity(value.Contract.SourceObject)]; ok {
				out.ConnectorInboundFinalization = append(out.ConnectorInboundFinalization, value)
			}
		}
		for _, value := range source.OAuthPublicScopes {
			if _, objectIncluded := included[foldIdentity(value.ObjectAPIName)]; objectIncluded {
				out.OAuthPublicScopes = append(out.OAuthPublicScopes, value)
			}
		}
	}
	for _, value := range source.CreateSourceGuards {
		if _, ok := included[foldIdentity(value.TargetObject)]; ok {
			out.CreateSourceGuards = append(out.CreateSourceGuards, value)
		}
	}
	if capabilityID == flowsemantic.CapabilityName {
		// ExecuteFlow is admitted separately from the record-event worker. Its
		// AppInstallation authorization is proven against AppMetadataView at
		// the serving boundary, while this view supplies exact active Runtime
		// versions from the same immutable Catalog revision.
		out.FlowDefinitions = append(out.FlowDefinitions, source.FlowDefinitions...)
		out.FlowVersions = append(out.FlowVersions, source.FlowVersions...)
		sortMutationDefinitions(out)
		if err := validateRuntimeFlowExecutionDefinitions(out, included, formulaEngine); err != nil {
			return nil, conflict(ConflictInvalidProjection, "mutation.flowExecution", "%v", err)
		}
		return out, nil
	}
	// Subscriptions select root flows. Native subflows and actionType=flow calls
	// are synchronous mutation dependencies, so walk their transitive active
	// definition/version closure inside this same Catalog revision.
	definitionsByID := make(map[string]runtimeFlowDefinitionRecord, len(source.FlowDefinitions))
	definitionsByName := make(map[string]runtimeFlowDefinitionRecord, len(source.FlowDefinitions))
	versionsByID := make(map[string]runtimeFlowVersionRecord, len(source.FlowVersions))
	for _, value := range source.FlowDefinitions {
		definitionsByID[foldIdentity(value.ID)] = value
		definitionsByName[foldIdentity(value.DeveloperName)] = value
	}
	for _, value := range source.FlowVersions {
		versionsByID[foldIdentity(value.ID)] = value
	}
	neededDefinitions := map[string]struct{}{}
	neededVersions := map[string]struct{}{}
	type flowQueueEntry struct {
		versionKey            string
		allowAsyncAfterCommit bool
	}
	queue := make([]flowQueueEntry, 0)
	visitedEntry := map[string]struct{}{}
	enqueue := func(definitionID, versionID string, allowAsyncAfterCommit bool) error {
		definition, defOK := definitionsByID[foldIdentity(definitionID)]
		version, versionOK := versionsByID[foldIdentity(versionID)]
		if !defOK || !versionOK || definition.ActiveVersionID != version.ID || version.DefinitionID != definition.ID {
			return conflict(ConflictInvalidProjection, "mutation.flows", "flow definition %q/version %q is not one active revision-bound pair", definitionID, versionID)
		}
		defKey, versionKey := foldIdentity(definition.ID), foldIdentity(version.ID)
		neededDefinitions[defKey] = struct{}{}
		neededVersions[versionKey] = struct{}{}
		entryKey := versionKey + "\x00" + strconv.FormatBool(allowAsyncAfterCommit)
		if _, exists := visitedEntry[entryKey]; !exists {
			visitedEntry[entryKey] = struct{}{}
			queue = append(queue, flowQueueEntry{
				versionKey:            versionKey,
				allowAsyncAfterCommit: allowAsyncAfterCommit,
			})
		}
		return nil
	}
	if profile != ProjectionApprovalExecutionV1 {
		for _, value := range source.TriggerSubscriptions {
			// The compiler has already verified the subscription against a flow;
			// object scoping is rechecked from the referenced definitions below.
			definition := findFlowDefinition(source.FlowDefinitions, value.FlowDefinitionID)
			if definition == nil || definition.TriggerObjectOrEvent == "" {
				continue
			}
			if _, ok := included[foldIdentity(definition.TriggerObjectOrEvent)]; !ok {
				continue
			}
			out.TriggerSubscriptions = append(out.TriggerSubscriptions, value)
			allowAsyncAfterCommit := isAfterSaveTrigger(value.EventType) && isAfterSaveTrigger(definition.TriggerType)
			if err := enqueue(value.FlowDefinitionID, value.FlowVersionID, allowAsyncAfterCommit); err != nil {
				return nil, err
			}
		}
	}
	for _, rawName := range rootFlowNames {
		name := strings.TrimSpace(rawName)
		definition, ok := definitionsByName[foldIdentity(name)]
		if name == "" || !ok {
			return nil, conflict(
				ConflictInvalidProjection,
				"mutation.approvalFlows",
				"approval Flow dependency %q is absent from this revision",
				rawName,
			)
		}
		if err := enqueue(definition.ID, definition.ActiveVersionID, false); err != nil {
			return nil, err
		}
	}
	for len(queue) > 0 {
		entry := queue[0]
		queue = queue[1:]
		version := versionsByID[entry.versionKey]
		definition := definitionsByID[foldIdentity(version.DefinitionID)]
		var children []string
		var err error
		if profile == ProjectionApprovalExecutionV1 {
			children, err = runtimeApprovalNotificationFlowChildNames(version)
		} else {
			children, err = runtimeMutationFlowChildNames(
				version,
				entry.allowAsyncAfterCommit,
				runtimeOutboundIntentActionAdmission(definition, outboundIntentRoutes, out),
			)
		}
		if err != nil {
			return nil, conflict(ConflictInvalidProjection, "mutation.flowVersions."+version.ID, "%v", err)
		}
		objectNames, err := flowReferencedObjectNames(version)
		if err != nil {
			return nil, conflict(ConflictInvalidProjection, "mutation.flowVersions."+version.ID, "%v", err)
		}
		for _, objectName := range objectNames {
			if _, ok := included[foldIdentity(objectName)]; !ok {
				return nil, conflict(ConflictInvalidProjection, "mutation.flowVersions."+version.ID, "flow object dependency %q is outside the projection", objectName)
			}
		}
		for _, childName := range children {
			definition, ok := definitionsByName[foldIdentity(childName)]
			if !ok {
				return nil, conflict(ConflictInvalidProjection, "mutation.flowVersions."+version.ID, "active child flow %q is absent from the revision", childName)
			}
			if err := enqueue(definition.ID, definition.ActiveVersionID, false); err != nil {
				return nil, err
			}
		}
	}
	for _, value := range source.FlowDefinitions {
		if _, ok := neededDefinitions[foldIdentity(value.ID)]; ok {
			out.FlowDefinitions = append(out.FlowDefinitions, value)
		}
	}
	for _, value := range source.FlowVersions {
		if _, ok := neededVersions[foldIdentity(value.ID)]; ok {
			out.FlowVersions = append(out.FlowVersions, value)
		}
	}
	sortMutationDefinitions(out)
	usesCommand, err := mutationUsesFlowActionType(out, unsupportedFlowAcceptCommandActionType)
	if err != nil {
		return nil, conflict(ConflictInvalidProjection, "mutation", "%v", err)
	}
	if usesCommand {
		return nil, conflict(
			ConflictInvalidProjection,
			"mutation.flowVersions",
			"Flow acceptCommand action is outside the Runtime mutation worker closure",
		)
	}
	var validationErr error
	if profile == ProjectionApprovalExecutionV1 {
		validationErr = validateRuntimeApprovalMutationDefinitions(out, nil, formulaEngine)
	} else {
		validationErr = validateRuntimeMutationDefinitions(out, nil, outboundIntentRoutes, formulaEngine)
	}
	if validationErr != nil {
		return nil, conflict(ConflictInvalidProjection, "mutation", "%v", validationErr)
	}
	return out, nil
}

func isFlowExecutionCapability(capabilityID string) bool {
	return capabilityID == flowsemantic.CapabilityName
}

func validateRuntimeFlowExecutionDefinitions(
	doc *mutationDefinitionDocument,
	included map[string]struct{},
	formulaEngine *formula.Engine,
) error {
	if err := validateMutationDefinitions(doc, nil, formulaEngine); err != nil {
		return err
	}
	if doc == nil {
		return errors.New("Flow execution projection has no mutation closure")
	}
	definitionsByName := make(map[string]runtimeFlowDefinitionRecord, len(doc.FlowDefinitions))
	versionsByID := make(map[string]runtimeFlowVersionRecord, len(doc.FlowVersions))
	for _, definition := range doc.FlowDefinitions {
		definitionsByName[foldIdentity(definition.DeveloperName)] = definition
	}
	for _, version := range doc.FlowVersions {
		versionsByID[foldIdentity(version.ID)] = version
	}
	for _, definition := range doc.FlowDefinitions {
		version, ok := versionsByID[foldIdentity(definition.ActiveVersionID)]
		if !ok || version.DefinitionID != definition.ID {
			return fmt.Errorf(
				"Flow definition %q has no exact active version in the projection",
				definition.DeveloperName,
			)
		}
		children, err := flowChildNames(version)
		if err != nil {
			return fmt.Errorf("Flow %q has invalid child closure: %w", definition.DeveloperName, err)
		}
		for _, childName := range children {
			child, ok := definitionsByName[foldIdentity(childName)]
			if !ok {
				return fmt.Errorf(
					"Flow %q references absent active child %q",
					definition.DeveloperName,
					childName,
				)
			}
			if _, ok := versionsByID[foldIdentity(child.ActiveVersionID)]; !ok {
				return fmt.Errorf(
					"Flow %q child %q has no exact active version",
					definition.DeveloperName,
					childName,
				)
			}
		}
		objectNames, err := flowReferencedObjectNames(version)
		if err != nil {
			return fmt.Errorf("Flow %q object closure: %w", definition.DeveloperName, err)
		}
		for _, objectName := range objectNames {
			if _, ok := included[foldIdentity(objectName)]; !ok {
				return fmt.Errorf(
					"Flow %q object dependency %q is outside the projection",
					definition.DeveloperName,
					objectName,
				)
			}
		}
	}
	return nil
}

func flowReferencedObjectNames(version runtimeFlowVersionRecord) ([]string, error) {
	var metadata models.FlowMetadata
	if err := json.Unmarshal(version.Metadata, &metadata); err != nil {
		return nil, fmt.Errorf("decode Flow metadata: %w", err)
	}
	seen := map[string]string{}
	add := func(value string) {
		value = strings.TrimSpace(value)
		if value != "" {
			seen[foldIdentity(value)] = value
		}
	}
	for _, value := range metadata.RecordCreates {
		add(value.Object)
		if value.FieldMapping != nil {
			add(value.FieldMapping.SourceObject)
			add(value.FieldMapping.TargetObject)
		}
	}
	for _, value := range metadata.RecordUpdates {
		add(value.Object)
		if value.FieldMapping != nil {
			add(value.FieldMapping.SourceObject)
			add(value.FieldMapping.TargetObject)
		}
	}
	for _, value := range metadata.RecordDeletes {
		add(value.Object)
	}
	for _, value := range metadata.RecordLookups {
		add(value.Object)
	}
	for _, value := range metadata.AtomicParentPrelocks {
		add(value.ParentObject)
		add(value.SourceObject)
	}
	for _, value := range metadata.AtomicHistoryPrelocks {
		add(value.Object)
	}
	out := make([]string, 0, len(seen))
	for _, name := range seen {
		out = append(out, name)
	}
	sort.Slice(out, func(i, j int) bool { return foldIdentity(out[i]) < foldIdentity(out[j]) })
	return out, nil
}

func flowChildNames(version runtimeFlowVersionRecord) ([]string, error) {
	var metadata models.FlowMetadata
	if err := json.Unmarshal(version.Metadata, &metadata); err != nil {
		return nil, fmt.Errorf("decode Flow metadata: %w", err)
	}
	seen := map[string]string{}
	for _, subflow := range metadata.Subflows {
		name := strings.TrimSpace(subflow.FlowName)
		if name == "" {
			return nil, fmt.Errorf("subflow %q has no child flow name", subflow.Name)
		}
		seen[foldIdentity(name)] = name
	}
	for _, action := range metadata.ActionCalls {
		if !strings.EqualFold(strings.TrimSpace(action.ActionType), "flow") {
			continue
		}
		name := strings.TrimSpace(action.ActionName)
		if name == "" {
			return nil, fmt.Errorf("flow action %q has no child flow name", action.Name)
		}
		seen[foldIdentity(name)] = name
	}
	out := make([]string, 0, len(seen))
	for _, name := range seen {
		out = append(out, name)
	}
	sort.Slice(out, func(i, j int) bool { return foldIdentity(out[i]) < foldIdentity(out[j]) })
	return out, nil
}

// runtimeMutationFlowChildNames is the activation boundary between the
// Runtime-owned record-event worker and D2C Integration capabilities. The D2B
// worker composes the generic Flow/subflow interpreter, but deliberately owns
// no App action handler, connector, webhook, email provider, or metadata-backed
// ActionService. Silently carrying such an action into the immutable view
// would make execution depend on an absent owner (and historically could reach
// nil legacy repositories). Reject it while Catalog still owns activation.
func runtimeMutationFlowChildNames(
	version runtimeFlowVersionRecord,
	allowAsyncAfterCommit bool,
	allowOutboundIntentAction func(*models.FlowActionCall) bool,
) ([]string, error) {
	var metadata models.FlowMetadata
	if err := json.Unmarshal(version.Metadata, &metadata); err != nil {
		return nil, fmt.Errorf("decode Flow metadata: %w", err)
	}
	if len(metadata.Waits) > 0 {
		return nil, fmt.Errorf("wait element %q is outside the D2B Runtime mutation worker closure; scheduled/resumable execution needs its D2C owner before activation", metadata.Waits[0].Name)
	}
	if len(metadata.Screens) > 0 {
		return nil, fmt.Errorf("screen element %q is outside the D2B Runtime mutation worker closure", metadata.Screens[0].Name)
	}
	hasScheduledRecordPath := false
	asyncAfterCommit := false
	if metadata.Start != nil {
		if len(metadata.Start.ScheduledPaths) > 0 {
			hasScheduledRecordPath = true
			// The record-event worker owns the narrow canonical
			// AsyncAfterCommit shape. B1 additionally allows one positive,
			// RecordTriggerEvent-relative timed path: the event transaction
			// persists its exact entry and immutable Flow binding for the
			// existing scheduled worker. Multiple branches and other time
			// sources remain outside the launch closure.
			path := metadata.Start.ScheduledPaths[0]
			asyncAfterCommit = strings.EqualFold(strings.TrimSpace(path.PathType), "AsyncAfterCommit") &&
				strings.TrimSpace(path.TimeSource) == "" &&
				path.OffsetNumber == 0 &&
				strings.TrimSpace(path.OffsetUnit) == "" &&
				metadata.Start.Connector == nil &&
				strings.TrimSpace(metadata.StartElementReference) == ""
			timed := strings.TrimSpace(path.PathType) == "" &&
				strings.EqualFold(strings.TrimSpace(path.TimeSource), "RecordTriggerEvent") &&
				isLaunchTimedOffset(path.OffsetNumber, path.OffsetUnit) &&
				strings.TrimSpace(metadata.StartElementReference) == ""
			if !allowAsyncAfterCommit || !isAfterSaveTrigger(metadata.Start.TriggerType) ||
				len(metadata.Start.ScheduledPaths) != 1 ||
				path.Connector == nil || strings.TrimSpace(path.Connector.TargetReference) == "" ||
				(timed && strings.TrimSpace(path.Name) == "") ||
				(!asyncAfterCommit && !timed) {
				return nil, fmt.Errorf("scheduled path %q is outside the D2B Runtime mutation worker closure; scheduled/resumable execution needs its D2C owner before activation", path.Name)
			}
		}
		if metadata.Start.Schedule != nil {
			return nil, errors.New("scheduled Flow start is outside the D2B Runtime mutation worker closure; scheduled/resumable execution needs its D2C owner before activation")
		}
		if metadata.Start.ConnectorSource != nil {
			return nil, errors.New("connector-sourced Flow start is outside the D2B Runtime mutation worker closure; Integration ownership is required before activation")
		}
	}
	for _, create := range metadata.RecordCreates {
		if create.ConnectorFieldMapping != nil {
			return nil, fmt.Errorf("recordCreate %q connector field mapping is outside the D2B Runtime mutation worker closure; Integration ownership is required before activation", create.Name)
		}
	}
	for index := range metadata.ActionCalls {
		action := &metadata.ActionCalls[index]
		actionType := strings.TrimSpace(action.ActionType)
		// Outbound-intent actions require the intent worker. The scheduled Flow
		// worker intentionally does not wire that owner, so reject them at
		// activation before a strict AsyncAfterCommit path can enqueue durable
		// work that will later fail plan preparation.
		if runtimeOutboundIntentActionOperation(actionType) != "" && !hasScheduledRecordPath &&
			allowAsyncAfterCommit && allowOutboundIntentAction != nil && allowOutboundIntentAction(action) {
			continue
		}
		identityAction := strings.EqualFold(actionType, flowactions.ActionTypeResolveConnectorRemoteIdentity) ||
			strings.EqualFold(actionType, flowactions.ActionTypeTransferConnectorRemoteIdentity)
		if identityAction && asyncAfterCommit && metadata.RunAtomically &&
			!version.DefinitionIsCustomized && strings.TrimSpace(version.DefinitionOwnerPack) != "" &&
			strings.EqualFold(strings.TrimSpace(version.Source), "standard") {
			continue
		}
		if !strings.EqualFold(actionType, "flow") {
			if actionType == "" {
				actionType = "<blank>"
			}
			return nil, fmt.Errorf("actionCall type %q is outside the D2B Runtime mutation worker closure; classify and implement its D2C owner before activation", actionType)
		}
	}
	return flowChildNames(version)
}

// runtimeApprovalNotificationFlowChildNames applies the same narrow execution
// profile owned by ApprovalService: an approval lifecycle may invoke a
// notification-only Flow after its transaction commits, but that Flow cannot
// acquire record, child-Flow, interactive, scheduled, or connector authority.
// The three accepted action types are generic L0 handlers registered by the
// Runtime automation engine; app-owned aliases remain outside Catalog.
func runtimeApprovalNotificationFlowChildNames(version runtimeFlowVersionRecord) ([]string, error) {
	var metadata models.FlowMetadata
	if err := json.Unmarshal(version.Metadata, &metadata); err != nil {
		return nil, fmt.Errorf("decode Flow metadata: %w", err)
	}
	if metadata.Start != nil && (strings.TrimSpace(metadata.Start.Object) != "" ||
		strings.TrimSpace(metadata.Start.TriggerType) != "" ||
		strings.TrimSpace(metadata.Start.RecordTriggerType) != "" ||
		len(metadata.Start.ScheduledPaths) != 0 || metadata.Start.Schedule != nil) {
		return nil, errors.New("approval notification Flow must be invoked only from the approval lifecycle")
	}
	if len(metadata.RecordCreates) != 0 || len(metadata.RecordUpdates) != 0 ||
		len(metadata.RecordDeletes) != 0 || len(metadata.RecordLookups) != 0 {
		return nil, errors.New("record mutation or lookup is outside approval notification scope")
	}
	if len(metadata.Subflows) != 0 {
		return nil, errors.New("subflow is outside approval notification scope")
	}
	if len(metadata.Screens) != 0 || len(metadata.Waits) != 0 || len(metadata.CustomErrors) != 0 {
		return nil, errors.New("interactive, paused, or gating elements are outside approval notification scope")
	}
	for _, action := range metadata.ActionCalls {
		actionType := strings.TrimSpace(action.ActionType)
		if strings.EqualFold(actionType, flowactions.ActionTypeResolveEmailRecipients) ||
			strings.EqualFold(actionType, flowactions.ActionTypeSendEmail) ||
			strings.EqualFold(actionType, flowactions.ActionTypeSendNotification) {
			continue
		}
		if actionType == "" {
			actionType = "<blank>"
		}
		return nil, fmt.Errorf("actionCall type %q is outside approval notification scope", actionType)
	}
	return nil, nil
}

func runtimeOutboundIntentActionAdmission(
	definition runtimeFlowDefinitionRecord,
	routes []OutboundIntentRoute,
	mutation *mutationDefinitionDocument,
) func(*models.FlowActionCall) bool {
	return func(action *models.FlowActionCall) bool {
		if action == nil || definition.IsCustomized || strings.TrimSpace(definition.OwnerPack) == "" {
			return false
		}
		operation := runtimeOutboundIntentActionOperation(action.ActionType)
		if operation == "" {
			return false
		}
		matches := 0
		for _, route := range routes {
			routeOperation := normalizedOutboundIntentOperation(route.Operation)
			hasOperationAuthority := routeOperation == outboundintentv1.OperationSendMail && route.MailCompletion != nil
			if routeOperation == outboundintentv1.OperationUpsert {
				hasOperationAuthority = outboundIntentRouteHasFinalizationAuthority(route, mutation)
			}
			if routeOperation == operation &&
				route.FlowDeveloperName == definition.DeveloperName &&
				route.FlowActionName == strings.TrimSpace(action.Name) &&
				route.SourceObject == definition.TriggerObjectOrEvent &&
				route.ConnectorOwnerKey == definition.OwnerPack &&
				hasOperationAuthority {
				matches++
			}
		}
		return matches == 1
	}
}

func runtimeOutboundIntentActionOperation(actionType string) outboundintentv1.Operation {
	switch strings.TrimSpace(actionType) {
	case constants.ActionTypePushToConnector:
		return outboundintentv1.OperationUpsert
	case flowactions.ActionTypeEnqueueOutboundMail, flowactions.ActionTypeEnqueueOutboundMailV3:
		return outboundintentv1.OperationSendMail
	default:
		return ""
	}
}

func mutationUsesFlowActionType(doc *mutationDefinitionDocument, actionType string) (bool, error) {
	if doc == nil {
		return false, nil
	}
	for _, version := range doc.FlowVersions {
		var metadata models.FlowMetadata
		if err := json.Unmarshal(version.Metadata, &metadata); err != nil {
			return false, fmt.Errorf("decode Flow metadata: %w", err)
		}
		for _, action := range metadata.ActionCalls {
			if strings.EqualFold(strings.TrimSpace(action.ActionType), actionType) {
				return true, nil
			}
		}
	}
	return false, nil
}

// validateRuntimeMutationDefinitions applies the ordinary revision/closure
// checks and the stricter worker-execution boundary required by a serving
// Runtime mutation view. A Runtime view must fail closed even when a hostile
// caller supplies canonically encoded, correctly digested bytes directly to
// OpenRuntimeMetadataView instead of going through the compiler projection.
func validateRuntimeMutationDefinitions(
	doc *mutationDefinitionDocument,
	objects map[string]*models.ObjectMetadata,
	outboundIntentRoutes []OutboundIntentRoute,
	formulaEngine *formula.Engine,
) error {
	if err := validateMutationDefinitions(doc, objects, formulaEngine); err != nil {
		return err
	}
	if doc == nil {
		return nil
	}
	definitionsByID := make(map[string]runtimeFlowDefinitionRecord, len(doc.FlowDefinitions))
	definitionsByName := make(map[string]runtimeFlowDefinitionRecord, len(doc.FlowDefinitions))
	for _, definition := range doc.FlowDefinitions {
		definitionsByID[foldIdentity(definition.ID)] = definition
		definitionsByName[foldIdentity(definition.DeveloperName)] = definition
	}
	rootVersions := map[string]struct{}{}
	for _, subscription := range doc.TriggerSubscriptions {
		definition, ok := definitionsByID[foldIdentity(subscription.FlowDefinitionID)]
		if ok && isAfterSaveTrigger(subscription.EventType) && isAfterSaveTrigger(definition.TriggerType) {
			rootVersions[foldIdentity(subscription.FlowVersionID)] = struct{}{}
		}
	}
	childVersions := map[string]struct{}{}
	for _, version := range doc.FlowVersions {
		children, err := flowChildNames(version)
		if err != nil {
			return fmt.Errorf("mutation flow version %q has invalid child closure: %w", version.ID, err)
		}
		for _, childName := range children {
			if definition, ok := definitionsByName[foldIdentity(childName)]; ok {
				childVersions[foldIdentity(definition.ActiveVersionID)] = struct{}{}
			}
		}
	}
	for _, version := range doc.FlowVersions {
		versionKey := foldIdentity(version.ID)
		_, isRoot := rootVersions[versionKey]
		_, isChild := childVersions[versionKey]
		definition := definitionsByID[foldIdentity(version.DefinitionID)]
		if _, err := runtimeMutationFlowChildNames(
			version,
			isRoot && !isChild,
			runtimeOutboundIntentActionAdmission(definition, outboundIntentRoutes, doc),
		); err != nil {
			return fmt.Errorf("mutation flow version %q is outside the Runtime mutation worker closure: %w", version.ID, err)
		}
		if err := validateRuntimeFlowFieldMappingDependencies(version, doc.FieldMappingRules); err != nil {
			return fmt.Errorf("mutation flow version %q has incomplete field-mapping closure: %w", version.ID, err)
		}
	}
	return nil
}

func validateRuntimeApprovalMutationDefinitions(
	doc *mutationDefinitionDocument,
	objects map[string]*models.ObjectMetadata,
	formulaEngine *formula.Engine,
) error {
	if err := validateMutationDefinitions(doc, objects, formulaEngine); err != nil {
		return err
	}
	if doc == nil {
		return nil
	}
	for _, version := range doc.FlowVersions {
		if _, err := runtimeApprovalNotificationFlowChildNames(version); err != nil {
			return fmt.Errorf(
				"approval notification flow version %q is outside the Runtime approval closure: %w",
				version.ID,
				err,
			)
		}
		if err := validateRuntimeFlowFieldMappingDependencies(version, doc.FieldMappingRules); err != nil {
			return fmt.Errorf(
				"approval notification flow version %q has incomplete field-mapping closure: %w",
				version.ID,
				err,
			)
		}
	}
	return nil
}

func isLaunchTimedOffset(number int, value string) bool {
	if number <= 0 {
		return false
	}
	max := 0
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "minute", "minutes":
		max = 366 * 24 * 60
	case "hour", "hours":
		max = 366 * 24
	case "day", "days":
		max = 366
	default:
		return false
	}
	return number <= max
}

func isAfterSaveTrigger(value string) bool {
	canonical, ok := triggerprojection.CanonicalRecordTriggerType(value)
	return ok && canonical == "RecordAfterSave"
}

func validateRuntimeFlowFieldMappingDependencies(version runtimeFlowVersionRecord, rules []FieldMappingRule) error {
	var metadata models.FlowMetadata
	if err := json.Unmarshal(version.Metadata, &metadata); err != nil {
		return fmt.Errorf("decode Flow metadata: %w", err)
	}
	available := make(map[string]struct{}, len(rules))
	for _, rule := range rules {
		key := foldIdentity(rule.SourceObject) + "\x00" + foldIdentity(rule.TargetObject) + "\x00" + foldIdentity(rule.ActionType)
		available[key] = struct{}{}
	}
	validate := func(kind, name, fallbackTarget string, ref *models.FlowFieldMappingRef) error {
		if ref == nil {
			return nil
		}
		if strings.TrimSpace(ref.SourceReference) == "" {
			return fmt.Errorf("%s %q fieldMapping has blank sourceReference", kind, name)
		}
		target := ref.TargetObject
		if target == "" {
			target = fallbackTarget
		}
		for _, item := range []struct{ field, identity string }{
			{field: "sourceObject", identity: ref.SourceObject},
			{field: "targetObject", identity: target},
			{field: "actionType", identity: ref.ActionType},
		} {
			if err := requireCanonicalIdentity("fieldMapping."+item.field, item.identity); err != nil {
				return fmt.Errorf("%s %q: %w", kind, name, err)
			}
		}
		key := foldIdentity(ref.SourceObject) + "\x00" + foldIdentity(target) + "\x00" + foldIdentity(ref.ActionType)
		if _, ok := available[key]; !ok {
			return fmt.Errorf("%s %q references absent mapping group (%s, %s, %s)", kind, name, ref.SourceObject, target, ref.ActionType)
		}
		return nil
	}
	for i := range metadata.RecordCreates {
		value := &metadata.RecordCreates[i]
		if err := validate("recordCreate", value.Name, value.Object, value.FieldMapping); err != nil {
			return err
		}
	}
	for i := range metadata.RecordUpdates {
		value := &metadata.RecordUpdates[i]
		if err := validate("recordUpdate", value.Name, value.Object, value.FieldMapping); err != nil {
			return err
		}
	}
	return nil
}

func findFlowDefinition(values []runtimeFlowDefinitionRecord, id string) *runtimeFlowDefinitionRecord {
	for i := range values {
		if values[i].ID == id {
			return &values[i]
		}
	}
	return nil
}

func validateMutationDefinitions(
	doc *mutationDefinitionDocument,
	objects map[string]*models.ObjectMetadata,
	formulaEngine *formula.Engine,
) error {
	if doc == nil {
		return nil
	}
	seen := map[string]struct{}{}
	definitions := map[string]runtimeFlowDefinitionRecord{}
	definitionNames := map[string]string{}
	versions := map[string]runtimeFlowVersionRecord{}
	for i, value := range doc.FlowDefinitions {
		if err := validateUniqueMutationIdentity(seen, "flowDefinition", value.ID, i); err != nil {
			return err
		}
		if value.ID == "" || value.DeveloperName == "" || value.ActiveVersionID == "" || value.IsDeleted {
			return fmt.Errorf("mutation flow definition %q is not an active canonical definition", value.ID)
		}
		nameKey := foldIdentity(value.DeveloperName)
		if prior, exists := definitionNames[nameKey]; exists {
			return fmt.Errorf("mutation flow definitions %q and %q share developer name %q", prior, value.ID, value.DeveloperName)
		}
		definitionNames[nameKey] = value.ID
		definitions[foldIdentity(value.ID)] = value
	}
	for i, value := range doc.FlowVersions {
		if err := validateUniqueMutationIdentity(seen, "flowVersion", value.ID, i); err != nil {
			return err
		}
		if value.ID == "" || value.DefinitionID == "" || value.Status != models.FlowStatusActive || value.IsDeleted || len(value.Metadata) == 0 {
			return fmt.Errorf("mutation flow version %q is not active and complete", value.ID)
		}
		if _, err := canonicalFlowMetadata(flowVersionToModel(value)); err != nil {
			return fmt.Errorf("mutation flow version %q metadata is invalid: %w", value.ID, err)
		}
		versions[foldIdentity(value.ID)] = value
	}
	for _, definition := range definitions {
		version, ok := versions[foldIdentity(definition.ActiveVersionID)]
		if !ok || version.DefinitionID != definition.ID {
			return fmt.Errorf("mutation flow definition %q has no matching active version", definition.ID)
		}
	}
	for _, version := range versions {
		definition, ok := definitions[foldIdentity(version.DefinitionID)]
		if !ok || definition.ActiveVersionID != version.ID {
			return fmt.Errorf("mutation flow version %q has no matching active definition", version.ID)
		}
		children, err := flowChildNames(version)
		if err != nil {
			return fmt.Errorf("mutation flow version %q child closure is invalid: %w", version.ID, err)
		}
		for _, childName := range children {
			childDefinitionID, ok := definitionNames[foldIdentity(childName)]
			if !ok {
				return fmt.Errorf("mutation flow version %q references absent active child flow %q", version.ID, childName)
			}
			childDefinition := definitions[foldIdentity(childDefinitionID)]
			if _, ok := versions[foldIdentity(childDefinition.ActiveVersionID)]; !ok {
				return fmt.Errorf("mutation flow version %q child flow %q has no active version", version.ID, childName)
			}
		}
	}
	for i, value := range doc.TriggerSubscriptions {
		if err := validateUniqueMutationIdentity(seen, "triggerSubscription", value.ID, i); err != nil {
			return err
		}
		definition, defOK := definitions[foldIdentity(value.FlowDefinitionID)]
		version, versionOK := versions[foldIdentity(value.FlowVersionID)]
		if !value.IsActive || !defOK || !versionOK || definition.ActiveVersionID != value.FlowVersionID || version.DefinitionID != value.FlowDefinitionID {
			return fmt.Errorf("mutation trigger subscription %q has a mixed or incomplete flow closure", value.ID)
		}
		if objects != nil {
			object := objects[foldIdentity(definition.TriggerObjectOrEvent)]
			if object == nil || object.KeyPrefix == nil || !strings.EqualFold(strings.TrimSpace(*object.KeyPrefix), value.KeyPrefix) {
				return fmt.Errorf("mutation trigger subscription %q is not bound to its trigger object's key prefix", value.ID)
			}
		}
	}
	for i, value := range doc.ValidationRules {
		if err := validateUniqueMutationIdentity(seen, "validationRule", value.ID, i); err != nil {
			return err
		}
	}
	for i, value := range doc.RecordTypes {
		if err := validateUniqueMutationIdentity(seen, "recordType", value.ID, i); err != nil {
			return err
		}
	}
	for i, value := range doc.AutoNumberRules {
		if err := validateUniqueMutationIdentity(seen, "autoNumberRule", value.FieldID, i); err != nil {
			return err
		}
	}
	for i, value := range doc.FieldMappingRules {
		key := fieldMappingRuleKey(value)
		if err := validateFieldMappingRule(value, objects); err != nil {
			return fmt.Errorf("mutation field mapping rule %d is invalid: %w", i, err)
		}
		seenKey := "fieldMappingRule\x00" + key
		if _, exists := seen[seenKey]; exists {
			return fmt.Errorf("mutation field mapping rule %d is duplicated", i)
		}
		seen[seenKey] = struct{}{}
		if objects != nil && (objects[foldIdentity(value.SourceObject)] == nil ||
			objects[foldIdentity(value.TargetObject)] == nil) {
			return fmt.Errorf("mutation field mapping rule %d references an absent object", i)
		}
	}
	for i, value := range doc.ConnectorInboundFieldMappingRules {
		if err := validateConnectorInboundFieldMappingRule(value, objects, formulaEngine); err != nil {
			return fmt.Errorf("mutation connector inbound field mapping rule %d is invalid: %w", i, err)
		}
		key := "connectorInboundFieldMappingRule\x00" + connectorInboundFieldMappingRuleKey(value)
		if _, exists := seen[key]; exists {
			return fmt.Errorf("mutation connector inbound field mapping rule %d is duplicated", i)
		}
		seen[key] = struct{}{}
	}
	authorshipByKey := make(map[string]ConnectorInboundAuthorshipBinding, len(doc.ConnectorInboundAuthorship))
	for i, value := range doc.ConnectorInboundAuthorship {
		if err := validateConnectorInboundAuthorshipBinding(value, objects); err != nil {
			return fmt.Errorf("mutation connector inbound authorship binding %d is invalid: %w", i, err)
		}
		key := "connectorInboundAuthorship\x00" + connectorInboundAuthorshipBindingKey(value)
		if _, exists := seen[key]; exists {
			return fmt.Errorf("mutation connector inbound authorship binding %d is duplicated", i)
		}
		seen[key] = struct{}{}
		authorshipByKey[connectorInboundAuthorshipBindingKey(value)] = value
	}
	contracts := make([]connector.ConnectorPushDurableMetadataContract, 0, len(doc.ConnectorInboundFinalization))
	mappings := make([]connector.ConnectorFieldMappingInput, 0)
	contractGenerationByOwner := make(map[string]string)
	profileAuthorityByKey := make(map[string][]byte)
	mappingClosureByKey := make(map[string][]byte)
	for i, value := range doc.ConnectorInboundFinalization {
		if err := validateConnectorInboundFinalizationAuthority(value, objects, formulaEngine); err != nil {
			return fmt.Errorf("mutation connector inbound finalization authority %d is invalid: %w", i, err)
		}
		key := "connectorInboundFinalization\x00" + connectorInboundFinalizationAuthorityKey(value)
		if _, exists := seen[key]; exists {
			return fmt.Errorf("mutation connector inbound finalization authority %d is duplicated", i)
		}
		seen[key] = struct{}{}
		expectedAuthorship := connectorInboundAuthorshipForFinalization(value)
		authorship, exists := authorshipByKey[connectorInboundAuthorshipBindingKey(expectedAuthorship)]
		if !exists || authorship != expectedAuthorship {
			return fmt.Errorf("mutation connector inbound finalization authority %d has no exact authorship binding", i)
		}
		ownerKey := foldIdentity(value.Contract.ConnectorOwnerKey)
		if generation, exists := contractGenerationByOwner[ownerKey]; exists && generation != value.ContractGeneration {
			return fmt.Errorf("mutation connector inbound finalization authority %d mixes contract generations for owner %q", i, value.Contract.ConnectorOwnerKey)
		}
		contractGenerationByOwner[ownerKey] = value.ContractGeneration
		profileKey := foldIdentity(value.Profile.OwnerPack) + "\x00" + foldIdentity(value.Profile.ProfileName)
		if err := registerCanonicalFinalizationFact(profileAuthorityByKey, profileKey, value.Profile); err != nil {
			return fmt.Errorf("mutation connector inbound finalization authority %d has conflicting profile authority: %w", i, err)
		}
		mappingKey := strings.Join([]string{
			foldIdentity(value.Contract.ConnectorOwnerKey),
			foldIdentity(value.Contract.SourceObject),
			foldIdentity(value.Contract.TargetObject),
			foldIdentity(value.Contract.ActionType),
		}, "\x00")
		if err := registerCanonicalFinalizationFact(mappingClosureByKey, mappingKey, value.OutboundMappings); err != nil {
			return fmt.Errorf("mutation connector inbound finalization authority %d has conflicting outbound mapping closure: %w", i, err)
		}
		contracts = append(contracts, value.Contract)
		mappings = append(mappings, value.OutboundMappings...)
	}
	if len(contracts) != 0 {
		validated, err := cloneJSON(contracts)
		if err != nil {
			return fmt.Errorf("clone connector inbound finalization contracts: %w", err)
		}
		before, err := marshalCanonical(validated)
		if err != nil {
			return fmt.Errorf("encode connector inbound finalization contracts: %w", err)
		}
		if err := kernelconnector.ValidateConnectorPushDurableMetadataContracts(validated, mappings); err != nil {
			return fmt.Errorf("mutation connector inbound finalization contract closure is invalid: %w", err)
		}
		after, err := marshalCanonical(validated)
		if err != nil {
			return fmt.Errorf("encode validated connector inbound finalization contracts: %w", err)
		}
		if !bytes.Equal(before, after) {
			return fmt.Errorf("mutation connector inbound finalization contract closure is not fully resolved")
		}
	}
	for i, value := range doc.CreateSourceGuards {
		key := createSourceGuardRuleKey(value)
		for name, identity := range map[string]string{
			"targetObject": value.TargetObject, "sourceFkField": value.SourceFKField,
			"owningStandardActionId": value.OwningStandardActionID, "actionName": value.ActionName,
		} {
			if err := requireCanonicalIdentity(fmt.Sprintf("mutation.createSourceGuards[%d].%s", i, name), identity); err != nil {
				return err
			}
		}
		seenKey := "createSourceGuard\x00" + key
		if _, exists := seen[seenKey]; exists {
			return fmt.Errorf("mutation create-from-source guard %d is duplicated", i)
		}
		seen[seenKey] = struct{}{}
		if objects != nil {
			object := objects[foldIdentity(value.TargetObject)]
			field := models.FindField(object, value.SourceFKField)
			if object == nil || !isCreateSourceGuardField(field) {
				return fmt.Errorf("mutation create-from-source guard %d references an absent or non-reference field", i)
			}
		}
	}
	for i, value := range doc.OAuthPublicScopes {
		if value.ObjectAPIName != strings.TrimSpace(value.ObjectAPIName) ||
			value.Name != strings.ToLower(strings.TrimSpace(value.Name)) || value.Name == "" ||
			value.OwnerPack != strings.TrimSpace(value.OwnerPack) {
			return fmt.Errorf("mutation OAuth public scope rule %d is not canonical", i)
		}
		if err := requireCanonicalIdentity(fmt.Sprintf("mutation.oauthPublicScopes[%d].objectApiName", i), value.ObjectAPIName); err != nil {
			return err
		}
		if err := requireCanonicalIdentity(fmt.Sprintf("mutation.oauthPublicScopes[%d].name", i), value.Name); err != nil {
			return err
		}
		if value.OwnerPack != "" {
			if err := requireCanonicalIdentity(fmt.Sprintf("mutation.oauthPublicScopes[%d].ownerPack", i), value.OwnerPack); err != nil {
				return err
			}
		}
		key := "oauthPublicScope\x00" + oauthPublicScopeRuleKey(value)
		if _, exists := seen[key]; exists {
			return fmt.Errorf("mutation OAuth public scope rule %d is duplicated", i)
		}
		seen[key] = struct{}{}
		if objects != nil && objects[foldIdentity(value.ObjectAPIName)] == nil {
			return fmt.Errorf("mutation OAuth public scope rule %d references absent object %q", i, value.ObjectAPIName)
		}
	}
	if objects != nil {
		for _, value := range doc.ValidationRules {
			if objects[foldIdentity(value.ObjectAPIName)] == nil {
				return fmt.Errorf("validation rule %q references absent object", value.ID)
			}
		}
		for _, value := range doc.RecordTypes {
			if objects[foldIdentity(value.ObjectAPIName)] == nil {
				return fmt.Errorf("record type %q references absent object", value.ID)
			}
		}
		for _, value := range doc.AutoNumberRules {
			if err := validateAutoNumberRuleBinding(value, objects); err != nil {
				return fmt.Errorf("autonumber rule %q has invalid field binding: %w", value.FieldID, err)
			}
		}
	}
	return nil
}

func isCreateSourceGuardField(field *models.FieldMetadata) bool {
	return field != nil && (field.IsMasterDetail || constants.IsReferenceType(constants.NormalizeFieldType(string(field.Type))))
}

func validateFieldMappingRule(value FieldMappingRule, objects map[string]*models.ObjectMetadata) error {
	required := []struct {
		name  string
		value string
	}{
		{name: "sourceObject", value: value.SourceObject},
		{name: "targetObject", value: value.TargetObject},
		{name: "actionType", value: value.ActionType},
		{name: "targetField", value: value.TargetField},
	}
	for _, item := range required {
		if item.value == "" {
			return fmt.Errorf("%s is blank", item.name)
		}
	}
	if value.SourceField == "" && value.DefaultValue == "" {
		return fmt.Errorf("sourceField and defaultValue are both blank")
	}
	if value.ValueMap != "" {
		if _, err := connector.ParseConnectorValueMap(value.ValueMap); err != nil {
			return fmt.Errorf("valueMap is invalid: %w", err)
		}
	}
	switch value.OnUnmappedValue {
	case "", connector.OnUnmappedValueError, connector.OnUnmappedValueSkip, connector.OnUnmappedValuePassthrough:
	default:
		return fmt.Errorf("onUnmappedValue %q is invalid", value.OnUnmappedValue)
	}
	if objects == nil {
		return nil
	}
	source := objects[foldIdentity(value.SourceObject)]
	target := objects[foldIdentity(value.TargetObject)]
	if source == nil || target == nil {
		return fmt.Errorf("source or target object is absent")
	}
	if value.SourceField != "" && models.FindField(source, value.SourceField) == nil {
		return fmt.Errorf("source field %q is absent", value.SourceField)
	}
	if models.FindField(target, value.TargetField) == nil {
		return fmt.Errorf("target field %q is absent", value.TargetField)
	}
	if value.SourceLookupField != "" && models.FindField(target, value.SourceLookupField) == nil {
		return fmt.Errorf("source lookup field %q is absent", value.SourceLookupField)
	}
	return nil
}

func validateConnectorInboundFieldMappingRule(
	value ConnectorInboundFieldMappingRule,
	objects map[string]*models.ObjectMetadata,
	formulaEngine *formula.Engine,
) error {
	if strings.TrimSpace(value.Source) != value.Source {
		return fmt.Errorf("source is not canonical")
	}
	if err := kernelconnector.ValidateConnectorFieldMappingRowWithFormulaEngine(connector.ConnectorFieldMappingInput{
		Source:              value.Source,
		SourceObject:        value.SourceObject,
		TargetObject:        value.TargetObject,
		ActionType:          value.ActionType,
		SourceField:         value.SourceField,
		TargetField:         value.TargetField,
		Direction:           value.Direction,
		TransformExpression: value.TransformExpression,
		ConflictPolicy:      value.ConflictPolicy,
		DefaultValue:        value.DefaultValue,
		ValueMap:            value.ValueMap,
		OnUnmappedValue:     value.OnUnmappedValue,
		SortOrder:           value.SortOrder,
	}, formulaEngine); err != nil {
		return err
	}
	if value.Direction != connector.DirectionInbound {
		return fmt.Errorf("direction %q is not normalized inbound", value.Direction)
	}
	if value.SortOrder < 0 {
		return fmt.Errorf("sortOrder is negative")
	}
	if objects == nil {
		return nil
	}
	target := objects[foldIdentity(value.TargetObject)]
	if target == nil {
		return fmt.Errorf("target object %q is absent", value.TargetObject)
	}
	if models.FindField(target, value.TargetField) == nil {
		return fmt.Errorf("target field %q is absent", value.TargetField)
	}
	return nil
}

func validateConnectorInboundAuthorshipBinding(value ConnectorInboundAuthorshipBinding, objects map[string]*models.ObjectMetadata) error {
	for name, identity := range map[string]string{
		"connectorOwnerKey": value.ConnectorOwnerKey,
		"generation":        value.Generation,
		"actionSlotId":      value.ActionSlotID,
		"actionType":        value.ActionType,
		"sourceObject":      value.SourceObject,
		"destinationSlotId": value.DestinationSlotID,
		"targetObject":      value.TargetObject,
		"profileName":       value.ProfileName,
		"remoteIdField":     value.RemoteIDField,
	} {
		if strings.TrimSpace(identity) == "" || strings.TrimSpace(identity) != identity {
			return fmt.Errorf("%s is blank or non-canonical", name)
		}
	}
	if value.ActionType != constants.ActionTypePushToConnector {
		return fmt.Errorf("actionType %q is not %s", value.ActionType, constants.ActionTypePushToConnector)
	}
	if value.ProfileOwnerPack != "" && strings.TrimSpace(value.ProfileOwnerPack) != value.ProfileOwnerPack {
		return fmt.Errorf("profileOwnerPack is non-canonical")
	}
	if objects != nil {
		object := objects[foldIdentity(value.SourceObject)]
		if object == nil {
			return fmt.Errorf("source object %q is absent", value.SourceObject)
		}
		if models.FindField(object, value.RemoteIDField) == nil {
			return fmt.Errorf("remote id field %q is absent on %q", value.RemoteIDField, value.SourceObject)
		}
	}
	return nil
}

func normalizeConnectorInboundFinalizationAuthority(
	value *ConnectorInboundFinalizationAuthority,
	objects map[string]*models.ObjectMetadata,
) {
	if value == nil {
		return
	}
	sort.Slice(value.OutboundMappings, func(i, j int) bool {
		left, right := value.OutboundMappings[i], value.OutboundMappings[j]
		leftDirectionRank := connectorOutboundMappingDirectionRank(left.Direction)
		rightDirectionRank := connectorOutboundMappingDirectionRank(right.Direction)
		if leftDirectionRank != rightDirectionRank {
			return leftDirectionRank < rightDirectionRank
		}
		if left.SortOrder != right.SortOrder {
			return left.SortOrder < right.SortOrder
		}
		if foldIdentity(left.SourceField) != foldIdentity(right.SourceField) {
			return foldIdentity(left.SourceField) < foldIdentity(right.SourceField)
		}
		return foldIdentity(left.TargetField) < foldIdentity(right.TargetField)
	})
	if value.RecoverySourceEligibility != nil {
		if source := objects[foldIdentity(value.Contract.SourceObject)]; source != nil {
			for index, fieldName := range value.RecoverySourceEligibility.RequiredFalseBooleanFields {
				if field := models.FindField(source, fieldName); field != nil {
					// Catalog publishes the schema-owned API spelling. Runtime record
					// maps use that spelling exactly, so consumers never need a second
					// case-insensitive field lookup or repository fallback.
					value.RecoverySourceEligibility.RequiredFalseBooleanFields[index] = field.APIName
				}
			}
		}
		sort.Slice(value.RecoverySourceEligibility.RequiredFalseBooleanFields, func(i, j int) bool {
			return foldIdentity(value.RecoverySourceEligibility.RequiredFalseBooleanFields[i]) <
				foldIdentity(value.RecoverySourceEligibility.RequiredFalseBooleanFields[j])
		})
	}
}

func connectorOutboundMappingDirectionRank(direction string) int {
	if direction == connector.DirectionOutbound {
		return 0
	}
	return 1
}

func validateConnectorInboundFinalizationAuthority(
	value ConnectorInboundFinalizationAuthority,
	objects map[string]*models.ObjectMetadata,
	formulaEngine *formula.Engine,
) error {
	contract := value.Contract
	profileOwner := strings.TrimSpace(contract.ProfileOwnerPack)
	if profileOwner == "" {
		profileOwner = strings.TrimSpace(contract.ConnectorOwnerKey)
	}
	for name, identity := range map[string]string{
		"contractGeneration": value.ContractGeneration,
		"connectorOwnerKey":  contract.ConnectorOwnerKey,
		"actionSlotId":       contract.ActionSlotID,
		"actionType":         contract.ActionType,
		"flowDeveloperName":  contract.FlowDeveloperName,
		"flowActionName":     contract.FlowActionName,
		"sourceObject":       contract.SourceObject,
		"destinationSlotId":  contract.DestinationSlotID,
		"targetObject":       contract.TargetObject,
		"profileName":        contract.ProfileName,
		"createPathTemplate": contract.CreatePathTemplate,
		"updatePathTemplate": contract.UpdatePathTemplate,
		"responseIdPath":     contract.ResponseIDPath,
		"remoteIdField":      contract.RemoteIDField,
		"profileGeneration":  value.Profile.Generation,
		"profileOwnerPack":   value.Profile.OwnerPack,
		"profileProfileName": value.Profile.ProfileName,
		"credentialRef":      value.Profile.CredentialRef,
		"endpointClass":      value.Profile.EndpointClass,
	} {
		if strings.TrimSpace(identity) == "" || strings.TrimSpace(identity) != identity {
			return fmt.Errorf("%s is blank or non-canonical", name)
		}
	}
	if contract.ActionType != constants.ActionTypePushToConnector {
		return fmt.Errorf("actionType %q is not %s", contract.ActionType, constants.ActionTypePushToConnector)
	}
	if value.Profile.OwnerPack != profileOwner || value.Profile.ProfileName != contract.ProfileName {
		return fmt.Errorf("profile identity does not match the connector contract")
	}
	// Preserve the existing ConnectorPushProfile contract exactly. Its engine
	// treats zero/negative delays as zero and Multiplier <= 0 as the default 1;
	// finalization currently requires only an explicit positive attempt count.
	// Catalog must not create a stricter profile language or change the durable
	// profile hash while moving the read path behind an immutable view.
	if value.Profile.MaxAttempts < 1 {
		return fmt.Errorf("profile retry policy is invalid")
	}
	if len(value.OutboundMappings) == 0 {
		return fmt.Errorf("outbound mapping closure is empty")
	}
	if eligibility := value.RecoverySourceEligibility; eligibility != nil {
		if len(eligibility.RequiredFalseBooleanFields) == 0 {
			return fmt.Errorf("recovery source eligibility is empty")
		}
		seenEligibility := make(map[string]struct{}, len(eligibility.RequiredFalseBooleanFields))
		for index, fieldName := range eligibility.RequiredFalseBooleanFields {
			if strings.TrimSpace(fieldName) == "" || strings.TrimSpace(fieldName) != fieldName {
				return fmt.Errorf("recovery source eligibility field %d is blank or non-canonical", index)
			}
			key := foldIdentity(fieldName)
			if _, duplicate := seenEligibility[key]; duplicate {
				return fmt.Errorf("recovery source eligibility duplicates field %q", fieldName)
			}
			seenEligibility[key] = struct{}{}
			if objects != nil {
				source := objects[foldIdentity(contract.SourceObject)]
				if source == nil {
					return fmt.Errorf("source object %q is absent", contract.SourceObject)
				}
				field := models.FindField(source, fieldName)
				if field == nil {
					return fmt.Errorf("recovery source eligibility field %q is absent on %q", fieldName, contract.SourceObject)
				}
				if constants.NormalizeFieldType(string(field.Type)) != constants.FieldTypeBoolean {
					return fmt.Errorf("recovery source eligibility field %q on %q is not boolean", fieldName, contract.SourceObject)
				}
			}
		}
	}
	seenSourceFields := make(map[string]struct{}, len(value.OutboundMappings))
	seenTargetFields := make(map[string]struct{}, len(value.OutboundMappings))
	for index, mapping := range value.OutboundMappings {
		if err := kernelconnector.ValidateConnectorFieldMappingRowWithFormulaEngine(mapping, formulaEngine); err != nil {
			return fmt.Errorf("outbound mapping %d is invalid: %w", index, err)
		}
		if mapping.Direction != connector.DirectionOutbound && mapping.Direction != connector.DirectionBidirectional {
			return fmt.Errorf("outbound mapping %d direction %q is not outbound-capable", index, mapping.Direction)
		}
		if mapping.Source != contract.ConnectorOwnerKey ||
			!strings.EqualFold(mapping.SourceObject, contract.SourceObject) ||
			!strings.EqualFold(mapping.TargetObject, contract.TargetObject) ||
			!strings.EqualFold(mapping.ActionType, contract.ActionType) {
			return fmt.Errorf("outbound mapping %d does not match the connector contract", index)
		}
		sourceKey := foldIdentity(mapping.SourceField)
		if _, duplicate := seenSourceFields[sourceKey]; duplicate {
			return fmt.Errorf("outbound mapping %d duplicates effective source field %q", index, mapping.SourceField)
		}
		seenSourceFields[sourceKey] = struct{}{}
		targetKey := foldIdentity(mapping.TargetField)
		if _, duplicate := seenTargetFields[targetKey]; duplicate {
			return fmt.Errorf("outbound mapping %d duplicates effective target field %q", index, mapping.TargetField)
		}
		seenTargetFields[targetKey] = struct{}{}
	}
	if provenanceField := strings.TrimSpace(contract.RemoteIDProvenanceField); provenanceField != "" &&
		strings.EqualFold(strings.TrimSpace(contract.RemoteIDField), provenanceField) {
		return fmt.Errorf("remote id and provenance fields must be distinct")
	}
	if objects != nil {
		source := objects[foldIdentity(contract.SourceObject)]
		if source == nil {
			return fmt.Errorf("source object %q is absent", contract.SourceObject)
		}
		if models.FindField(source, contract.RemoteIDField) == nil {
			return fmt.Errorf("remote id field %q is absent on %q", contract.RemoteIDField, contract.SourceObject)
		}
		if provenanceField := strings.TrimSpace(contract.RemoteIDProvenanceField); provenanceField != "" &&
			models.FindField(source, provenanceField) == nil {
			return fmt.Errorf("remote id provenance field %q is absent on %q", provenanceField, contract.SourceObject)
		}
		for index, mapping := range value.OutboundMappings {
			if models.FindField(source, mapping.SourceField) == nil {
				return fmt.Errorf("outbound mapping %d source field %q is absent", index, mapping.SourceField)
			}
		}
	}
	return nil
}

func registerCanonicalFinalizationFact(seen map[string][]byte, key string, value any) error {
	encoded, err := marshalCanonical(value)
	if err != nil {
		return err
	}
	if existing, ok := seen[key]; ok && !bytes.Equal(existing, encoded) {
		return fmt.Errorf("selector %q resolves to more than one immutable fact", key)
	}
	seen[key] = encoded
	return nil
}

func connectorInboundAuthorshipForFinalization(value ConnectorInboundFinalizationAuthority) ConnectorInboundAuthorshipBinding {
	contract := value.Contract
	profileOwner := strings.TrimSpace(contract.ProfileOwnerPack)
	if profileOwner == "" {
		profileOwner = value.Profile.OwnerPack
	}
	return ConnectorInboundAuthorshipBinding{
		ConnectorOwnerKey: contract.ConnectorOwnerKey,
		Generation:        value.ContractGeneration,
		ActionSlotID:      contract.ActionSlotID,
		ActionType:        contract.ActionType,
		SourceObject:      contract.SourceObject,
		DestinationSlotID: contract.DestinationSlotID,
		TargetObject:      contract.TargetObject,
		ProfileOwnerPack:  profileOwner,
		ProfileName:       contract.ProfileName,
		RemoteIDField:     contract.RemoteIDField,
	}
}

func validateUniqueMutationIdentity(seen map[string]struct{}, kind, identity string, index int) error {
	path := fmt.Sprintf("mutation.%s[%d]", kind, index)
	if err := requireCanonicalIdentity(path+".id", identity); err != nil {
		return err
	}
	key := kind + "\x00" + foldIdentity(identity)
	if _, exists := seen[key]; exists {
		return fmt.Errorf("%s duplicates %s %q", path, kind, identity)
	}
	seen[key] = struct{}{}
	return nil
}