Apply sequence policy across loading tiers

internal/engine/imports.go

@@ -73,16 +73,16 @@ func loadImportedDocumentsTrackedWithVersion(source Source, context *FilesystemC
 	if versionState != nil && versionState.mode != sourceVersionModeDefault {
 		processing = versionState.mainProcessing
 	}
-	return loadImportedDocumentsTrackedWithProcessing(source, context, main, destination, rule, limits, accounting, visited, hostToken, versionState, processing)
+	return loadImportedDocumentsTrackedWithProcessing(source, context, main, destination, rule, limits, accounting, visited, hostToken, versionState, processing, sequencePolicy{})
 }
 
-func loadImportedDocumentsTrackedWithProcessing(source Source, context *FilesystemContext, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, hostToken *namespaceToken, versionState *sourceVersionState, processing *sourceVersionProcessingResult) (*composedDocument, []SourceInfo, []Diagnostic) {
+func loadImportedDocumentsTrackedWithProcessing(source Source, context *FilesystemContext, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, hostToken *namespaceToken, versionState *sourceVersionState, processing *sourceVersionProcessingResult, policy sequencePolicy) (*composedDocument, []SourceInfo, []Diagnostic) {
 	if nativeContext(source, context) {
 		access, basePath, mainPath, diagnostic := nativeFilesystemAccess(source, context)
 		if diagnostic != nil {
 			return nil, sourceInfos(main.source), []Diagnostic{*diagnostic}
 		}
-		return loadImportedWithAccessTrackedAndProcessing(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing)
+		return loadImportedWithAccessTrackedAndProcessingPolicy(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy)
 	}
 	rootPath, basePath, mainPath, diagnostic := importFilesystem(source, context, main)
 	if diagnostic != nil {
@@ -97,7 +97,7 @@ func loadImportedDocumentsTrackedWithProcessing(source Source, context *Filesyst
 	if hostToken != nil {
 		access.token = hostToken
 	}
-	return loadImportedWithAccessTrackedAndProcessing(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing)
+	return loadImportedWithAccessTrackedAndProcessingPolicy(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy)
 }
 
 func loadImportedWithAccess(access *filesystemAccess, basePath, mainPath string, main parsedSource, rule *VersionRule, limits importLimits, accounting *importAccounting) (*composedDocument, []SourceInfo, []Diagnostic) {
@@ -117,6 +117,10 @@ func loadImportedWithAccessTrackedWithVersion(access *filesystemAccess, basePath
 }
 
 func loadImportedWithAccessTrackedAndProcessing(access *filesystemAccess, basePath, mainPath string, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, versionState *sourceVersionState, processing *sourceVersionProcessingResult) (*composedDocument, []SourceInfo, []Diagnostic) {
+	return loadImportedWithAccessTrackedAndProcessingPolicy(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, sequencePolicy{})
+}
+
+func loadImportedWithAccessTrackedAndProcessingPolicy(access *filesystemAccess, basePath, mainPath string, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, versionState *sourceVersionState, processing *sourceVersionProcessingResult, policy sequencePolicy) (*composedDocument, []SourceInfo, []Diagnostic) {
 	if !access.contains(basePath) || (mainPath != "" && !access.contains(mainPath)) {
 		_ = access.close()
 		return nil, sourceInfos(main.source), []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "selected source and import base must stay inside the configuration root", nil)}
@@ -237,7 +241,7 @@ func loadImportedWithAccessTrackedAndProcessing(access *filesystemAccess, basePa
 			return nil, cloneSourceInfos(state.sources), []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", nil)}
 		}
 		accounting.nodes += placementNodes
-		composed, failures := composeDocuments(placed)
+		composed, failures := composeDocumentsWithPolicy(policy, placed)
 		if composed != nil {
 			synthetic := acquisitionScope{contributions: []acquiredContribution{{
 				source: cloneSourceInfo(mainInfo), payload: completed.root, document: placed,
@@ -255,7 +259,7 @@ func loadImportedWithAccessTrackedAndProcessing(access *filesystemAccess, basePa
 	for _, contribution := range state.scope.contributions {
 		documents = append(documents, contribution.document)
 	}
-	composed, diagnostics := composeDocuments(documents...)
+	composed, diagnostics := composeDocumentsWithPolicy(policy, documents...)
 	if composed != nil {
 		projectNestedEvidence(composed, state.scope)
 		projectAcquiredDestinations(composed, state.scope, state.sources)

internal/engine/load.go

@@ -32,6 +32,7 @@ type Options[T any] struct {
 	SourceVersionMode sourceVersionMode
 	destination       Path
 	placementMode     placementMode
+	sequencePolicy    sequencePolicy
 }
 
 // FilesystemContext supplies filesystem roots for imports and overlays.
@@ -162,6 +163,7 @@ func executeLoadRequest[T any](request loadRequest[T], secrets SecretBinding, lo
 	accumulator := diagnosticAccumulator{}
 	accounting := &importAccounting{}
 	mainOptions := Options[T]{Definition: request.definition, Imports: request.main.imports, Filesystem: request.main.filesystem, destination: request.main.destination.clone(), placementMode: request.main.placementMode}
+	mainOptions.sequencePolicy = request.sequencePolicy
 	mainOptions.SourceVersionMode = request.sourceVersionMode
 	mainOptions.Definition.Version = request.protectionVersion()
 	mainOptions.Overrides = request.overrides
@@ -205,6 +207,14 @@ func executeLoadRequest[T any](request loadRequest[T], secrets SecretBinding, lo
 			return nil, accumulator.loadError()
 		}
 	}
+	if request.ordinary {
+		input.composition, diagnostics = compositionForInputWithPolicy(input, request.sequencePolicy)
+		if len(diagnostics) != 0 {
+			accumulator.add(diagnostics...)
+			return nil, accumulator.loadError()
+		}
+		input.document = input.composition.root
+	}
 	if request.ordinary {
 		input, diagnostics = applyInputLayers(request.main.source, mainOptions, input)
 	} else {

internal/engine/load_input.go

@@ -121,7 +121,7 @@ func assembleBaseInputTrackedForRole[T any](source Source, options Options[T], d
 		}
 	}
 	if options.Imports {
-		input.composition, input.sources, diagnostics = loadImportedDocumentsTrackedWithProcessing(source, options.Filesystem, parsed, destination, options.Definition.Version, limits, accounting, visited, hostToken, versionState, versionProcessing)
+		input.composition, input.sources, diagnostics = loadImportedDocumentsTrackedWithProcessing(source, options.Filesystem, parsed, destination, options.Definition.Version, limits, accounting, visited, hostToken, versionState, versionProcessing, options.sequencePolicy)
 		if len(diagnostics) != 0 {
 			return input, diagnostics
 		}
@@ -196,7 +196,7 @@ func applyInputLayers[T any](source Source, options Options[T], input assembledI
 	if !activeOverlay(options.Overlay) && len(options.Overrides) == 0 {
 		return input, nil
 	}
-	composition, diagnostics := compositionForInput(input)
+	composition, diagnostics := compositionForInputWithPolicy(input, options.sequencePolicy)
 	if len(diagnostics) != 0 {
 		return input, diagnostics
 	}
@@ -231,13 +231,17 @@ func applyInputLayers[T any](source Source, options Options[T], input assembledI
 // compositionForInput initializes one handoff for overlays and CLI, including
 // defaults-only CLI input. Raw source bytes and mapping keys retain their budgets.
 func compositionForInput(input assembledInput) (*composedDocument, []Diagnostic) {
+	return compositionForInputWithPolicy(input, sequencePolicy{})
+}
+
+func compositionForInputWithPolicy(input assembledInput, policy sequencePolicy) (*composedDocument, []Diagnostic) {
 	if input.composition != nil {
 		return input.composition, nil
 	}
 	if input.document == nil {
 		return &composedDocument{root: &document.Node{Kind: document.Mapping}, origins: make(map[string]Origin)}, nil
 	}
-	composition, diagnostics := composeDocuments(input.document)
+	composition, diagnostics := composeDocumentsWithPolicy(policy, input.document)
 	if len(diagnostics) != 0 {
 		return nil, diagnostics
 	}

internal/engine/load_request.go

@@ -77,7 +77,7 @@ func loadGroupedWithSecretLookup[T any](request loadRequest[T], secrets SecretBi
 func assembleLayeredInput[T any](request loadRequest[T], input assembledInput, accounting *importAccounting) (assembledInput, []Diagnostic) {
 	groups := make([]*composedDocument, 0, len(request.underlays))
 	for index, group := range request.underlays {
-		composition, sources, contexts, diagnostics := acquireUnderlayGroupWithVersion(group, index+1, request.protectionVersion(), accounting, input.versionState)
+		composition, sources, contexts, diagnostics := acquireUnderlayGroupWithPolicyVersion(group, index+1, request.protectionVersion(), accounting, input.versionState, request.sequencePolicy)
 		input.sources = append(input.sources, sources...)
 		mergeResourceContexts(&input.contexts, contexts)
 		if len(diagnostics) != 0 {
@@ -99,7 +99,7 @@ func assembleLayeredInput[T any](request loadRequest[T], input assembledInput, a
 		}
 	}
 	if input.document != nil {
-		mainComposition, diagnostics := compositionForInput(input)
+		mainComposition, diagnostics := compositionForInputWithPolicy(input, request.sequencePolicy)
 		if len(diagnostics) != 0 {
 			return input, diagnostics
 		}
@@ -151,6 +151,10 @@ func acquireUnderlayGroup(group loadGroup, index int, rule *VersionRule, account
 }
 
 func acquireUnderlayGroupWithVersion(group loadGroup, index int, rule *VersionRule, accounting *importAccounting, versionState *sourceVersionState) (*composedDocument, []SourceInfo, map[string]ResourceContext, []Diagnostic) {
+	return acquireUnderlayGroupWithPolicyVersion(group, index, rule, accounting, versionState, sequencePolicy{})
+}
+
+func acquireUnderlayGroupWithPolicyVersion(group loadGroup, index int, rule *VersionRule, accounting *importAccounting, versionState *sourceVersionState, policy sequencePolicy) (*composedDocument, []SourceInfo, map[string]ResourceContext, []Diagnostic) {
 	initialBytes, initialNodes := accounting.bytes, accounting.nodes
 	visited := make(map[resourceIdentity]importRoute)
 	hostToken := &namespaceToken{}
@@ -161,7 +165,7 @@ func acquireUnderlayGroupWithVersion(group loadGroup, index int, rule *VersionRu
 		options := Options[struct{}]{
 			Imports: contribution.imports, Filesystem: contribution.filesystem,
 			Definition: Definition[struct{}]{Version: rule}, destination: contribution.destination.clone(),
-			placementMode: contribution.placementMode,
+			placementMode: contribution.placementMode, sequencePolicy: policy,
 		}
 		acquired, diagnostics := assembleBaseInputTrackedForRole(contribution.source, options, contribution.destination, accounting, visited, hostToken, versionState, sourceVersionContribution)
 		for _, source := range acquired.sources {
@@ -194,7 +198,7 @@ func acquireUnderlayGroupWithVersion(group loadGroup, index int, rule *VersionRu
 	if len(documents) == 0 {
 		return nil, sources, contexts, nil
 	}
-	composition, diagnostics := composeDocuments(documents...)
+	composition, diagnostics := composeDocumentsWithPolicy(policy, documents...)
 	if len(diagnostics) != 0 {
 		return nil, sources, contexts, diagnostics
 	}

internal/engine/sequence_registration_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"testing/fstest"
 )
 
 func TestCompileSequencePolicyNormalizesRegistrations(t *testing.T) {
@@ -125,3 +126,181 @@ func requireSequenceRegistrationOptionError(t *testing.T, err error) {
 		t.Fatalf("diagnostics = %#v, want one options diagnostic", diagnostics)
 	}
 }
+
+func TestRegisteredSequencePolicyAcrossLoadingLifecycle(t *testing.T) {
+	t.Parallel()
+	type item struct {
+		ID string `yaml:"id"`
+	}
+	type config struct {
+		Items []item `yaml:"items"`
+	}
+	definition := Definition[config]{sequenceRegistrations: []sequenceRegistration{{path: Keys("items"), identityField: "id"}}}
+	registered := Options[config]{Definition: definition}
+	loadCode := func(t *testing.T, err error) Code {
+		t.Helper()
+		var loadErr *LoadError
+		if !errors.As(err, &loadErr) || len(loadErr.Diagnostics()) == 0 {
+			t.Fatalf("error = %v, want a diagnostic load error", err)
+		}
+		return loadErr.Diagnostics()[0].Code
+	}
+
+	t.Run("ordinary authored list is checked with one source", func(t *testing.T) {
+		if _, err := testLoad[config](Bytes("main", []byte("items: [{id: same}, {id: same}]\n")), registered); err == nil || loadCode(t, err) != CodeComposeConflict {
+			t.Fatalf("duplicate identity error = %v", err)
+		}
+		result, err := testLoad[config](Bytes("main", []byte("items: [{id: one}]\n")), registered)
+		if err != nil || len(result.Config.Items) != 1 || result.Config.Items[0].ID != "one" {
+			t.Fatalf("single-source load = %#v, %v", result, err)
+		}
+	})
+
+	t.Run("ordinary imports retain registered composition", func(t *testing.T) {
+		directory := t.TempDir()
+		mainPath := writeImportFile(t, directory, "main.yml", "configkit: {imports: [{path: fragment.yml}]}\nitems: [{id: authored}]\n")
+		writeImportFile(t, directory, "fragment.yml", "items: [{id: imported}]\n")
+		result, err := testLoad[config](File(mainPath), Options[config]{Imports: true, Definition: definition})
+		if err != nil || len(result.Config.Items) != 2 || result.Config.Items[1].ID != "imported" {
+			t.Fatalf("registered imported list = %#v, %v", result, err)
+		}
+	})
+
+	t.Run("strict underlay group appends while separate tiers reset identities", func(t *testing.T) {
+		request := loadRequest[config]{
+			main: loadInput{source: Bytes("main", []byte("items: [{id: shared}]\n"))},
+			underlays: []loadGroup{{inputs: []loadInput{
+				{source: Bytes("first", []byte("items: [{id: lower}]\n"))},
+				{source: Bytes("second", []byte("items: [{id: other}]\n"))},
+			}}},
+			definition: definition, sequencePolicy: mustCompileSequencePolicy(t, definition.sequenceRegistrations),
+		}
+		result, err := testGroupedLoad(request)
+		if err != nil || len(result.Config.Items) != 1 || result.Config.Items[0].ID != "shared" {
+			t.Fatalf("grouped load = %#v, %v", result, err)
+		}
+
+		request.underlays = []loadGroup{{inputs: []loadInput{
+			{source: Bytes("first", []byte("items: [{id: shared}]\n"))},
+			{source: Bytes("second", []byte("items: [{id: lower}]\n"))},
+		}}}
+		request.main.source = Bytes("main", []byte("items: [{id: shared}]\n"))
+		if _, err := testGroupedLoad(request); err != nil {
+			t.Fatalf("same identity in a separate tier: %v", err)
+		}
+	})
+
+	t.Run("empty registered contribution clears lower sequence", func(t *testing.T) {
+		result, err := testLoad[config](Bytes("main", []byte("items: []\n")), registered)
+		if err != nil || len(result.Config.Items) != 0 {
+			t.Fatalf("empty sequence load = %#v, %v", result, err)
+		}
+		request := loadRequest[config]{
+			main:       loadInput{source: Bytes("main", []byte("items: []\n"))},
+			underlays:  []loadGroup{{inputs: []loadInput{{source: Bytes("lower", []byte("items: [{id: lower}]\n"))}}}},
+			definition: definition, sequencePolicy: mustCompileSequencePolicy(t, definition.sequenceRegistrations),
+		}
+		result, err = testGroupedLoad(request)
+		if err != nil || len(result.Config.Items) != 0 {
+			t.Fatalf("empty higher tier = %#v, %v", result, err)
+		}
+	})
+
+	t.Run("unregistered list overlap remains atomic", func(t *testing.T) {
+		request := loadRequest[config]{
+			main: loadInput{source: Bytes("main", []byte("items: [{id: main}]\n"))},
+			underlays: []loadGroup{{inputs: []loadInput{
+				{source: Bytes("first", []byte("items: [{id: lower}]\n"))},
+				{source: Bytes("second", []byte("items: [{id: other}]\n"))},
+			}}},
+		}
+		_, err := testGroupedLoad(request)
+		if got := loadCode(t, err); got != CodeComposeConflict {
+			t.Fatalf("unregistered overlap error = %s", got)
+		}
+	})
+
+	t.Run("malformed lower tier fails before precedence replacement", func(t *testing.T) {
+		request := loadRequest[config]{
+			main:       loadInput{source: Bytes("main", []byte("items: [{id: main}]\n"))},
+			underlays:  []loadGroup{{inputs: []loadInput{{source: Bytes("lower", []byte("items: [{id: same}, {id: same}]\n"))}}}},
+			definition: definition, sequencePolicy: mustCompileSequencePolicy(t, definition.sequenceRegistrations),
+		}
+		if got := loadCode(t, func() error { _, err := testGroupedLoad(request); return err }()); got != CodeComposeConflict {
+			t.Fatalf("hidden lower-tier error = %s", got)
+		}
+
+		overlayFilesystem := NewFilesystem("overlay", fstest.MapFS{
+			"overlays/testing.yml": &fstest.MapFile{Data: []byte("items: [{id: overlay}]\n")},
+		})
+		request.overlay = "testing"
+		request.overlayFilesystem = &FilesystemContext{Root: ".", BaseDir: ".", FS: &overlayFilesystem}
+		if got := loadCode(t, func() error { _, err := testGroupedLoad(request); return err }()); got != CodeComposeConflict {
+			t.Fatalf("selected replacement overlay hid lower-tier error: %s", got)
+		}
+	})
+
+	t.Run("replacement overlay and CLI may introduce duplicates", func(t *testing.T) {
+		directory := t.TempDir()
+		mainPath := writeImportFile(t, directory, "main.yml", "items: [{id: base}]\n")
+		writeImportFile(t, directory, "overlays/testing.yml", "items: [{id: duplicate}, {id: duplicate}]\n")
+		result, err := testLoad[config](File(mainPath), Options[config]{Definition: definition, Overlay: "testing"})
+		if err != nil {
+			t.Fatalf("replacement overlay error code = %s", loadCode(t, err))
+		}
+		if len(result.Config.Items) != 2 {
+			t.Fatalf("replacement overlay = %#v, %v", result, err)
+		}
+
+		result, err = testLoad[config](Bytes("main", []byte("items: [{id: base}]\n")), Options[config]{
+			Definition: definition,
+			Overrides:  []Override{{Path: Keys("items"), Flag: "--items", Value: []any{map[string]any{"id": "duplicate"}, map[string]any{"id": "duplicate"}}}},
+		})
+		if err != nil || len(result.Config.Items) != 2 {
+			t.Fatalf("replacement CLI value = %#v, %v", result, err)
+		}
+
+		resolved := definition
+		resolved.Resolve = func(value *config, resolution *Resolution) error {
+			value.Items[1].ID = value.Items[0].ID
+			resolution.Record(Keys("items"), ChangeNormalization, "registered identities collide")
+			return nil
+		}
+		result, err = testLoad[config](Bytes("main", []byte("items: [{id: first}, {id: second}]\n")), Options[config]{Definition: resolved})
+		if err != nil {
+			t.Fatalf("resolver replacement error code = %s", loadCode(t, err))
+		}
+		if len(result.Config.Items) != 2 {
+			t.Fatalf("resolver replacement = %#v, %v", result, err)
+		}
+	})
+
+	t.Run("application validation can reject replacement duplicates", func(t *testing.T) {
+		checked := definition
+		checked.Validate = func(value *config, validation *Validation) {
+			seen := make(map[string]bool)
+			for index, entry := range value.Items {
+				if seen[entry.ID] {
+					validation.Error(Keys("items").Index(index).Key("id"), "identity must be unique")
+				}
+				seen[entry.ID] = true
+			}
+		}
+		_, err := testLoad[config](Bytes("main", []byte("items: [{id: base}]\n")), Options[config]{
+			Definition: checked,
+			Overrides:  []Override{{Path: Keys("items"), Flag: "--items", Value: []any{map[string]any{"id": "duplicate"}, map[string]any{"id": "duplicate"}}}},
+		})
+		if got := loadCode(t, err); got != CodeValidateError {
+			t.Fatalf("application validation error = %s", got)
+		}
+	})
+}
+
+func mustCompileSequencePolicy(t *testing.T, registrations []sequenceRegistration) sequencePolicy {
+	t.Helper()
+	policy, diagnostics := compileSequencePolicy(registrations)
+	if len(diagnostics) != 0 {
+		t.Fatal(diagnostics)
+	}
+	return policy
+}