package catalogsnapshot

import (
	"bytes"
	"encoding/json"
	"errors"
	"sort"
	"strings"
	"testing"
	"time"

	"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"
)

const (
	testMutationCapability = "runtime.records.mutate.v1"
	testMutationComponent  = "definition.record-mutation"
	testOAuthPolicyObject  = "CredentialClient"
)

func TestRecordMutationProjectionIsDeterministicRevisionClosedAndSingleFact(t *testing.T) {
	input := mutationCompileInput()
	first := mustCompile(t, input)

	reordered := mutationCompileInput()
	reverseObjects(reordered.Objects)
	reverseRelationships(reordered.Relationships)
	for left, right := 0, len(reordered.MutationDefinitions)-1; left < right; left, right = left+1, right-1 {
		reordered.MutationDefinitions[left], reordered.MutationDefinitions[right] = reordered.MutationDefinitions[right], reordered.MutationDefinitions[left]
	}
	second := mustCompile(t, reordered)
	if first.Digest() != second.Digest() || !bytes.Equal(first.Bytes(), second.Bytes()) {
		t.Fatal("mutation input ordering changed canonical snapshot bytes")
	}

	viewArtifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, first), ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project mutation view: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if view.Profile() != ProjectionRecordMutationV1 || view.Lineage().CatalogRevision != testRevision {
		t.Fatalf("mutation view scope/lineage = %q/%#v", view.Profile(), view.Lineage())
	}
	workItem, ok := view.Object("WorkItem")
	if !ok || workItem.KeyPrefix == nil || *workItem.KeyPrefix != "0WI" || workItem.AccessPolicy == nil || workItem.AccessPolicy.RecordMutation == nil {
		t.Fatalf("write-engine object closure was not reconstructed: %#v", workItem)
	}
	subject := models.FindField(workItem, "Subject")
	if subject == nil || subject.ID != "field-subject" || !subject.Createable || !subject.Updateable || !subject.TrackHistory || !subject.DefaultedOnCreate {
		t.Fatalf("write-engine field closure was not reconstructed: %#v", subject)
	}
	recordTypes := view.RecordTypesForObject("WorkItem")
	if len(view.ValidationRulesForObject("WorkItem")) != 1 || len(recordTypes) != 1 {
		t.Fatal("mutation validation/record-type closure is incomplete")
	}
	if recordTypes[0].Name != "Internal" || recordTypes[0].DeveloperName != "Internal" ||
		recordTypes[0].Label != "" || recordTypes[0].Description != nil {
		t.Fatalf("record-type execution identity/presentation projection = %#v", recordTypes[0])
	}
	if version, ok := view.ActiveFlowVersion("NormalizeWorkItem"); !ok || version.Id != "flow-version-1" || !version.DefinitionRuntimeLoaded {
		t.Fatalf("active flow closure = %#v, %v", version, ok)
	}
	if version, ok := view.ActiveFlowVersion("NormalizeWorkspace"); !ok || version.Id != "flow-version-child" {
		t.Fatalf("transitive subflow closure = %#v, %v", version, ok)
	}
	if rules := view.AutoNumberRulesForObject("WorkItem"); len(rules) != 1 || rules[0].FieldID != "field-subject" {
		t.Fatalf("autonumber closure = %#v", rules)
	}
	if rules := view.FieldMappingRulesForTarget("WorkItem"); len(rules) != 1 || rules[0].SourceObject != "Workspace" || rules[0].TargetField != "Subject" {
		t.Fatalf("field-mapping closure = %#v", rules)
	}
	if guards := view.CreateFromSourceGuardsForTarget("WorkItem"); len(guards) != 1 || guards[0].SourceFKField != "WorkspaceId" || guards[0].OwningStandardActionID != "action-create-work-item" {
		t.Fatalf("create-from-source guard closure = %#v", guards)
	}
	if !view.AllowsPublicOAuthScope("External.Records.Read") {
		t.Fatal("ownerless public OAuth scope was not projected with its owning metadata object")
	}
	if view.AllowsPublicOAuthScope("pack.records.read") {
		t.Fatal("pack-owned OAuth scope was exposed through the ownerless policy surface")
	}
	if !view.AllowsPublicOAuthScopeForPack("pack.records.read", "io.example.pack") {
		t.Fatal("pack-owned OAuth scope owner was not preserved in the neutral projection")
	}
	var foundCascade bool
	for _, relationship := range view.RelationshipsForChild("WorkItem") {
		if relationship.ParentObjectAPIName == "Unrelated" && relationship.FieldAPIName == "UnrelatedId" {
			foundCascade = relationship.CascadeDelete && !relationship.RestrictedDelete
		}
	}
	if !foundCascade {
		t.Fatalf("mutation relationship delete policy was not reconstructed: %#v", view.RelationshipsForChild("WorkItem"))
	}
	if bytes.Contains(viewArtifact.Bytes(), []byte(`"label":"Work Item"`)) {
		t.Fatal("mutation view contains object presentation metadata")
	}
	for name, payload := range map[string][]byte{
		"Catalog snapshot": first.Bytes(),
		"Runtime view":     viewArtifact.Bytes(),
	} {
		if bytes.Contains(payload, []byte(`"identifierRule`)) {
			t.Fatalf("%s reintroduced retired IdentifierRule metadata", name)
		}
	}

	// Every public getter must detach nested values from the immutable view.
	// Mutating one caller's result must not alter later requests.
	subject.Options[0].Tags[0] = "changed"
	subject.Options[0].Extensions["opaque"][0] = 'x'
	subject.PicklistDependency["Open"][0] = "changed"
	*subject.RollupConfig.Filter = "changed"
	*subject.RollupConfig.ResultPrecision = 1
	recordTypes = view.RecordTypesForObject("WorkItem")
	*recordTypes[0].BusinessProcessID = "changed"
	again, ok := view.Object("WorkItem")
	if !ok {
		t.Fatal("mutation object disappeared after caller mutation")
	}
	againSubject := models.FindField(again, "Subject")
	againRecordTypes := view.RecordTypesForObject("WorkItem")
	if againSubject.Options[0].Tags[0] != "active" || string(againSubject.Options[0].Extensions["opaque"]) != `{"source":"catalog"}` ||
		againSubject.PicklistDependency["Open"][0] != "Ready" || *againSubject.RollupConfig.Filter != "Status = 'Open'" ||
		*againSubject.RollupConfig.ResultPrecision != 18 || *againRecordTypes[0].BusinessProcessID != "process-work-item" {
		t.Fatalf("caller mutation changed immutable view: field=%#v recordTypes=%#v", againSubject, againRecordTypes)
	}
}

func TestFieldMappingValuePoliciesRequireVersionedCatalogAndRuntimeGrammars(t *testing.T) {
	input := mutationCompileInput()
	for index := range input.MutationDefinitions {
		rule := input.MutationDefinitions[index].FieldMappingRule
		if rule == nil {
			continue
		}
		rule.ValueMap = `{"legacy":"Current"}`
		rule.OnUnmappedValue = connector.OnUnmappedValueSkip
	}

	snapshotArtifact := mustCompile(t, input)
	snapshot := mustOpenSnapshot(t, snapshotArtifact)
	if snapshot.doc.Schema != CatalogSnapshotSchemaV18 {
		t.Fatalf("Catalog schema = %q, want %q", snapshot.doc.Schema, CatalogSnapshotSchemaV18)
	}
	effective, err := BuildEffectiveMetadata(input)
	if err != nil {
		t.Fatalf("BuildEffectiveMetadata: %v", err)
	}
	var effectiveDoc catalogDocument
	if err := decodeStrict(effective.Bytes(), &effectiveDoc); err != nil {
		t.Fatalf("decode effective metadata: %v", err)
	}
	if effectiveDoc.Schema != EffectiveMetadataSchemaV17 {
		t.Fatalf("effective metadata schema = %q, want %q", effectiveDoc.Schema, EffectiveMetadataSchemaV17)
	}
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project Runtime view: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if view.doc.Schema != RuntimeMetadataViewSchemaV25 {
		t.Fatalf("Runtime view schema = %q, want %q", view.doc.Schema, RuntimeMetadataViewSchemaV25)
	}
	rules := view.FieldMappingRulesForTarget("WorkItem")
	if len(rules) != 1 || rules[0].ValueMap != `{"legacy":"Current"}` ||
		rules[0].OnUnmappedValue != connector.OnUnmappedValueSkip {
		t.Fatalf("Runtime field-mapping policies = %#v", rules)
	}

	downgradedCatalog := snapshot.doc
	downgradedCatalog.Schema = CatalogSnapshotSchemaV4
	downgradedCatalogBytes, err := marshalCanonical(downgradedCatalog)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenCatalogSnapshot(makeArtifact(ArtifactCatalogSnapshot, downgradedCatalogBytes)); err == nil || !strings.Contains(err.Error(), "pre-v18 Catalog snapshot contains field-mapping value policies") {
		t.Fatalf("downgraded Catalog error = %v", err)
	}

	downgradedRuntime := view.doc
	downgradedRuntime.Schema = RuntimeMetadataViewSchemaV4
	downgradedRuntimeBytes, err := marshalCanonical(downgradedRuntime)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, downgradedRuntimeBytes)); err == nil || !strings.Contains(err.Error(), "pre-v25 Runtime metadata view contains field-mapping value policies") {
		t.Fatalf("downgraded Runtime error = %v", err)
	}
}

func TestRuntimeRollupDependencyIndexReturnsDetachedMatchingFields(t *testing.T) {
	artifact := mustCompile(t, mutationCompileInput())
	viewArtifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, artifact), ProjectionKey{
		AppInstallationID: testInstallA,
		CapabilityID:      testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project mutation view: %v", err)
	}
	view := mustOpenView(t, viewArtifact)

	dependencies := view.RollupChildDependencies("unrelated")
	if len(dependencies) != 1 || dependencies[0].ParentObjectAPIName != "WorkItem" ||
		dependencies[0].RollupField.APIName != "Subject" ||
		dependencies[0].RollupField.RollupConfig == nil ||
		dependencies[0].RollupField.RollupConfig.SummaryObject != "Unrelated" {
		t.Fatalf("indexed dependencies = %#v", dependencies)
	}
	if !view.HasRollupChildDependencies("UNRELATED") ||
		view.HasRollupChildDependencies("Workspace") ||
		len(view.RollupChildDependencies("Workspace")) != 0 {
		t.Fatal("rollup dependency index did not preserve case folding or empty lookup")
	}

	*dependencies[0].RollupField.RollupConfig.Filter = "changed"
	again := view.RollupChildDependencies("Unrelated")
	if len(again) != 1 || again[0].RollupField.RollupConfig == nil ||
		again[0].RollupField.RollupConfig.Filter == nil ||
		*again[0].RollupField.RollupConfig.Filter != "Status = 'Open'" {
		t.Fatalf("caller mutation changed immutable rollup index: %#v", again)
	}
}

func TestRuntimeMutationViewIndexesSiblingMaxPresence(t *testing.T) {
	baseArtifact := mustCompile(t, mutationCompileInput())
	baseViewArtifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, baseArtifact), ProjectionKey{
		AppInstallationID: testInstallA,
		CapabilityID:      testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project base mutation view: %v", err)
	}
	if mustOpenView(t, baseViewArtifact).HasAnySiblingMaxDefaults() {
		t.Fatal("base mutation view unexpectedly reports sibling-max defaults")
	}

	input := mutationCompileInput()
	parentField := "WorkspaceId"
	input.Objects[0].Metadata.Fields = append(input.Objects[0].Metadata.Fields, models.FieldMetadata{
		ID: "field-work-item-sort-order", APIName: "SortOrder__c", Label: "Sort Order",
		Type: models.FieldType("integer"), Createable: true, Updateable: true,
		SiblingMaxParentField: &parentField,
	})
	artifact := mustCompile(t, input)
	viewArtifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, artifact), ProjectionKey{
		AppInstallationID: testInstallA,
		CapabilityID:      testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project sibling-max mutation view: %v", err)
	}
	if !mustOpenView(t, viewArtifact).HasAnySiblingMaxDefaults() {
		t.Fatal("sibling-max mutation view did not retain its precomputed safety fact")
	}
}

func TestConnectorRuntimeFactsAreImmutableRevisionBoundAndTargetScoped(t *testing.T) {
	input := mutationCompileInput()
	input.MutationDefinitions = append(input.MutationDefinitions,
		MutationDefinitionInput{
			CatalogRevision: testRevision,
			ConnectorInboundFieldMappingRule: &ConnectorInboundFieldMappingRule{
				Source: "hubspot", SourceObject: "contacts",
				TargetObject: "WorkItem", ActionType: "upsert",
				SourceField: "firstname", TargetField: "Subject",
				Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite,
				TransformExpression: "UPPER(firstname)",
				OnUnmappedValue:     connector.OnUnmappedValueError, SortOrder: 10,
			},
		},
		MutationDefinitionInput{
			CatalogRevision: testRevision,
			ConnectorInboundFieldMappingRule: &ConnectorInboundFieldMappingRule{
				Source: "hubspot", PackAuthored: true, SourceObject: "contacts",
				TargetObject: "WorkItem", ActionType: "upsert",
				SourceField: "firstname", TargetField: "Subject",
				Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite,
				TransformExpression: "UPPER(firstname)",
				OnUnmappedValue:     connector.OnUnmappedValueError, SortOrder: 10,
			},
		},
		MutationDefinitionInput{
			CatalogRevision: testRevision,
			ConnectorInboundAuthorship: &ConnectorInboundAuthorshipBinding{
				ConnectorOwnerKey: "hubspot-crm-connector", Generation: "pack-installation-17",
				ActionSlotID: "hubspot-contact-upsert", ActionType: constants.ActionTypePushToConnector,
				SourceObject: "WorkItem", DestinationSlotID: "hubspot-contact", TargetObject: "contacts",
				ProfileOwnerPack: "pingcap-hubspot-companion", ProfileName: "hubspot-main", RemoteIDField: "Subject",
			},
		},
	)
	artifact := mustCompile(t, input)
	snapshot := mustOpenSnapshot(t, artifact)
	if snapshot.doc.Schema != CatalogSnapshotSchemaV11 {
		t.Fatalf("Catalog schema=%q want=%q", snapshot.doc.Schema, CatalogSnapshotSchemaV11)
	}
	effective, err := BuildEffectiveMetadata(input)
	if err != nil {
		t.Fatalf("BuildEffectiveMetadata: %v", err)
	}
	var effectiveDoc catalogDocument
	if err := decodeStrict(effective.Bytes(), &effectiveDoc); err != nil {
		t.Fatalf("decode effective metadata: %v", err)
	}
	if effectiveDoc.Schema != EffectiveMetadataSchemaV10 {
		t.Fatalf("effective metadata schema=%q want=%q", effectiveDoc.Schema, EffectiveMetadataSchemaV10)
	}
	recompiled, err := CompileEffectiveMetadata(effective)
	if err != nil {
		t.Fatalf("CompileEffectiveMetadata: %v", err)
	}
	if !bytes.Equal(recompiled.Bytes(), artifact.Bytes()) {
		t.Fatal("effective metadata round trip changed Catalog snapshot bytes")
	}
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA,
		CapabilityID:      testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	view := mustOpenView(t, viewArtifact)
	if view.doc.Schema != RuntimeMetadataViewSchemaV13 {
		t.Fatalf("Runtime view schema=%q want=%q", view.doc.Schema, RuntimeMetadataViewSchemaV13)
	}
	legacyMappings := view.FieldMappingRulesForTarget("workitem")
	if len(legacyMappings) != 1 || legacyMappings[0].SourceObject != "Workspace" {
		t.Fatalf("legacy mapping closure=%#v", legacyMappings)
	}
	mappings := view.InboundConnectorFieldMappingRulesForTarget("workitem")
	packMappings := 0
	for _, mapping := range mappings {
		if mapping.Source != "hubspot" {
			t.Fatalf("connector mapping getter returned non-connector rule: %#v", mapping)
		}
		if mapping.PackAuthored {
			packMappings++
		}
	}
	if len(mappings) != 2 || packMappings != 1 {
		t.Fatalf("connector mapping closure=%#v", mappings)
	}
	authorship := view.ConnectorInboundAuthorshipBindings("hubspot-crm-connector")
	if len(authorship) != 1 || authorship[0].Generation != "pack-installation-17" || authorship[0].RemoteIDField != "Subject" {
		t.Fatalf("connector authorship closure=%#v", authorship)
	}
	if caseVariant := view.ConnectorInboundAuthorshipBindings("HubSpot-CRM-Connector"); len(caseVariant) != 0 {
		t.Fatalf("case-variant opaque owner matched connector authorship: %#v", caseVariant)
	}

	changed := input
	changed.MutationDefinitions = append([]MutationDefinitionInput(nil), input.MutationDefinitions...)
	for index := range changed.MutationDefinitions {
		if mapping := changed.MutationDefinitions[index].ConnectorInboundFieldMappingRule; mapping != nil && mapping.Source == "hubspot" {
			copy := *mapping
			copy.TargetField = "Id"
			changed.MutationDefinitions[index].ConnectorInboundFieldMappingRule = &copy
		}
	}
	if mustCompile(t, changed).Digest() == artifact.Digest() {
		t.Fatal("connector runtime fact change did not change Catalog digest")
	}
}

func TestConnectorInboundFinalizationAuthorityIsRevisionBoundAndClosed(t *testing.T) {
	input := mutationCompileInput()
	authorship, authority := connectorInboundFinalizationTestAuthority()
	input.MutationDefinitions = append(input.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &authorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &authority},
	)
	artifact := mustCompile(t, input)
	snapshot := mustOpenSnapshot(t, artifact)
	if snapshot.doc.Schema != CatalogSnapshotSchemaV12 {
		t.Fatalf("Catalog schema=%q want=%q", snapshot.doc.Schema, CatalogSnapshotSchemaV12)
	}
	effective, err := BuildEffectiveMetadata(input)
	if err != nil {
		t.Fatalf("BuildEffectiveMetadata: %v", err)
	}
	var effectiveDoc catalogDocument
	if err := decodeStrict(effective.Bytes(), &effectiveDoc); err != nil {
		t.Fatalf("decode effective metadata: %v", err)
	}
	if effectiveDoc.Schema != EffectiveMetadataSchemaV11 {
		t.Fatalf("effective metadata schema=%q want=%q", effectiveDoc.Schema, EffectiveMetadataSchemaV11)
	}
	recompiled, err := CompileEffectiveMetadata(effective)
	if err != nil {
		t.Fatalf("CompileEffectiveMetadata: %v", err)
	}
	if !bytes.Equal(recompiled.Bytes(), artifact.Bytes()) {
		t.Fatal("effective finalization metadata round trip changed Catalog snapshot bytes")
	}
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA,
		CapabilityID:      testMutationCapability,
	})
	if err != nil {
		t.Fatalf("ProjectRuntimeMetadataView: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if view.doc.Schema != RuntimeMetadataViewSchemaV14 {
		t.Fatalf("Runtime view schema=%q want=%q", view.doc.Schema, RuntimeMetadataViewSchemaV14)
	}
	values := view.ConnectorInboundFinalizationAuthorities("example-connector-pack")
	if len(values) != 1 || values[0].ContractGeneration != "pack-installation-17" ||
		values[0].Profile.Generation != "profile-installation-9" ||
		len(values[0].OutboundMappings) != 1 {
		t.Fatalf("connector inbound finalization closure=%#v", values)
	}
	if got := view.ConnectorInboundFinalizationAuthorities("Example-Connector-Pack"); len(got) != 0 {
		t.Fatalf("case-variant owner matched finalization authority: %#v", got)
	}
	if values[0].Profile.RetryableStatusCodes[0] != 503 {
		t.Fatalf("profile status-code order was rewritten: %#v", values[0].Profile.RetryableStatusCodes)
	}
	values[0].Profile.RetryableStatusCodes[0] = 500
	values[0].OutboundMappings[0].TargetField = "changed"
	again := view.ConnectorInboundFinalizationAuthorities("example-connector-pack")
	if again[0].Profile.RetryableStatusCodes[0] != 503 || again[0].OutboundMappings[0].TargetField != "subject" {
		t.Fatalf("caller mutation changed immutable finalization authority: %#v", again)
	}

	changed := mutationCompileInput()
	changedAuthorship, changedAuthority := connectorInboundFinalizationTestAuthority()
	changedAuthority.Profile.Generation = "profile-installation-10"
	changed.MutationDefinitions = append(changed.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &changedAuthorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &changedAuthority},
	)
	if mustCompile(t, changed).Digest() == artifact.Digest() {
		t.Fatal("profile generation change did not change Catalog digest")
	}

	invalid := mutationCompileInput()
	invalidAuthorship, invalidAuthority := connectorInboundFinalizationTestAuthority()
	invalidAuthority.Profile.OwnerPack = "different-profile-owner"
	invalid.MutationDefinitions = append(invalid.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &invalidAuthorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &invalidAuthority},
	)
	if _, err := CompileCatalogSnapshot(invalid); err == nil {
		t.Fatal("mixed contract/profile authority was accepted")
	}

	missingAuthorship := mutationCompileInput()
	_, missingAuthority := connectorInboundFinalizationTestAuthority()
	missingAuthorship.MutationDefinitions = append(missingAuthorship.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &missingAuthority},
	)
	if _, err := CompileCatalogSnapshot(missingAuthorship); err == nil {
		t.Fatal("finalization authority without exact authorship binding was accepted")
	}

	for name, clearRequiredContractField := range map[string]func(*connector.ConnectorPushDurableMetadataContract){
		"create path": func(value *connector.ConnectorPushDurableMetadataContract) {
			value.CreatePathTemplate = ""
		},
		"update path": func(value *connector.ConnectorPushDurableMetadataContract) {
			value.UpdatePathTemplate = " "
		},
		"response id path": func(value *connector.ConnectorPushDurableMetadataContract) {
			value.ResponseIDPath = ""
		},
	} {
		t.Run("reject missing "+name, func(t *testing.T) {
			missingContractField := mutationCompileInput()
			authorship, authority := connectorInboundFinalizationTestAuthority()
			clearRequiredContractField(&authority.Contract)
			missingContractField.MutationDefinitions = append(
				missingContractField.MutationDefinitions,
				MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &authorship},
				MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &authority},
			)
			if _, err := CompileCatalogSnapshot(missingContractField); err == nil {
				t.Fatalf("finalization authority without %s was accepted", name)
			}
		})
	}

	mixedGeneration := mutationCompileInput()
	mixedAuthorship, mixedAuthority := connectorInboundFinalizationTestAuthority()
	mixedAuthorship.Generation = "pack-installation-18"
	mixedGeneration.MutationDefinitions = append(mixedGeneration.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &mixedAuthorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &mixedAuthority},
	)
	if _, err := CompileCatalogSnapshot(mixedGeneration); err == nil {
		t.Fatal("mixed authorship/contract generation was accepted")
	}

	defaultRetry := mutationCompileInput()
	defaultRetryAuthorship, defaultRetryAuthority := connectorInboundFinalizationTestAuthority()
	defaultRetryAuthority.Profile.BaseDelayNanos = 0
	defaultRetryAuthority.Profile.Multiplier = 0
	defaultRetryAuthority.Profile.MaxDelayNanos = 0
	defaultRetryAuthority.Profile.RetryableStatusCodes = nil
	defaultRetry.MutationDefinitions = append(defaultRetry.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &defaultRetryAuthorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &defaultRetryAuthority},
	)
	if _, err := CompileCatalogSnapshot(defaultRetry); err != nil {
		t.Fatalf("existing default retry policy was rejected: %v", err)
	}

	bidirectional := mutationCompileInput()
	bidirectionalAuthorship, bidirectionalAuthority := connectorInboundFinalizationTestAuthority()
	bidirectionalAuthority.OutboundMappings = append(bidirectionalAuthority.OutboundMappings,
		connector.ConnectorFieldMappingInput{
			Source: bidirectionalAuthority.Contract.ConnectorOwnerKey, SourceObject: bidirectionalAuthority.Contract.SourceObject,
			TargetObject: bidirectionalAuthority.Contract.TargetObject, ActionType: bidirectionalAuthority.Contract.ActionType,
			SourceField: "Id", TargetField: "remote_id", SortOrder: 1,
			Direction: connector.DirectionBidirectional,
		},
	)
	bidirectionalAuthority.OutboundMappings[0].SortOrder = 10
	bidirectional.MutationDefinitions = append(bidirectional.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &bidirectionalAuthorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &bidirectionalAuthority},
	)
	bidirectionalSnapshot := mustOpenSnapshot(t, mustCompile(t, bidirectional))
	bidirectionalViewArtifact, err := ProjectRuntimeMetadataView(bidirectionalSnapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	bidirectionalView := mustOpenView(t, bidirectionalViewArtifact)
	ordered := bidirectionalView.ConnectorInboundFinalizationAuthorities("example-connector-pack")[0].OutboundMappings
	if len(ordered) != 2 || ordered[0].Direction != connector.DirectionOutbound || ordered[1].Direction != connector.DirectionBidirectional {
		t.Fatalf("outbound effective mapping order = %#v", ordered)
	}

	for name, mutate := range map[string]func(*ConnectorInboundFinalizationAuthority){
		"owner generation": func(value *ConnectorInboundFinalizationAuthority) {
			value.ContractGeneration = "pack-installation-18"
		},
		"profile authority": func(value *ConnectorInboundFinalizationAuthority) {
			value.Profile.Generation = "profile-installation-10"
		},
		"mapping closure": func(value *ConnectorInboundFinalizationAuthority) {
			value.OutboundMappings[0].TargetField = "display_name"
		},
	} {
		t.Run("reject conflicting "+name, func(t *testing.T) {
			conflicting := mutationCompileInput()
			firstAuthorship, firstAuthority := connectorInboundFinalizationTestAuthority()
			secondAuthorship, secondAuthority := connectorInboundFinalizationTestAuthority()
			secondAuthority.Contract.FlowDeveloperName = "WorkItem_To_Remote_Alternate"
			secondAuthority.Contract.FlowActionName = "PushWorkItemAlternate"
			secondAuthority.Contract.ActionSlotID = "remote-contact-upsert-alternate"
			secondAuthority.Contract.DestinationSlotID = "remote-contact-alternate"
			secondAuthority.Contract.RemoteIDField = "Id"
			mutate(&secondAuthority)
			secondAuthorship = connectorInboundAuthorshipForFinalization(secondAuthority)
			conflicting.MutationDefinitions = append(conflicting.MutationDefinitions,
				MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &firstAuthorship},
				MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &firstAuthority},
				MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &secondAuthorship},
				MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &secondAuthority},
			)
			if _, err := CompileCatalogSnapshot(conflicting); err == nil {
				t.Fatalf("conflicting %s was accepted", name)
			}
		})
	}
}

func TestConnectorRecoveryEligibilityIsRevisionBoundAndBooleanOnly(t *testing.T) {
	input := mutationCompileInput()
	input.Objects[0].Metadata.Fields = append(input.Objects[0].Metadata.Fields, models.FieldMetadata{
		ID: "field-recovery-deleted", APIName: "Archived", Type: constants.FieldTypeBoolean,
	})
	authorship, authority := connectorInboundFinalizationTestAuthority()
	authority.RecoverySourceEligibility = &connector.ConnectorPushRecoverySourceEligibilityContract{
		// Authored identities are matched case-insensitively, but the immutable
		// Runtime view must carry the schema-owned API spelling used by records.
		RequiredFalseBooleanFields: []string{"archived"},
	}
	input.MutationDefinitions = append(input.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &authorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &authority},
	)
	artifact := mustCompile(t, input)
	snapshot := mustOpenSnapshot(t, artifact)
	if snapshot.doc.Schema != CatalogSnapshotSchemaV13 {
		t.Fatalf("Catalog schema=%q want=%q", snapshot.doc.Schema, CatalogSnapshotSchemaV13)
	}
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatalf("ProjectRuntimeMetadataView: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if view.doc.Schema != RuntimeMetadataViewSchemaV15 {
		t.Fatalf("Runtime view schema=%q want=%q", view.doc.Schema, RuntimeMetadataViewSchemaV15)
	}
	values := view.ConnectorInboundFinalizationAuthorities("example-connector-pack")
	if len(values) != 1 || values[0].RecoverySourceEligibility == nil ||
		len(values[0].RecoverySourceEligibility.RequiredFalseBooleanFields) != 1 ||
		values[0].RecoverySourceEligibility.RequiredFalseBooleanFields[0] != "Archived" {
		t.Fatalf("recovery eligibility=%#v", values)
	}
	values[0].RecoverySourceEligibility.RequiredFalseBooleanFields[0] = "Changed"
	if got := view.ConnectorInboundFinalizationAuthorities("example-connector-pack")[0].RecoverySourceEligibility.RequiredFalseBooleanFields[0]; got != "Archived" {
		t.Fatalf("caller mutation changed immutable eligibility: %q", got)
	}

	invalid := mutationCompileInput()
	invalidAuthorship, invalidAuthority := connectorInboundFinalizationTestAuthority()
	invalidAuthority.RecoverySourceEligibility = &connector.ConnectorPushRecoverySourceEligibilityContract{
		RequiredFalseBooleanFields: []string{"Subject"},
	}
	invalid.MutationDefinitions = append(invalid.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &invalidAuthorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &invalidAuthority},
	)
	if _, err := CompileCatalogSnapshot(invalid); err == nil || !strings.Contains(err.Error(), "not boolean") {
		t.Fatalf("non-boolean recovery field error=%v", err)
	}
}

func TestConnectorFinalizationRejectsAliasedRemoteIdentityFieldsAtPublication(t *testing.T) {
	input := mutationCompileInput()
	authorship, authority := connectorInboundFinalizationTestAuthority()
	authority.Contract.RemoteIDProvenanceField = "subject"
	input.MutationDefinitions = append(input.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &authorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &authority},
	)
	if _, err := CompileCatalogSnapshot(input); err == nil || !strings.Contains(err.Error(), "must be distinct") {
		t.Fatalf("aliased remote identity fields error=%v", err)
	}
}

func connectorInboundFinalizationTestAuthority() (ConnectorInboundAuthorshipBinding, ConnectorInboundFinalizationAuthority) {
	contract := connector.ConnectorPushDurableMetadataContract{
		ConnectorOwnerKey:  "example-connector-pack",
		ActionSlotID:       "remote-contact-upsert",
		ActionType:         constants.ActionTypePushToConnector,
		FlowDeveloperName:  "WorkItem_To_Remote",
		FlowActionName:     "PushWorkItem",
		SourceObject:       "WorkItem",
		DestinationSlotID:  "remote-contact",
		ProfileOwnerPack:   "example-profile-pack",
		ProfileName:        "primary",
		TargetObject:       "contacts",
		CreatePathTemplate: "/crm/v3/objects/contacts",
		UpdatePathTemplate: "/crm/v3/objects/contacts/{remoteId}",
		ResponseIDPath:     "id",
		RemoteIDField:      "Subject",
	}
	authorship := ConnectorInboundAuthorshipBinding{
		ConnectorOwnerKey: contract.ConnectorOwnerKey,
		Generation:        "pack-installation-17",
		ActionSlotID:      contract.ActionSlotID,
		ActionType:        contract.ActionType,
		SourceObject:      contract.SourceObject,
		DestinationSlotID: contract.DestinationSlotID,
		TargetObject:      contract.TargetObject,
		ProfileOwnerPack:  contract.ProfileOwnerPack,
		ProfileName:       contract.ProfileName,
		RemoteIDField:     contract.RemoteIDField,
	}
	return authorship, ConnectorInboundFinalizationAuthority{
		ContractGeneration: authorship.Generation,
		Contract:           contract,
		Profile: ConnectorInboundFinalizationProfile{
			Generation: "profile-installation-9", OwnerPack: contract.ProfileOwnerPack,
			ProfileName: contract.ProfileName, CredentialRef: "primary-credential",
			EndpointClass: "crm.objects.write", MaxAttempts: 3,
			BaseDelayNanos: int64(time.Second), Multiplier: 2,
			MaxDelayNanos: int64(10 * time.Second), RetryableStatusCodes: []int{503, 429},
		},
		OutboundMappings: []connector.ConnectorFieldMappingInput{{
			Source: contract.ConnectorOwnerKey, SourceObject: contract.SourceObject,
			TargetObject: contract.TargetObject, ActionType: contract.ActionType,
			SourceField: "Subject", TargetField: "subject",
			Direction: connector.DirectionOutbound,
		}},
	}
}

func TestRuntimeMutationViewSchemaFollowsProjectedClosure(t *testing.T) {
	withoutConnectorFacts := emptyMutationDefinitionDocument()
	if got := runtimeMutationViewSchemaFor(CatalogSnapshotSchemaV11, withoutConnectorFacts); got != RuntimeMetadataViewSchemaV4 {
		t.Fatalf("Runtime view without projected connector facts schema=%q want=%q", got, RuntimeMetadataViewSchemaV4)
	}

	withConnectorFacts := emptyMutationDefinitionDocument()
	withConnectorFacts.ConnectorInboundFieldMappingRules = append(withConnectorFacts.ConnectorInboundFieldMappingRules, ConnectorInboundFieldMappingRule{
		Source: "hubspot", SourceObject: "contacts", TargetObject: "WorkItem",
		ActionType: "upsert", SourceField: "firstname", TargetField: "Subject",
		Direction: connector.DirectionInbound,
	})
	if got := runtimeMutationViewSchemaFor(CatalogSnapshotSchemaV11, withConnectorFacts); got != RuntimeMetadataViewSchemaV13 {
		t.Fatalf("Runtime view with projected connector facts schema=%q want=%q", got, RuntimeMetadataViewSchemaV13)
	}

	if got := runtimeMutationViewSchemaFor(CatalogSnapshotSchemaV10, withConnectorFacts); got != RuntimeMetadataViewSchemaV4 {
		t.Fatalf("pre-v11 Catalog promoted connector facts to schema=%q", got)
	}
	if got := runtimeMutationViewSchemaFor(CatalogSnapshotSchemaV12, withConnectorFacts); got != RuntimeMetadataViewSchemaV13 {
		t.Fatalf("v12 Catalog connector-only projection schema=%q want=%q", got, RuntimeMetadataViewSchemaV13)
	}
	_, authority := connectorInboundFinalizationTestAuthority()
	withConnectorFacts.ConnectorInboundFinalization = append(withConnectorFacts.ConnectorInboundFinalization, authority)
	if got := runtimeMutationViewSchemaFor(CatalogSnapshotSchemaV12, withConnectorFacts); got != RuntimeMetadataViewSchemaV14 {
		t.Fatalf("v12 Catalog finalization projection schema=%q want=%q", got, RuntimeMetadataViewSchemaV14)
	}
}

func TestConnectorRuntimeFactsFailClosedUnderOlderSchemas(t *testing.T) {
	input := mutationCompileInput()
	input.MutationDefinitions = append(input.MutationDefinitions, MutationDefinitionInput{
		CatalogRevision: testRevision,
		ConnectorInboundFieldMappingRule: &ConnectorInboundFieldMappingRule{
			Source: "hubspot", SourceObject: "contacts", TargetObject: "WorkItem",
			ActionType: "upsert", SourceField: "firstname", TargetField: "Subject",
			Direction: connector.DirectionInbound,
		},
	})
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	downgradedCatalog := snapshot.doc
	downgradedCatalog.Schema = CatalogSnapshotSchemaV9
	data, err := marshalCanonical(downgradedCatalog)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenCatalogSnapshot(makeArtifact(ArtifactCatalogSnapshot, data)); err == nil {
		t.Fatal("older Catalog schema accepted connector execution facts")
	}

	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA,
		CapabilityID:      testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	view := mustOpenView(t, viewArtifact)
	downgradedView := view.doc
	downgradedView.Schema = RuntimeMetadataViewSchemaV4
	data, err = marshalCanonical(downgradedView)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, data)); err == nil {
		t.Fatal("older Runtime schema accepted connector execution facts")
	}
}

func TestConnectorFinalizationAuthorityFailsClosedUnderOlderSchemas(t *testing.T) {
	input := mutationCompileInput()
	authorship, authority := connectorInboundFinalizationTestAuthority()
	input.MutationDefinitions = append(input.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &authorship},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundFinalization: &authority},
	)
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	downgradedCatalog := snapshot.doc
	downgradedCatalog.Schema = CatalogSnapshotSchemaV11
	data, err := marshalCanonical(downgradedCatalog)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenCatalogSnapshot(makeArtifact(ArtifactCatalogSnapshot, data)); err == nil {
		t.Fatal("v11 Catalog schema accepted finalization authority")
	}

	effective, err := BuildEffectiveMetadata(input)
	if err != nil {
		t.Fatal(err)
	}
	var downgradedEffective catalogDocument
	if err := decodeStrict(effective.Bytes(), &downgradedEffective); err != nil {
		t.Fatal(err)
	}
	downgradedEffective.Schema = EffectiveMetadataSchemaV10
	data, err = marshalCanonical(downgradedEffective)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenEffectiveMetadata(ComputeDigest(data), data); err == nil {
		t.Fatal("v10 effective metadata schema accepted finalization authority")
	}

	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA,
		CapabilityID:      testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	view := mustOpenView(t, viewArtifact)
	downgradedView := view.doc
	downgradedView.Schema = RuntimeMetadataViewSchemaV13
	data, err = marshalCanonical(downgradedView)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, data)); err == nil {
		t.Fatal("v13 Runtime schema accepted finalization authority")
	}
}

func TestConnectorInboundAuthorshipRejectsAmbiguousInstalledSelector(t *testing.T) {
	input := mutationCompileInput()
	base := ConnectorInboundAuthorshipBinding{
		ConnectorOwnerKey: "hubspot-crm-connector", Generation: "pack-installation-17",
		ActionSlotID: "hubspot-contact-upsert", ActionType: constants.ActionTypePushToConnector,
		SourceObject: "WorkItem", DestinationSlotID: "hubspot-contact", TargetObject: "contacts",
		ProfileOwnerPack: "pingcap-hubspot-companion", ProfileName: "hubspot-main", RemoteIDField: "Subject",
	}
	other := base
	other.ActionSlotID = "hubspot-contact-other"
	other.DestinationSlotID = "hubspot-contact-other"
	input.MutationDefinitions = append(input.MutationDefinitions,
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &base},
		MutationDefinitionInput{CatalogRevision: testRevision, ConnectorInboundAuthorship: &other},
	)
	if _, err := CompileCatalogSnapshot(input); err == nil {
		t.Fatal("ambiguous installed authorship selector was accepted")
	}
}

func TestExternalConnectorSourceRequiresExplicitConnectorAuthority(t *testing.T) {
	input := mutationCompileInput()
	input.MutationDefinitions = append(input.MutationDefinitions, MutationDefinitionInput{
		CatalogRevision: testRevision,
		FieldMappingRule: &FieldMappingRule{
			SourceObject: "contacts", TargetObject: "WorkItem", ActionType: "upsert",
			SourceField: "firstname", TargetField: "Subject",
		},
	})
	if _, err := CompileCatalogSnapshot(input); err == nil {
		t.Fatal("external source object without connector authority was accepted")
	}
}

func TestConnectorFieldMappingRejectsInvalidRows(t *testing.T) {
	base := ConnectorInboundFieldMappingRule{
		Source: "hubspot", SourceObject: "contacts", TargetObject: "WorkItem",
		ActionType: "upsert", SourceField: "firstname", TargetField: "Subject",
		Direction: connector.DirectionInbound,
	}
	for name, mutate := range map[string]func(*ConnectorInboundFieldMappingRule){
		"outbound":       func(rule *ConnectorInboundFieldMappingRule) { rule.Direction = connector.DirectionOutbound },
		"missing source": func(rule *ConnectorInboundFieldMappingRule) { rule.Source = "" },
		"missing source field with default": func(rule *ConnectorInboundFieldMappingRule) {
			rule.SourceField = ""
			rule.DefaultValue = "fallback"
		},
		"malformed transform": func(rule *ConnectorInboundFieldMappingRule) {
			rule.TransformExpression = "UPPER("
		},
		"session-global transform": func(rule *ConnectorInboundFieldMappingRule) {
			rule.TransformExpression = "$User.Name"
		},
		"relationship transform": func(rule *ConnectorInboundFieldMappingRule) {
			rule.TransformExpression = "$Record.Account.Name"
		},
		"unsupported function": func(rule *ConnectorInboundFieldMappingRule) {
			rule.TransformExpression = "DOES_NOT_EXIST(firstname)"
		},
	} {
		t.Run(name, func(t *testing.T) {
			input := mutationCompileInput()
			rule := base
			mutate(&rule)
			input.MutationDefinitions = append(input.MutationDefinitions, MutationDefinitionInput{
				CatalogRevision:                  testRevision,
				ConnectorInboundFieldMappingRule: &rule,
			})
			if _, err := CompileCatalogSnapshot(input); err == nil {
				t.Fatal("invalid connector mapping was accepted")
			}
		})
	}
}

func TestConnectorCatalogRejectsPartiallyScopedMutationProjection(t *testing.T) {
	input := mutationCompileInput()
	input.MutationDefinitions = append(input.MutationDefinitions, MutationDefinitionInput{
		CatalogRevision: testRevision,
		ConnectorInboundFieldMappingRule: &ConnectorInboundFieldMappingRule{
			Source: "hubspot", SourceObject: "contacts", TargetObject: "WorkItem",
			ActionType: "upsert", SourceField: "firstname", TargetField: "Subject",
			Direction: connector.DirectionInbound,
		},
	})
	input.Projections[0].ObjectAPINames = []string{"WorkItem"}
	_, err := CompileCatalogSnapshot(input)
	if err == nil || !strings.Contains(err.Error(), "full same-revision schema closure") {
		t.Fatalf("partial record-mutation projection error = %v", err)
	}
}

func TestFieldMappingRuleKeyPreservesLegacyScopeToken(t *testing.T) {
	legacy := FieldMappingRule{SourceObject: "WorkItem", TargetObject: "WorkItem", ActionType: "create", TargetField: "Subject"}
	const legacyKey = "workitem\x00workitem\x00create\x00subject"
	if got := fieldMappingRuleKey(legacy); got != legacyKey {
		t.Fatalf("legacy field-mapping key = %q, want %q", got, legacyKey)
	}
	connectorRule := ConnectorInboundFieldMappingRule{Source: "companion", SourceObject: "contacts", TargetObject: "Lead", ActionType: "syncContact", TargetField: "Email"}
	const connectorKey = "companion\x00contacts\x00lead\x00synccontact\x00email"
	if got := connectorInboundFieldMappingRuleKey(connectorRule); got != connectorKey {
		t.Fatalf("connector field-mapping key = %q, want %q", got, connectorKey)
	}
	connectorRule.PackAuthored = true
	if got := connectorInboundFieldMappingRuleKey(connectorRule); got != connectorKey+"\x00pack-authored" {
		t.Fatalf("pack-authored field-mapping key = %q", got)
	}
}

func TestFlowExecutionProjectionCarriesExactActiveManualClosureWithoutChangingWorkerView(t *testing.T) {
	input := mutationCompileInput()
	input.MutationDefinitions = append(input.MutationDefinitions,
		MutationDefinitionInput{
			CatalogRevision: testRevision,
			FlowDefinition: &models.FlowDefinition{
				Id: "flow-definition-manual", DeveloperName: "ManualAction",
				ProcessType: "AutoLaunchedFlow", ActiveVersionId: "flow-version-manual",
			},
		},
		MutationDefinitionInput{
			CatalogRevision: testRevision,
			FlowVersion: &models.FlowVersion{
				Id: "flow-version-manual", DefinitionId: "flow-definition-manual",
				VersionNumber: 1, Status: models.FlowStatusActive,
				ProcessType: "AutoLaunchedFlow", Source: "standard",
				MetadataRaw: []byte(`{"status":"Active","runAtomically":true,"start":{},"subflows":[{"name":"NormalizeParent","flowName":"NormalizeWorkspace","locationX":0,"locationY":0}]}`),
			},
		},
	)
	input.Projections = append(input.Projections, RuntimeProjectionInput{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: flowsemantic.CapabilityName, Profile: ProjectionRecordMutationV1,
		ObjectAPINames: append([]string(nil), input.Projections[0].ObjectAPINames...),
	})
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))

	mutationArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	mutationView := mustOpenView(t, mutationArtifact)
	if _, ok := mutationView.ActiveFlowVersion("ManualAction"); ok {
		t.Fatal("record-event worker projection gained an unselected manual root")
	}

	flowArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: flowsemantic.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	flowView := mustOpenView(t, flowArtifact)
	manual, ok := flowView.ActiveFlowVersion("ManualAction")
	if !ok || manual.Id != "flow-version-manual" {
		t.Fatalf("manual exact version = %#v, %v", manual, ok)
	}
	child, ok := flowView.ActiveFlowVersion("NormalizeWorkspace")
	if !ok || child.Id != "flow-version-child" {
		t.Fatalf("manual child exact version = %#v, %v", child, ok)
	}
}

func TestRecordMutationProjectionPreservesImmutableAfterCreateWriteProtection(t *testing.T) {
	input := mutationCompileInput()
	for _, object := range input.Objects {
		if object.Metadata == nil || object.Metadata.APIName != "WorkItem" {
			continue
		}
		field := models.FindField(object.Metadata, "Subject")
		if field == nil {
			t.Fatal("mutation input has no WorkItem.Subject field")
		}
		field.WriteProtectionPolicy = constants.FieldWriteProtectionImmutableAfterCreate
	}

	view := mustOpenView(t, mustProject(
		t,
		mustOpenSnapshot(t, mustCompile(t, input)),
		testInstallA,
		testMutationCapability,
	))
	object, ok := view.Object("WorkItem")
	if !ok {
		t.Fatal("mutation projection omitted WorkItem")
	}
	field := models.FindField(object, "Subject")
	if field == nil || field.WriteProtectionPolicy != constants.FieldWriteProtectionImmutableAfterCreate {
		t.Fatalf("mutation projection lost immutable-after-create policy: %#v", field)
	}
	if !field.Createable || field.Updateable {
		t.Fatalf("immutable-after-create projected capabilities = createable=%v updateable=%v, want true/false", field.Createable, field.Updateable)
	}
}

func TestRecordMutationProjectionFailsClosedForUnknownWriteProtectionPolicy(t *testing.T) {
	input := mutationCompileInput()
	for _, object := range input.Objects {
		if object.Metadata == nil || object.Metadata.APIName != "WorkItem" {
			continue
		}
		field := models.FindField(object.Metadata, "Subject")
		if field == nil {
			t.Fatal("mutation input has no WorkItem.Subject field")
		}
		field.WriteProtectionPolicy = "future_unknown_policy"
	}

	view := mustOpenView(t, mustProject(
		t,
		mustOpenSnapshot(t, mustCompile(t, input)),
		testInstallA,
		testMutationCapability,
	))
	object, ok := view.Object("WorkItem")
	if !ok {
		t.Fatal("mutation projection omitted WorkItem")
	}
	field := models.FindField(object, "Subject")
	if field == nil {
		t.Fatal("mutation projection omitted WorkItem.Subject")
	}
	if field.Createable || field.Updateable {
		t.Fatalf("unknown write-protection policy projected writable capabilities: createable=%v updateable=%v", field.Createable, field.Updateable)
	}
}

func TestRuntimeMutationViewRejectsAmbiguousKeyPrefixOwnership(t *testing.T) {
	artifact := mustProject(
		t,
		mustOpenSnapshot(t, mustCompile(t, mutationCompileInput())),
		testInstallA,
		testMutationCapability,
	)
	var doc runtimeViewDocument
	if err := json.Unmarshal(artifact.Bytes(), &doc); err != nil {
		t.Fatal(err)
	}
	for i := range doc.Objects {
		if doc.Objects[i].APIName == "Workspace" {
			doc.Objects[i].Mutation.KeyPrefix = stringPointer("0WI")
		}
	}
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) ||
		!strings.Contains(err.Error(), `duplicate key prefix "0WI"`) {
		t.Fatalf("open ambiguous mutation view error = %v, want corrupt duplicate identity", err)
	}
}

func TestRecordMutationProjectionAppliesCompleteEffectiveBehaviorAcrossInstallations(t *testing.T) {
	const otherComponent = "definition.other-mutation"
	input := mutationCompileInput()
	input.Definitions = append(input.Definitions, DefinitionInput{
		CatalogRevision: testRevision, ComponentID: otherComponent, DefinitionVersion: "1",
		ContentDigest: Digest("sha256:" + string(bytes.Repeat([]byte{'4'}, 64))),
	})
	input.Projections = append(input.Projections, RuntimeProjectionInput{
		CatalogRevision: testRevision, AppInstallationID: testInstallB,
		CapabilityID: testMutationCapability, Profile: ProjectionRecordMutationV1,
		ObjectAPINames:         append([]string(nil), input.Projections[0].ObjectAPINames...),
		DefinitionComponentIDs: []string{otherComponent},
	})
	input.MutationDefinitions = append(input.MutationDefinitions, MutationDefinitionInput{
		CatalogRevision: testRevision,
		ValidationRule: &models.ValidationRule{
			ID: "validation-other-work-item", ObjectAPIName: "WorkItem", Name: "OtherRule",
			Active: true, Condition: "Subject = 'other'", ErrorMessage: "Other component rule",
		},
	})
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))

	firstArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	secondArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallB, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	first := mustOpenView(t, firstArtifact)
	second := mustOpenView(t, secondArtifact)
	firstRules := first.ValidationRulesForObject("WorkItem")
	secondRules := second.ValidationRulesForObject("WorkItem")
	for name, value := range map[string]struct {
		view  *RuntimeMetadataView
		rules []*models.ValidationRule
	}{"first": {first, firstRules}, "second": {second, secondRules}} {
		rules := value.rules
		if len(rules) != 2 || rules[0].ID != "validation-other-work-item" || rules[1].ID != "validation-work-item" {
			t.Fatalf("%s installation omitted effective behavior for its object scope: %#v", name, rules)
		}
		object, ok := value.view.Object("WorkItem")
		subject := models.FindField(object, "Subject")
		if !ok || subject == nil || !subject.DefaultedOnCreate || !value.view.HasTriggerSubscription("0WI") {
			t.Fatalf("%s installation omitted effective derived or Flow behavior: object=%#v trigger=%v", name, object, value.view.HasTriggerSubscription("0WI"))
		}
	}
}

func TestRecordMutationProjectionRequiresCompleteFlowFieldMappingClosure(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		value := &input.MutationDefinitions[i]
		if value.FlowVersion != nil && value.FlowVersion.Id == "flow-version-1" {
			value.FlowVersion.MetadataRaw = flowWithRecordCreateFieldMappingMetadata()
		}
	}
	complete := append([]MutationDefinitionInput(nil), input.MutationDefinitions...)
	input.MutationDefinitions = input.MutationDefinitions[:0]
	for _, value := range complete {
		if value.FieldMappingRule == nil {
			input.MutationDefinitions = append(input.MutationDefinitions, value)
		}
	}

	_, err := CompileCatalogSnapshot(input)
	var conflictErr *ConflictError
	if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidProjection ||
		!strings.Contains(err.Error(), "references absent mapping group") {
		t.Fatalf("missing effective field-mapping closure error = %v, want invalid projection", err)
	}

	input.MutationDefinitions = complete
	if _, err := CompileCatalogSnapshot(input); err != nil {
		t.Fatalf("compile complete field-mapping closure: %v", err)
	}
}

func TestRecordMutationProjectionRejectsConnectorRuleAsLegacyFlowDependency(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		value := &input.MutationDefinitions[i]
		if value.FlowVersion != nil && value.FlowVersion.Id == "flow-version-1" {
			value.FlowVersion.MetadataRaw = flowWithRecordCreateFieldMappingMetadata()
		}
	}
	definitions := make([]MutationDefinitionInput, 0, len(input.MutationDefinitions))
	for _, value := range input.MutationDefinitions {
		if value.FieldMappingRule == nil {
			definitions = append(definitions, value)
		}
	}
	input.MutationDefinitions = append(definitions, MutationDefinitionInput{
		CatalogRevision: testRevision,
		ConnectorInboundFieldMappingRule: &ConnectorInboundFieldMappingRule{
			Source: "connector-owner", SourceObject: "Workspace", TargetObject: "WorkItem",
			ActionType: "insertInheritFromLookup", SourceField: "Name", TargetField: "Subject",
			Direction: connector.DirectionInbound,
		},
	})

	_, err := CompileCatalogSnapshot(input)
	var conflictErr *ConflictError
	if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidProjection ||
		!strings.Contains(err.Error(), "references absent mapping group") {
		t.Fatalf("connector-only legacy Flow dependency error = %v, want invalid projection", err)
	}
}

func TestRecordMutationProjectionDoesNotManufactureAppComponentLineage(t *testing.T) {
	input := mutationCompileInput()
	input.Projections[0].DefinitionComponentIDs = nil
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	view := mustOpenView(t, artifact)
	if _, _, ok := view.Definition(testMutationComponent); ok || len(view.ValidationRulesForObject("WorkItem")) != 1 {
		t.Fatal("effective mutation facts were confused with AppRelease component lineage")
	}
}

func TestLegacyV3MutationSnapshotMigratesToCompleteV4Behavior(t *testing.T) {
	legacy := legacyV3MutationSnapshot(t)
	snapshot, err := OpenCatalogSnapshot(legacy)
	if err != nil {
		t.Fatalf("open legacy v3 Catalog snapshot: %v", err)
	}
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project corrected view from legacy v3 snapshot: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if rules := view.ValidationRulesForObject("WorkItem"); len(rules) != 1 || rules[0].ID != "validation-work-item" {
		t.Fatalf("migrated effective behavior = %#v", rules)
	}
	if _, _, ok := view.Definition(testMutationComponent); ok {
		t.Fatal("legacy mutation component association escaped into the migrated Runtime view")
	}
	var viewDoc runtimeViewDocument
	if err := decodeStrict(viewArtifact.Bytes(), &viewDoc); err != nil {
		t.Fatal(err)
	}
	if viewDoc.Schema != RuntimeMetadataViewSchemaV4 || len(viewDoc.Mutation.LegacyScopes) != 0 {
		t.Fatalf("migrated Runtime view schema/scopes = %q/%#v", viewDoc.Schema, viewDoc.Mutation.LegacyScopes)
	}

	// A v3 mutation view cannot be made complete from its own already-filtered
	// bytes. It must fail closed and be regenerated from the v3 Catalog source.
	viewDoc.Schema = RuntimeMetadataViewSchemaV3
	legacyViewBytes, err := marshalCanonical(viewDoc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, legacyViewBytes)); !errors.Is(err, ErrUnsupportedSchema) {
		t.Fatalf("legacy v3 mutation Runtime view error = %v, want unsupported schema", err)
	}

	var effectiveDoc catalogDocument
	if err := decodeStrict(legacy.Bytes(), &effectiveDoc); err != nil {
		t.Fatal(err)
	}
	effectiveDoc.Schema = EffectiveMetadataSchemaV2
	effectiveBytes, err := marshalCanonical(effectiveDoc)
	if err != nil {
		t.Fatal(err)
	}
	effective, err := OpenEffectiveMetadata(ComputeDigest(effectiveBytes), effectiveBytes)
	if err != nil {
		t.Fatalf("open legacy v2 effective metadata: %v", err)
	}
	migrated, err := CompileEffectiveMetadata(effective)
	if err != nil {
		t.Fatalf("compile legacy effective metadata to v4: %v", err)
	}
	var migratedDoc catalogDocument
	if err := decodeStrict(migrated.Bytes(), &migratedDoc); err != nil {
		t.Fatal(err)
	}
	if migratedDoc.Schema != CatalogSnapshotSchemaV4 || len(migratedDoc.Mutation.LegacyScopes) != 0 ||
		len(migratedDoc.Projections[0].DefinitionComponentIDs) != 0 {
		t.Fatalf("migrated Catalog schema/scopes/definitions = %q/%#v/%#v", migratedDoc.Schema,
			migratedDoc.Mutation.LegacyScopes, migratedDoc.Projections[0].DefinitionComponentIDs)
	}
}

func TestV4MutationEffectiveMetadataRoundTripIsExact(t *testing.T) {
	input := mutationCompileInput()
	direct := mustCompile(t, input)
	effective, err := BuildEffectiveMetadata(input)
	if err != nil {
		t.Fatal(err)
	}
	var effectiveDoc catalogDocument
	if err := decodeStrict(effective.Bytes(), &effectiveDoc); err != nil {
		t.Fatal(err)
	}
	if effectiveDoc.Schema != EffectiveMetadataSchemaV3 || len(effectiveDoc.Mutation.LegacyScopes) != 0 {
		t.Fatalf("current effective mutation schema/scopes = %q/%#v", effectiveDoc.Schema, effectiveDoc.Mutation.LegacyScopes)
	}
	opened, err := OpenEffectiveMetadata(effective.Digest(), effective.Bytes())
	if err != nil {
		t.Fatal(err)
	}
	compiled, err := CompileEffectiveMetadata(opened)
	if err != nil {
		t.Fatal(err)
	}
	if compiled.Digest() != direct.Digest() || !bytes.Equal(compiled.Bytes(), direct.Bytes()) {
		t.Fatal("current v4 mutation effective metadata did not compile to the exact direct snapshot")
	}
}

func legacyV3MutationSnapshot(t *testing.T) Artifact {
	t.Helper()
	input := mutationCompileInput()
	input.Definitions = append(input.Definitions, DefinitionInput{
		CatalogRevision: testRevision, ComponentID: testMutationComponent, DefinitionVersion: "1",
		ContentDigest: Digest("sha256:" + string(bytes.Repeat([]byte{'3'}, 64))),
	})
	input.Projections[0].DefinitionComponentIDs = []string{testMutationComponent}
	current := mustCompile(t, input)
	var doc catalogDocument
	if err := decodeStrict(current.Bytes(), &doc); err != nil {
		t.Fatal(err)
	}
	doc.Schema = CatalogSnapshotSchemaV3
	appendScope := func(kind, factKey string) {
		doc.Mutation.LegacyScopes = append(doc.Mutation.LegacyScopes, legacyMutationDefinitionScope{
			ComponentID: testMutationComponent, Kind: kind, FactKey: factKey,
		})
	}
	for _, value := range doc.Mutation.ValidationRules {
		appendScope(legacyMutationScopeValidationRule, value.ID)
	}
	for _, value := range doc.Mutation.RecordTypes {
		appendScope(legacyMutationScopeRecordType, value.ID)
	}
	for _, value := range doc.Mutation.TriggerSubscriptions {
		appendScope(legacyMutationScopeTriggerSubscription, value.ID)
	}
	for _, value := range doc.Mutation.FlowDefinitions {
		appendScope(legacyMutationScopeFlowDefinition, value.ID)
	}
	for _, value := range doc.Mutation.FlowVersions {
		appendScope(legacyMutationScopeFlowVersion, value.ID)
	}
	for _, value := range doc.Mutation.AutoNumberRules {
		appendScope(legacyMutationScopeAutoNumberRule, value.FieldID)
	}
	for _, value := range doc.Mutation.FieldMappingRules {
		appendScope(legacyMutationScopeFieldMappingRule, fieldMappingRuleKey(value))
	}
	for _, value := range doc.Mutation.CreateSourceGuards {
		appendScope(legacyMutationScopeCreateSourceGuard, createSourceGuardRuleKey(value))
	}
	for _, value := range doc.Mutation.OAuthPublicScopes {
		appendScope(legacyMutationScopeOAuthPublicScope, oauthPublicScopeRuleKey(value))
	}
	sort.Slice(doc.Mutation.LegacyScopes, func(i, j int) bool {
		left, right := doc.Mutation.LegacyScopes[i], doc.Mutation.LegacyScopes[j]
		return legacyMutationScopeToken(left.Kind, left.FactKey) < legacyMutationScopeToken(right.Kind, right.FactKey)
	})
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	return makeArtifact(ArtifactCatalogSnapshot, encoded)
}

func TestRecordMutationProjectionRequiresCompleteEffectiveFlowClosure(t *testing.T) {
	input := mutationCompileInput()
	complete := append([]MutationDefinitionInput(nil), input.MutationDefinitions...)
	input.MutationDefinitions = input.MutationDefinitions[:0]
	for _, value := range complete {
		if value.FlowDefinition != nil && value.FlowDefinition.Id == "flow-definition-child" ||
			value.FlowVersion != nil && value.FlowVersion.Id == "flow-version-child" {
			continue
		}
		input.MutationDefinitions = append(input.MutationDefinitions, value)
	}

	_, err := CompileCatalogSnapshot(input)
	var conflictErr *ConflictError
	if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidDefinition ||
		!strings.Contains(err.Error(), `absent active child flow "NormalizeWorkspace"`) {
		t.Fatalf("incomplete effective flow closure error = %v, want invalid definition", err)
	}

	input.MutationDefinitions = complete
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project complete effective flow closure: %v", err)
	}
	view := mustOpenView(t, artifact)
	if version, ok := view.ActiveFlowVersion("NormalizeWorkspace"); !ok || version.Id != "flow-version-child" {
		t.Fatalf("effective child flow closure = %#v, %v", version, ok)
	}
}

func TestV4ReadViewRejectsMutationClosure(t *testing.T) {
	input := mutationCompileInput()
	input.Projections = append(input.Projections, RuntimeProjectionInput{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: testCapability, Profile: ProjectionSelectedRootReadV1,
		ObjectAPINames: []string{"Workspace", "WorkItem"},
	})
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	var doc runtimeViewDocument
	if err := decodeStrict(artifact.Bytes(), &doc); err != nil {
		t.Fatal(err)
	}
	if doc.Schema != RuntimeMetadataViewSchemaV4 || doc.Profile != ProjectionSelectedRootReadV1 || doc.Mutation != nil {
		t.Fatalf("read fixture shape = schema %q profile %q mutation %#v", doc.Schema, doc.Profile, doc.Mutation)
	}
	doc.Mutation = emptyMutationDefinitionDocument()
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("v4 read view with mutation closure error = %v, want corruption", err)
	}
}

func TestV3ReadViewRejectsFieldMutationMetadata(t *testing.T) {
	input := mutationCompileInput()
	input.Projections = append(input.Projections, RuntimeProjectionInput{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: testCapability, Profile: ProjectionSelectedRootReadV1,
		ObjectAPINames: []string{"Workspace", "WorkItem"},
	})
	artifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, mustCompile(t, input)), ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	var doc runtimeViewDocument
	if err := decodeStrict(artifact.Bytes(), &doc); err != nil {
		t.Fatal(err)
	}
	doc.Objects[0].Fields[0].Mutation = &runtimeMutationFieldRecord{Createable: true}
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("v3 read view with field mutation metadata error = %v, want corruption", err)
	}
}

func TestV3ReadViewRejectsRelationshipMutationMetadata(t *testing.T) {
	input := mutationCompileInput()
	input.Projections = append(input.Projections, RuntimeProjectionInput{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: testCapability, Profile: ProjectionSelectedRootReadV1,
		ObjectAPINames: []string{"Workspace", "WorkItem"},
	})
	artifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, mustCompile(t, input)), ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	var doc runtimeViewDocument
	if err := decodeStrict(artifact.Bytes(), &doc); err != nil {
		t.Fatal(err)
	}
	if len(doc.Relationships) == 0 {
		t.Fatal("read fixture has no relationship to attack")
	}
	doc.Relationships[0].Mutation = &runtimeMutationRelationshipRecord{CascadeDelete: true}
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("v3 read view with relationship mutation metadata error = %v, want corruption", err)
	}
}

func TestRecordMutationProjectionRejectsOAuthPolicyWithoutOwningObject(t *testing.T) {
	input := mutationCompileInput()
	filteredObjects := input.Objects[:0]
	for _, object := range input.Objects {
		if object.Metadata.APIName != testOAuthPolicyObject {
			filteredObjects = append(filteredObjects, object)
		}
	}
	input.Objects = filteredObjects
	filteredNames := input.Projections[0].ObjectAPINames[:0]
	for _, name := range input.Projections[0].ObjectAPINames {
		if name != testOAuthPolicyObject {
			filteredNames = append(filteredNames, name)
		}
	}
	input.Projections[0].ObjectAPINames = filteredNames
	_, err := CompileCatalogSnapshot(input)
	var conflictErr *ConflictError
	if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidDefinition ||
		!strings.Contains(err.Error(), "object \"CredentialClient\" is absent") {
		t.Fatalf("orphan OAuth policy error = %v, want invalid definition", err)
	}
}

func TestRecordMutationProjectionAllowsExplicitEmptyDefinitionClosure(t *testing.T) {
	input := mutationCompileInput()
	input.MutationDefinitions = nil
	snapshotArtifact, err := CompileCatalogSnapshot(input)
	if err != nil {
		t.Fatalf("compile mutation snapshot with empty definition closure: %v", err)
	}
	snapshot, err := OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project mutation view with empty definition closure: %v", err)
	}
	view, err := OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	if view.Profile() != ProjectionRecordMutationV1 {
		t.Fatalf("profile = %q", view.Profile())
	}
}

func TestRecordMutationProjectionFailsClosedWhenSynchronousSubflowIsAbsent(t *testing.T) {
	input := mutationCompileInput()
	filtered := input.MutationDefinitions[:0]
	for _, value := range input.MutationDefinitions {
		if value.FlowDefinition != nil && value.FlowDefinition.Id == "flow-definition-child" {
			continue
		}
		if value.FlowVersion != nil && value.FlowVersion.Id == "flow-version-child" {
			continue
		}
		filtered = append(filtered, value)
	}
	input.MutationDefinitions = filtered
	_, err := CompileCatalogSnapshot(input)
	var conflictErr *ConflictError
	if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidDefinition {
		t.Fatalf("missing subflow error = %v, want invalid definition", err)
	}
}

func TestRecordMutationProjectionRejectsUnownedD2CActionCall(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		version := input.MutationDefinitions[i].FlowVersion
		if version == nil || version.Id != "flow-version-1" {
			continue
		}
		version.MetadataRaw = []byte(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave"},"actionCalls":[{"name":"NotifyExternal","actionName":"notify","actionType":"callWebhook","locationX":0,"locationY":0}]}`)
	}
	_, err := CompileCatalogSnapshot(input)
	var conflictErr *ConflictError
	if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidProjection ||
		!strings.Contains(err.Error(), "outside the D2B Runtime mutation worker closure") {
		t.Fatalf("unowned D2C action error = %v, want invalid projection", err)
	}
}

func TestRecordMutationProjectionRejectsUnownedExecutionShapes(t *testing.T) {
	tests := map[string]string{
		"wait":                    `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","connector":{"targetReference":"Pause"}},"waits":[{"name":"Pause","waitEvents":[]}]}`,
		"scheduled path":          `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"name":"Later","connector":{"targetReference":"UpdateLater"}}]}}`,
		"scheduled start":         `{"status":"Active","start":{"schedule":{"frequency":"Daily"},"connector":{"targetReference":"Run"}}}`,
		"connector source":        `{"status":"Active","start":{"connectorSource":{"ownerPack":"pack","profileName":"remote","objectKey":"issues","itemsPath":"$.items","itemIdPath":"$.id"},"connector":{"targetReference":"Run"}}}`,
		"connector field mapping": `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave"},"recordCreates":[{"name":"CreateMapped","object":"WorkItem","connectorFieldMapping":{"source":"pack","sourceObject":"Workspace","targetObject":"WorkItem","actionType":"create","direction":"inbound","sourceRecord":{"elementReference":"$Record"}}}]}`,
		"screen":                  `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","connector":{"targetReference":"Prompt"}},"screens":[{"name":"Prompt","fields":[]}]}`,
	}
	for name, raw := range tests {
		t.Run(name, func(t *testing.T) {
			input := mutationCompileInput()
			for i := range input.MutationDefinitions {
				version := input.MutationDefinitions[i].FlowVersion
				if version != nil && version.Id == "flow-version-1" {
					version.MetadataRaw = []byte(raw)
				}
			}
			_, err := CompileCatalogSnapshot(input)
			var conflictErr *ConflictError
			if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidProjection ||
				!strings.Contains(err.Error(), "outside the D2B Runtime mutation worker closure") {
				t.Fatalf("deferred execution error = %v, want invalid projection", err)
			}
		})
	}
}

func TestRecordMutationProjectionAcceptsSingleAsyncAfterCommitEntry(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		subscription := input.MutationDefinitions[i].TriggerSubscription
		if subscription != nil && subscription.FlowVersionID == "flow-version-1" {
			subscription.EventType = "RecordAfterSave"
		}
		definition := input.MutationDefinitions[i].FlowDefinition
		if definition != nil && definition.Id == "flow-definition-1" {
			definition.TriggerType = "RecordAfterSave"
		}
		version := input.MutationDefinitions[i].FlowVersion
		if version == nil || version.Id != "flow-version-1" {
			continue
		}
		version.MetadataRaw = []byte(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"UpdateLater"}}]},"recordUpdates":[{"name":"UpdateLater","inputReference":"$Record"}]}`)
	}
	if _, err := CompileCatalogSnapshot(input); err != nil {
		t.Fatalf("single AsyncAfterCommit entry rejected: %v", err)
	}
}

func TestRuntimeMutationFlowRejectsScheduledPushBeforeDurableDispatch(t *testing.T) {
	_, err := runtimeMutationFlowChildNames(runtimeFlowVersionRecord{
		Metadata: []byte(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"Push"}}]},"actionCalls":[{"name":"Push","actionType":"PushToConnector"}]}`),
	}, true, func(*models.FlowActionCall) bool { return true })
	if err == nil || !strings.Contains(err.Error(), "outside the D2B Runtime mutation worker closure") {
		t.Fatalf("scheduled PushToConnector error = %v, want activation rejection", err)
	}
}

func TestRuntimeMutationFlowRejectsAsyncIdentityActionWithoutPackOwnedStandardVersion(t *testing.T) {
	_, err := runtimeMutationFlowChildNames(runtimeFlowVersionRecord{
		Source:   "standard",
		Metadata: []byte(`{"status":"Active","runAtomically":true,"start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"Resolve"}}]},"actionCalls":[{"name":"Resolve","actionType":"resolveConnectorRemoteIdentity"}]}`),
	}, true, nil)
	if err == nil || !strings.Contains(err.Error(), "outside the D2B Runtime mutation worker closure") {
		t.Fatalf("unowned AsyncAfterCommit identity action error = %v, want activation rejection", err)
	}
}

func TestRecordMutationProjectionAcceptsAtomicAsyncAfterCommitIdentityActions(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		subscription := input.MutationDefinitions[i].TriggerSubscription
		if subscription != nil && subscription.FlowVersionID == "flow-version-1" {
			subscription.EventType = "RecordAfterSave"
		}
		definition := input.MutationDefinitions[i].FlowDefinition
		if definition != nil && definition.Id == "flow-definition-1" {
			definition.TriggerType = "RecordAfterSave"
		}
		version := input.MutationDefinitions[i].FlowVersion
		if version != nil && version.Id == "flow-version-1" {
			version.DefinitionOwnerPack = "example-connector-pack"
			version.MetadataRaw = []byte(`{"status":"Active","runAtomically":true,"start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"Resolve"}}]},"actionCalls":[{"name":"Resolve","actionType":"resolveConnectorRemoteIdentity","connector":{"targetReference":"Transfer"}},{"name":"Transfer","actionType":"transferConnectorRemoteIdentity"}]}`)
		}
	}
	if _, err := CompileCatalogSnapshot(input); err != nil {
		t.Fatalf("atomic AsyncAfterCommit identity actions rejected: %v", err)
	}
}

func TestRecordMutationProjectionAcceptsOnAllChangesAsyncAfterCommitAlias(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		subscription := input.MutationDefinitions[i].TriggerSubscription
		if subscription != nil && subscription.FlowVersionID == "flow-version-1" {
			subscription.EventType = "RecordAfterSave"
		}
		definition := input.MutationDefinitions[i].FlowDefinition
		if definition != nil && definition.Id == "flow-definition-1" {
			definition.TriggerType = "onAllChanges"
		}
		version := input.MutationDefinitions[i].FlowVersion
		if version != nil && version.Id == "flow-version-1" {
			version.MetadataRaw = []byte(`{"status":"Active","start":{"object":"WorkItem","triggerType":"onAllChanges","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"UpdateLater"}}]},"recordUpdates":[{"name":"UpdateLater","inputReference":"$Record"}]}`)
		}
	}
	if _, err := CompileCatalogSnapshot(input); err != nil {
		t.Fatalf("onAllChanges AsyncAfterCommit entry rejected: %v", err)
	}
}

func TestRecordMutationProjectionAcceptsOneLaunchTimedEntry(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		subscription := input.MutationDefinitions[i].TriggerSubscription
		if subscription != nil && subscription.FlowVersionID == "flow-version-1" {
			subscription.EventType = "RecordAfterSave"
		}
		definition := input.MutationDefinitions[i].FlowDefinition
		if definition != nil && definition.Id == "flow-definition-1" {
			definition.TriggerType = "RecordAfterSave"
		}
		version := input.MutationDefinitions[i].FlowVersion
		if version != nil && version.Id == "flow-version-1" {
			version.MetadataRaw = []byte(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","connector":{"targetReference":"RunNow"},"scheduledPaths":[{"name":"OneMinuteLater","timeSource":"RecordTriggerEvent","offsetNumber":1,"offsetUnit":"Minutes","connector":{"targetReference":"RunLater"}}]},"assignments":[{"name":"RunNow"},{"name":"RunLater"}]}`)
		}
	}
	if _, err := CompileCatalogSnapshot(input); err != nil {
		t.Fatalf("launch timed entry rejected: %v", err)
	}
}

func TestRecordMutationProjectionRejectsMixedOrMultipleAsyncAfterCommitEntries(t *testing.T) {
	tests := map[string]string{
		"before-save trigger":          `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordBeforeSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"RunLater"}}]}}`,
		"mixed immediate and async":    `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","connector":{"targetReference":"RunNow"},"scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"RunLater"}}]}}`,
		"mixed top-level and async":    `{"status":"Active","startElementReference":"RunNow","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"RunLater"}}]}}`,
		"multiple async":               `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"First"}},{"pathType":"AsyncAfterCommit","connector":{"targetReference":"Second"}}]}}`,
		"missing async connector":      `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit"}]}}`,
		"blank async connector target": `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"  "}}]}}`,
		"timed path mislabeled async":  `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","timeSource":"RecordTriggerEvent","offsetNumber":1,"offsetUnit":"Minutes","connector":{"targetReference":"RunLater"}}]}}`,
		"mixed top-level and timed":    `{"status":"Active","startElementReference":"RunLater","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"name":"Later","timeSource":"RecordTriggerEvent","offsetNumber":1,"offsetUnit":"Minutes","connector":{"targetReference":"RunLater"}}]}}`,
		"timed missing name":           `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"timeSource":"RecordTriggerEvent","offsetNumber":1,"offsetUnit":"Minutes","connector":{"targetReference":"RunLater"}}]}}`,
		"timed unsupported source":     `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"name":"Later","timeSource":"RecordField","offsetNumber":1,"offsetUnit":"Minutes","connector":{"targetReference":"RunLater"}}]}}`,
		"timed nonpositive offset":     `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"name":"Later","timeSource":"RecordTriggerEvent","offsetNumber":0,"offsetUnit":"Minutes","connector":{"targetReference":"RunLater"}}]}}`,
		"multiple timed":               `{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"name":"First","timeSource":"RecordTriggerEvent","offsetNumber":1,"offsetUnit":"Minutes","connector":{"targetReference":"First"}},{"name":"Second","timeSource":"RecordTriggerEvent","offsetNumber":2,"offsetUnit":"Minutes","connector":{"targetReference":"Second"}}]}}`,
	}
	for name, raw := range tests {
		t.Run(name, func(t *testing.T) {
			input := mutationCompileInput()
			for i := range input.MutationDefinitions {
				subscription := input.MutationDefinitions[i].TriggerSubscription
				if subscription != nil && subscription.FlowVersionID == "flow-version-1" {
					subscription.EventType = "RecordAfterSave"
				}
				definition := input.MutationDefinitions[i].FlowDefinition
				if definition != nil && definition.Id == "flow-definition-1" {
					definition.TriggerType = "RecordAfterSave"
				}
				version := input.MutationDefinitions[i].FlowVersion
				if version != nil && version.Id == "flow-version-1" {
					version.MetadataRaw = []byte(raw)
				}
			}
			_, err := CompileCatalogSnapshot(input)
			var conflictErr *ConflictError
			if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidProjection ||
				!strings.Contains(err.Error(), "outside the D2B Runtime mutation worker closure") {
				t.Fatalf("unsupported async entry error = %v, want invalid projection", err)
			}
		})
	}
}

func TestRecordMutationProjectionRejectsAsyncAfterCommitChildFlow(t *testing.T) {
	input := mutationCompileInput()
	for i := range input.MutationDefinitions {
		version := input.MutationDefinitions[i].FlowVersion
		if version != nil && version.Id == "flow-version-child" {
			version.MetadataRaw = []byte(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave","scheduledPaths":[{"pathType":"AsyncAfterCommit","connector":{"targetReference":"UpdateLater"}}]},"recordUpdates":[{"name":"UpdateLater","inputReference":"$Record"}]}`)
		}
	}
	_, err := CompileCatalogSnapshot(input)
	var conflictErr *ConflictError
	if !errors.As(err, &conflictErr) || conflictErr.Kind != ConflictInvalidProjection ||
		!strings.Contains(err.Error(), "outside the D2B Runtime mutation worker closure") {
		t.Fatalf("async child error = %v, want invalid projection", err)
	}
}

func TestRecordMutationViewRejectsValidlyDigestedUnownedIntegrationFlow(t *testing.T) {
	snapshot := mustOpenSnapshot(t, mustCompile(t, mutationCompileInput()))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	var doc runtimeViewDocument
	if err := decodeStrict(artifact.Bytes(), &doc); err != nil {
		t.Fatal(err)
	}
	for i := range doc.Mutation.FlowVersions {
		if doc.Mutation.FlowVersions[i].ID == "flow-version-1" {
			doc.Mutation.FlowVersions[i].Metadata = json.RawMessage(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordAfterSave"},"recordCreates":[{"name":"CreateMapped","object":"WorkItem","connectorFieldMapping":{}}]}`)
		}
	}
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("validly digested Runtime view with Integration-owned Flow error = %v, want corruption", err)
	}
}

func TestRecordMutationViewRejectsValidlyDigestedDuplicateDefinition(t *testing.T) {
	snapshot := mustOpenSnapshot(t, mustCompile(t, mutationCompileInput()))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{AppInstallationID: testInstallA, CapabilityID: testMutationCapability})
	if err != nil {
		t.Fatalf("project mutation view: %v", err)
	}
	var doc runtimeViewDocument
	if err := decodeStrict(artifact.Bytes(), &doc); err != nil {
		t.Fatalf("decode mutation view: %v", err)
	}
	doc.Mutation.ValidationRules = append(doc.Mutation.ValidationRules, doc.Mutation.ValidationRules[0])
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatalf("encode hostile view: %v", err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("duplicate mutation fact error = %v, want corruption", err)
	}
}

func TestRecordMutationViewRejectsValidlyDigestedAutoNumberFieldMismatch(t *testing.T) {
	artifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, mustCompile(t, mutationCompileInput())), ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	var doc runtimeViewDocument
	if err := decodeStrict(artifact.Bytes(), &doc); err != nil {
		t.Fatal(err)
	}
	doc.Mutation.AutoNumberRules[0].FieldID = "field-work-item-id"
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("validly digested Runtime view with mismatched autonumber field id error = %v, want corruption", err)
	}
}

func TestRecordMutationViewRejectsValidlyDigestedMissingFlowFieldMappingClosure(t *testing.T) {
	// Mutate already canonical, correctly digested artifacts so the reader
	// boundary is proved independently from compiler validation.
	tests := map[string]json.RawMessage{
		"record create": flowWithRecordCreateFieldMappingMetadata(),
		"record update": flowWithRecordUpdateFieldMappingMetadata(),
	}
	for name, flowMetadata := range tests {
		t.Run(name, func(t *testing.T) {
			input := mutationCompileInput()
			for i := range input.MutationDefinitions {
				version := input.MutationDefinitions[i].FlowVersion
				if version != nil && version.Id == "flow-version-1" {
					version.MetadataRaw = flowMetadata
				}
			}
			artifact, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, mustCompile(t, input)), ProjectionKey{
				AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
			})
			if err != nil {
				t.Fatal(err)
			}
			var doc runtimeViewDocument
			if err := decodeStrict(artifact.Bytes(), &doc); err != nil {
				t.Fatal(err)
			}
			doc.Mutation.FieldMappingRules = nil
			encoded, err := marshalCanonical(doc)
			if err != nil {
				t.Fatal(err)
			}
			if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
				t.Fatalf("validly digested Runtime view with missing Flow mapping closure error = %v, want corruption", err)
			}
		})
	}
}

func TestRecordMutationViewRejectsRetiredComponentScopeIndex(t *testing.T) {
	snapshot := mustOpenSnapshot(t, mustCompile(t, mutationCompileInput()))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testMutationCapability,
	})
	if err != nil {
		t.Fatalf("project mutation view: %v", err)
	}
	var doc map[string]any
	if err := json.Unmarshal(artifact.Bytes(), &doc); err != nil {
		t.Fatalf("decode mutation view: %v", err)
	}
	mutation, ok := doc["mutation"].(map[string]any)
	if !ok {
		t.Fatal("mutation view has no mutation document")
	}
	mutation["scopes"] = []any{map[string]any{
		"componentId": testMutationComponent,
		"kind":        "validationRule",
		"factKey":     "validation-work-item",
	}}
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatalf("encode hostile view: %v", err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("Runtime view with Catalog scope index error = %v, want corruption", err)
	}
}

func mutationCompileInput() CompileInput {
	input := neutralCompileInput()
	input.Projections[0] = RuntimeProjectionInput{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: testMutationCapability, Profile: ProjectionRecordMutationV1,
		ObjectAPINames: []string{testOAuthPolicyObject, "Unrelated", "Workspace", "WorkItem"},
	}
	input.Objects = append(input.Objects, ObjectInput{CatalogRevision: testRevision, Metadata: &models.ObjectMetadata{
		ID: "obj-credential-client", APIName: testOAuthPolicyObject, Label: "Credential Client", PluralLabel: "Credential Clients",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields:       []models.FieldMetadata{{ID: "field-credential-client-id", APIName: constants.FieldID, Type: constants.FieldTypeString, IsSystem: true}},
	}})
	workItem := input.Objects[0].Metadata
	workItem.KeyPrefix = stringPointer("0WI")
	workItem.DeleteMode = "soft"
	workItem.Fields[1].TrackHistory = true
	workItem.Fields[1].Options = models.PicklistOptions{{
		Value: "Ready", Label: "Ready", Tags: []string{"active"},
		Extensions: map[string]json.RawMessage{"opaque": json.RawMessage(`{"source":"catalog"}`)},
	}}
	workItem.Fields[1].PicklistDependency = map[string][]string{"Open": {"Ready"}}
	filter := "Status = 'Open'"
	precision := 18
	scale := 2
	resultType := constants.FieldTypeDouble
	workItem.Fields[1].RollupConfig = &models.RollupConfig{
		SummaryObject: "Unrelated", SummaryField: "Amount", RelationshipField: "WorkspaceId",
		CalcType: "SUM", Filter: &filter, ResultType: &resultType, ResultPrecision: &precision, ResultScale: &scale,
	}
	workItem.DefaultChildRows = []models.DefaultChildRow{{
		TargetObject: "Unrelated", TargetLookupField: "Id", FieldValues: models.SObject{},
	}}
	validationDescription := "presentation-only"
	input.MutationDefinitions = []MutationDefinitionInput{
		{CatalogRevision: testRevision, ValidationRule: &models.ValidationRule{
			ID: "validation-work-item", ObjectAPIName: "WorkItem", Name: "RequireSubject",
			Description: &validationDescription, Active: true, Condition: "ISBLANK(Subject)", ErrorMessage: "Subject is required",
		}},
		{CatalogRevision: testRevision, RecordType: &models.RecordType{
			ID: "record-type-work-item", ObjectAPIName: "WorkItem", Name: "Internal",
			DeveloperName: "Internal", Label: "Internal", IsActive: true, BusinessProcessID: stringPointer("process-work-item"),
		}},
		{CatalogRevision: testRevision, TriggerSubscription: &TriggerSubscription{
			ID: "trigger-work-item", KeyPrefix: "0WI", EventType: "RecordBeforeSave",
			RecordTriggerType: "Create", FlowDefinitionID: "flow-definition-1", FlowVersionID: "flow-version-1", IsActive: true,
		}},
		{CatalogRevision: testRevision, FlowDefinition: &models.FlowDefinition{
			Id: "flow-definition-1", DeveloperName: "NormalizeWorkItem", ProcessType: "AutoLaunchedFlow",
			TriggerType: "RecordBeforeSave", TriggerObjectOrEvent: "WorkItem", ActiveVersionId: "flow-version-1",
		}},
		{CatalogRevision: testRevision, FlowVersion: &models.FlowVersion{
			Id: "flow-version-1", DefinitionId: "flow-definition-1", VersionNumber: 1,
			Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow", Source: "standard",
			MetadataRaw: []byte(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordBeforeSave"},"subflows":[{"name":"NormalizeParent","flowName":"NormalizeWorkspace","locationX":0,"locationY":0}]}`),
		}},
		{CatalogRevision: testRevision, FlowDefinition: &models.FlowDefinition{
			Id: "flow-definition-child", DeveloperName: "NormalizeWorkspace", ProcessType: "AutoLaunchedFlow",
			ActiveVersionId: "flow-version-child",
		}},
		{CatalogRevision: testRevision, FlowVersion: &models.FlowVersion{
			Id: "flow-version-child", DefinitionId: "flow-definition-child", VersionNumber: 1,
			Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow", Source: "standard",
			MetadataRaw: []byte(`{"status":"Active","start":{}}`),
		}},
		{CatalogRevision: testRevision, AutoNumberRule: &AutoNumberRule{
			FieldID: "field-subject", ObjectAPIName: "WorkItem", FieldAPIName: "Subject",
			Format: "WI-{0000}", Start: 1, Scope: "global", DateSource: "assigned_at", OnChange: "keep_existing",
		}},
		{CatalogRevision: testRevision, FieldMappingRule: &FieldMappingRule{
			SourceObject: "Workspace", TargetObject: "WorkItem", ActionType: "insertInheritFromLookup",
			SourceField: "Name", TargetField: "Subject", SourceLookupField: "WorkspaceId",
		}},
		{CatalogRevision: testRevision, CreateSourceGuard: &CreateFromSourceGuardRule{
			TargetObject: "WorkItem", SourceFKField: "WorkspaceId",
			OwningStandardActionID: "action-create-work-item", ActionName: "CreateWorkItemFromWorkspace",
		}},
		{CatalogRevision: testRevision, OAuthPublicScope: &OAuthPublicScopeRule{ObjectAPIName: testOAuthPolicyObject, Name: "external.records.read"}},
		{CatalogRevision: testRevision, OAuthPublicScope: &OAuthPublicScopeRule{ObjectAPIName: testOAuthPolicyObject, Name: "pack.records.read", OwnerPack: "io.example.pack"}},
	}
	// Keep the ordinary relationship in scope so delete/relationship behavior
	// is derived from the same Catalog revision as the root mutation.
	input.Relationships[1].Metadata.CascadeDelete = true
	input.Objects[2].Metadata.Fields[0].Createable = true
	input.Objects[2].Metadata.Fields[0].Type = constants.FieldTypeString
	return input
}

func flowWithRecordCreateFieldMappingMetadata() json.RawMessage {
	return json.RawMessage(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordBeforeSave"},"recordCreates":[{"name":"CreateMapped","object":"WorkItem","fieldMapping":{"sourceObject":"Workspace","sourceReference":"workspace","actionType":"insertInheritFromLookup"}}],"subflows":[{"name":"NormalizeParent","flowName":"NormalizeWorkspace","locationX":0,"locationY":0}]}`)
}

func flowWithRecordUpdateFieldMappingMetadata() json.RawMessage {
	return json.RawMessage(`{"status":"Active","start":{"object":"WorkItem","triggerType":"RecordBeforeSave"},"recordUpdates":[{"name":"UpdateMapped","object":"WorkItem","fieldMapping":{"sourceObject":"Workspace","sourceReference":"workspace","actionType":"insertInheritFromLookup"}}],"subflows":[{"name":"NormalizeParent","flowName":"NormalizeWorkspace","locationX":0,"locationY":0}]}`)
}