Schedule underlay groups as dependency forests

2026-10-19 19:53:01 +00:00
parent 263dd2f1bd
commit 014b99504b
7 changed files with 437 additions and 34 deletions
--- a/internal/engine/import_graph.go
+++ b/internal/engine/import_graph.go
@@ -89,6 +89,15 @@ func (graph importInclusionGraph) clone() importInclusionGraph {
 	return cloned
 }

+func mergeImportInclusionGraph(target *importInclusionGraph, source importInclusionGraph) {
+	for _, node := range source.nodes {
+		target.addNode(node)
+	}
+	for _, edge := range source.edges {
+		target.addEdge(edge.importer, edge.imported, edge.location)
+	}
+}
+
 func (graph importInclusionGraph) scheduleCycle(closingEdge int, location Location) (Diagnostic, bool) {
 	keys := make([]dependencyKey, len(graph.nodes))
 	for index, node := range graph.nodes {
@@ -149,21 +158,41 @@ func (graph importInclusionGraph) scheduleCycle(closingEdge int, location Locati
 }

 func scheduleAcquisitionScope(scope acquisitionScope, graph importInclusionGraph) acquisitionScope {
-	positions := make(map[importNodeKey]int, len(scope.contributions))
 	slots := make([]int, 0, len(scope.contributions))
-	keys := make([]dependencyKey, 0, len(scope.contributions))
 	for index, contribution := range scope.contributions {
 		if contribution.scopeOwner != scope.owner {
 			continue
 		}
-		positions[contribution.identity] = len(keys)
 		slots = append(slots, index)
-		keys = append(keys, contribution.identity.dependencyKey())
-	}
-	if len(keys) < 2 {
-		return scope
 	}
+	scheduled := scope
+	scheduled.contributions = scheduleContributionSlots(scope.contributions, slots, graph)
+	return scheduled
+}

+func scheduleAcquisitionGroup(scopes []acquisitionScope, graph importInclusionGraph) acquisitionScope {
+	var contributions []acquiredContribution
+	for _, scope := range scopes {
+		contributions = append(contributions, scope.contributions...)
+	}
+	slots := make([]int, len(contributions))
+	for index := range slots {
+		slots[index] = index
+	}
+	return acquisitionScope{contributions: scheduleContributionSlots(contributions, slots, graph)}
+}
+
+func scheduleContributionSlots(contributions []acquiredContribution, slots []int, graph importInclusionGraph) []acquiredContribution {
+	if len(slots) < 2 {
+		return append([]acquiredContribution(nil), contributions...)
+	}
+	positions := make(map[importNodeKey]int, len(slots))
+	keys := make([]dependencyKey, len(slots))
+	for position, slot := range slots {
+		contribution := contributions[slot]
+		positions[contribution.identity] = position
+		keys[position] = contribution.identity.dependencyKey()
+	}
 	edges := make([]dependencyEdge, 0, len(graph.edges))
 	for _, edge := range graph.edges {
 		importer, importerOK := positions[edge.importer]
@@ -174,16 +203,86 @@ func scheduleAcquisitionScope(scope acquisitionScope, graph importInclusionGraph
 	}
 	order, _ := scheduleDependencies(keys, edges)
 	if len(order) != len(keys) {
-		return scope
+		return append([]acquiredContribution(nil), contributions...)
 	}
-	scheduled := scope
-	scheduled.contributions = append([]acquiredContribution(nil), scope.contributions...)
+	scheduled := append([]acquiredContribution(nil), contributions...)
 	for position, scheduledIndex := range order {
-		scheduled.contributions[slots[position]] = scope.contributions[slots[scheduledIndex]]
+		scheduled[slots[position]] = contributions[slots[scheduledIndex]]
 	}
 	return scheduled
 }

+type importNamespaceAnchors struct {
+	kind   string
+	host   string
+	native string
+	input  string
+}
+
+func explicitNamespaceAnchor(group, input int) string {
+	return fmt.Sprintf("%020d/%020d", group, input)
+}
+
+func reanchorImportKey(key importNodeKey, anchors importNamespaceAnchors) importNodeKey {
+	if anchors.kind != "" {
+		key.namespaceKind = anchors.kind
+	}
+	switch {
+	case key.resourceKind == "synthetic":
+		key.namespaceAnchor = anchors.input
+	case key.namespaceKind == "native":
+		key.namespaceAnchor = anchors.native
+	default:
+		key.namespaceAnchor = anchors.host
+	}
+	return key
+}
+
+func reanchorImportGraph(graph importInclusionGraph, anchors importNamespaceAnchors) (importInclusionGraph, map[importNodeKey]importNodeKey) {
+	reanchored := newImportInclusionGraph()
+	keys := make(map[importNodeKey]importNodeKey, len(graph.nodes))
+	for _, node := range graph.nodes {
+		old := node.key
+		node.key = reanchorImportKey(old, anchors)
+		keys[old] = node.key
+		reanchored.addNode(node)
+	}
+	for _, edge := range graph.edges {
+		reanchored.addEdge(keys[edge.importer], keys[edge.imported], edge.location)
+	}
+	return reanchored, keys
+}
+
+func reanchorAcquisitionScope(scope *acquisitionScope, keys map[importNodeKey]importNodeKey, anchors importNamespaceAnchors) {
+	if scope == nil {
+		return
+	}
+	if key, exists := keys[scope.owner]; exists {
+		scope.owner = key
+	} else {
+		scope.owner = reanchorImportKey(scope.owner, anchors)
+	}
+	for index := range scope.contributions {
+		contribution := &scope.contributions[index]
+		if key, exists := keys[contribution.identity]; exists {
+			contribution.identity = key
+		} else {
+			contribution.identity = reanchorImportKey(contribution.identity, anchors)
+		}
+		if key, exists := keys[contribution.scopeOwner]; exists {
+			contribution.scopeOwner = key
+		} else {
+			contribution.scopeOwner = reanchorImportKey(contribution.scopeOwner, anchors)
+		}
+		reanchorAcquisitionScope(contribution.pendingItem, keys, anchors)
+		if contribution.nested != nil {
+			for nestedIndex := range contribution.nested.acquisition {
+				reanchorAcquisitionScope(&contribution.nested.acquisition[nestedIndex], keys, anchors)
+			}
+		}
+	}
+}
+
 func importKeyFor(access *filesystemAccess, canonical string, synthetic bool) importNodeKey {
 	kind := "host"
 	resourceKind := "file"
--- a/internal/engine/imports.go
+++ b/internal/engine/imports.go
@@ -80,12 +80,16 @@ func loadImportedDocumentsTrackedWithVersion(source Source, context *FilesystemC
 }

 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) {
+	return loadImportedDocumentsTrackedWithProcessingDeferred(source, context, main, destination, rule, limits, accounting, visited, hostToken, versionState, processing, policy, false)
+}
+
+func loadImportedDocumentsTrackedWithProcessingDeferred(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, deferAssembly bool) (*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 loadImportedWithAccessTrackedAndProcessingPolicy(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy)
+		return loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy, deferAssembly)
 	}
 	rootPath, basePath, mainPath, diagnostic := importFilesystem(source, context, main)
 	if diagnostic != nil {
@@ -100,7 +104,7 @@ func loadImportedDocumentsTrackedWithProcessing(source Source, context *Filesyst
 	if hostToken != nil {
 		access.token = hostToken
 	}
-	return loadImportedWithAccessTrackedAndProcessingPolicy(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy)
+	return loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy, deferAssembly)
 }

 func loadImportedWithAccess(access *filesystemAccess, basePath, mainPath string, main parsedSource, rule *VersionRule, limits importLimits, accounting *importAccounting) (*composedDocument, []SourceInfo, []Diagnostic) {
@@ -124,6 +128,10 @@ func loadImportedWithAccessTrackedAndProcessing(access *filesystemAccess, basePa
 }

 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) {
+	return loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy, false)
+}
+
+func loadImportedWithAccessTrackedAndProcessingPolicyDeferred(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, deferAssembly bool) (*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)}
@@ -221,7 +229,9 @@ func loadImportedWithAccessTrackedAndProcessingPolicy(access *filesystemAccess,
 	}
 	if len(diagnostics) == 0 {
 		state.scheduleAcquisitionScopes(&state.scope)
-		diagnostics = state.composePendingItems(&state.scope)
+		if !deferAssembly {
+			diagnostics = state.composePendingItems(&state.scope)
+		}
 	}

 	closeErr := access.close()
@@ -234,6 +244,27 @@ func loadImportedWithAccessTrackedAndProcessingPolicy(access *filesystemAccess,
 	if closeErr != nil {
 		return nil, cloneSourceInfos(state.sources), []Diagnostic{importDiagnostic(CodeImportRead, "configuration root could not be closed", nil)}
 	}
+	if deferAssembly {
+		acquisition := state.scope
+		if mainIsItem {
+			localScope := state.scope
+			acquisition = acquisitionScope{
+				owner: mainNodeKey, destination: destination.clone(),
+				contributions: []acquiredContribution{{
+					source: cloneSourceInfo(mainInfo), payload: body, identity: mainNodeKey,
+					role: importRoleMain, scopeOwner: mainNodeKey, destination: destination.clone(),
+					processing: processing, pendingItem: &localScope,
+				}},
+			}
+		}
+		if diagnostic := reservePendingItemPlacements(&acquisition, limits, accounting); diagnostic != nil {
+			return nil, cloneSourceInfos(state.sources), []Diagnostic{*diagnostic}
+		}
+		return &composedDocument{
+			acquisition: []acquisitionScope{cloneAcquisitionScope(acquisition)},
+			importGraph: state.importGraph.clone(), inputBytes: accounting.bytes, inputNodes: accounting.nodes,
+		}, cloneSourceInfos(state.sources), nil
+	}
 	if mainIsItem {
 		localScope := state.scope
 		documents := make([]*document.Node, 0, len(localScope.contributions))
@@ -313,6 +344,29 @@ func loadImportedWithAccessTrackedAndProcessingPolicy(access *filesystemAccess,
 	return composed, cloneSourceInfos(state.sources), diagnostics
 }

+func reservePendingItemPlacements(scope *acquisitionScope, limits importLimits, accounting *importAccounting) *Diagnostic {
+	if scope == nil {
+		return nil
+	}
+	for index := range scope.contributions {
+		contribution := &scope.contributions[index]
+		if contribution.pendingItem == nil {
+			continue
+		}
+		placementNodes := destinationWrapperNodeCountForMode(contribution.destination, placementModeSequenceItem)
+		if accounting.nodes+placementNodes > limits.nodes {
+			diagnostic := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", contribution.source.ImportedFrom)
+			return &diagnostic
+		}
+		accounting.nodes += placementNodes
+		contribution.placementCharged = true
+		if diagnostic := reservePendingItemPlacements(contribution.pendingItem, limits, accounting); diagnostic != nil {
+			return diagnostic
+		}
+	}
+	return nil
+}
+
 func (s *importTraversal) checkInitialLimits() *Diagnostic {
 	if s.accounting.sources > s.limits.sources {
 		value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source-count limit", nil)
@@ -609,6 +663,9 @@ func (s *importTraversal) composePendingItems(scope *acquisitionScope) []Diagnos
 		if len(diagnostics) != 0 {
 			return projectItemDiagnosticPaths(diagnostics, *scope, contribution.destination, index)
 		}
+		if diagnostic := importedVersionClaim(documents, *itemScope, completed, s.rule); diagnostic != nil {
+			return []Diagnostic{*diagnostic}
+		}
 		completed.acquisition = []acquisitionScope{cloneAcquisitionScope(*itemScope)}
 		projectNestedEvidence(completed, *itemScope)
 		_, placed, err := placeDocumentWithMode(completed.root, contribution.destination, placementModeSequenceItem, contribution.payload.Location, maxDocumentDepth, maxDocumentNodes)
@@ -616,10 +673,12 @@ func (s *importTraversal) composePendingItems(scope *acquisitionScope) []Diagnos
 			return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source depth or node limit after destination placement", contribution.source.ImportedFrom)}
 		}
 		placementNodes := destinationWrapperNodeCountForMode(contribution.destination, placementModeSequenceItem)
-		if s.accounting.nodes+placementNodes > s.limits.nodes {
-			return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", contribution.source.ImportedFrom)}
+		if !contribution.placementCharged {
+			if s.accounting.nodes+placementNodes > s.limits.nodes {
+				return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", contribution.source.ImportedFrom)}
+			}
+			s.accounting.nodes += placementNodes
 		}
-		s.accounting.nodes += placementNodes
 		contribution.payload = completed.root
 		contribution.document = placed
 		contribution.nested = completed
--- a/internal/engine/load.go
+++ b/internal/engine/load.go
@@ -33,6 +33,7 @@ type Options[T any] struct {
 	destination       Path
 	placementMode     placementMode
 	sequencePolicy    sequencePolicy
+	deferAssembly     bool
 }

 // FilesystemContext supplies filesystem roots for imports and overlays.
--- a/internal/engine/load_input.go
+++ b/internal/engine/load_input.go
@@ -33,6 +33,7 @@ type acquiredContribution struct {
 	processing          *sourceVersionProcessingResult
 	nested              *composedDocument
 	pendingItem         *acquisitionScope
+	placementCharged    bool
 }

 // acquisitionScope is the ordered strict mapping scope produced by one input
@@ -54,6 +55,7 @@ func cloneAcquisitionScope(scope acquisitionScope) acquisitionScope {
 				document: contribution.document, destination: contribution.destination.clone(),
 				relativeDestination: contribution.relativeDestination.clone(), processing: contribution.processing,
 				nested: cloneComposedDocument(contribution.nested), pendingItem: cloneAcquisitionScopePointer(contribution.pendingItem),
+				placementCharged: contribution.placementCharged,
 			}
 		}
 	}
@@ -135,12 +137,18 @@ 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, options.sequencePolicy)
+		if options.deferAssembly {
+			input.composition, input.sources, diagnostics = loadImportedDocumentsTrackedWithProcessingDeferred(source, options.Filesystem, parsed, destination, options.Definition.Version, limits, accounting, visited, hostToken, versionState, versionProcessing, options.sequencePolicy, true)
+		} else {
+			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
 		}
-		input.acquisition = cloneAcquisitionScopes(input.composition.acquisition)
-		input.document = input.composition.root
+		if input.composition != nil {
+			input.acquisition = cloneAcquisitionScopes(input.composition.acquisition)
+			input.document = input.composition.root
+		}
 	} else {
 		input.sources = sourceInfos(parsed.source)
 		if versionProcessing != nil {
--- a/internal/engine/load_request.go
+++ b/internal/engine/load_request.go
@@ -158,14 +158,23 @@ func acquireUnderlayGroupWithPolicyVersion(group loadGroup, index int, rule *Ver
 	initialBytes, initialNodes := accounting.bytes, accounting.nodes
 	visited := make(map[resourceIdentity]importRoute)
 	hostToken := &namespaceToken{}
+	nativeAnchors := make(map[*namespaceToken]string)
+	for inputIndex, contribution := range group.inputs {
+		if token := nativeTokenForLoadInput(contribution); token != nil {
+			if _, exists := nativeAnchors[token]; !exists {
+				nativeAnchors[token] = explicitNamespaceAnchor(index, inputIndex+1)
+			}
+		}
+	}
 	var scopes []acquisitionScope
+	inclusionGraph := newImportInclusionGraph()
 	var sources []SourceInfo
 	var contexts map[string]ResourceContext
-	for _, contribution := range group.inputs {
+	for inputIndex, contribution := range group.inputs {
 		options := Options[struct{}]{
 			Imports: contribution.imports, Filesystem: contribution.filesystem,
 			Definition: Definition[struct{}]{Version: rule}, destination: contribution.destination.clone(),
-			placementMode: contribution.placementMode, sequencePolicy: policy,
+			placementMode: contribution.placementMode, sequencePolicy: policy, deferAssembly: true,
 		}
 		acquired, diagnostics := assembleBaseInputTrackedForRole(contribution.source, options, contribution.destination, accounting, visited, hostToken, versionState, sourceVersionContribution)
 		for _, source := range acquired.sources {
@@ -177,16 +186,37 @@ func acquireUnderlayGroupWithPolicyVersion(group loadGroup, index int, rule *Ver
 		if len(diagnostics) != 0 {
 			return nil, sources, contexts, projectPriorInputDiagnosticPaths(diagnostics, scopes)
 		}
-		if acquired.document == nil {
+		if acquired.document == nil && len(acquired.acquisition) == 0 {
 			continue
 		}
-		scopes = append(scopes, cloneAcquisitionScopes(acquired.acquisition)...)
+		anchors := importNamespaceAnchors{
+			kind: "host", host: explicitNamespaceAnchor(index, 0),
+			input: explicitNamespaceAnchor(index, inputIndex+1),
+		}
+		if token := nativeTokenForLoadInput(contribution); token != nil {
+			anchors.kind = "native"
+			anchors.native = nativeAnchors[token]
+		}
+		inputScopes := cloneAcquisitionScopes(acquired.acquisition)
+		inputGraph := newImportInclusionGraph()
+		if acquired.composition != nil {
+			inputGraph = acquired.composition.importGraph
+		}
+		inputGraph, keyMap := reanchorImportGraph(inputGraph, anchors)
+		for scopeIndex := range inputScopes {
+			reanchorAcquisitionScope(&inputScopes[scopeIndex], keyMap, anchors)
+		}
+		scopes = append(scopes, inputScopes...)
+		mergeImportInclusionGraph(&inclusionGraph, inputGraph)
+	}
+	groupScope := scheduleAcquisitionGroup(scopes, inclusionGraph)
+	traversal := importTraversal{accounting: accounting, limits: defaultImportLimits(), rule: rule, importGraph: inclusionGraph}
+	if diagnostics := traversal.composePendingItems(&groupScope); len(diagnostics) != 0 {
+		return nil, sources, contexts, diagnostics
 	}
 	var documents []*document.Node
-	for _, scope := range scopes {
-		for _, contribution := range scope.contributions {
-			documents = append(documents, contribution.document)
-		}
+	for _, contribution := range groupScope.contributions {
+		documents = append(documents, contribution.document)
 	}
 	if len(documents) == 0 {
 		return nil, sources, contexts, nil
@@ -197,6 +227,18 @@ func acquireUnderlayGroupWithPolicyVersion(group loadGroup, index int, rule *Ver
 	}
 	if composition != nil && rule != nil {
 		ranges := indexSequenceRanges(composition.sequenceRanges)
+		for sourceIndex, document := range documents {
+			if groupScope.contributions[sourceIndex].role == importRoleMain {
+				continue
+			}
+			if node := fragmentVersionClaimWithIndex(document, rule, sourceIndex, ranges); node != nil {
+				location := publicLocation(node.Location)
+				return nil, sources, contexts, []Diagnostic{{
+					Code: CodeImportVersion, Severity: SeverityError, Path: rule.Path.clone(),
+					Message: "imported fragments cannot contribute to the schema-version path", Primary: &location,
+				}}
+			}
+		}
 		for sourceIndex, document := range documents {
 			if node := fragmentVersionClaimWithIndex(document, rule, sourceIndex, ranges); node != nil {
 				location := publicLocation(node.Location)
@@ -209,13 +251,14 @@ func acquireUnderlayGroupWithPolicyVersion(group loadGroup, index int, rule *Ver
 	}
 	composition.inputBytes = accounting.bytes - initialBytes
 	composition.inputNodes = accounting.nodes - initialNodes
-	composition.acquisition = cloneAcquisitionScopes(scopes)
-	var projectionScope acquisitionScope
-	for _, scope := range scopes {
-		projectionScope.contributions = append(projectionScope.contributions, scope.contributions...)
+	groupScope.owner = importNodeKey{namespaceKind: "group", namespaceAnchor: explicitNamespaceAnchor(index, 0), resourceKind: "scope"}
+	for contributionIndex := range groupScope.contributions {
+		groupScope.contributions[contributionIndex].scopeOwner = groupScope.owner
 	}
-	projectNestedEvidence(composition, projectionScope)
-	projectAcquiredDestinations(composition, projectionScope, sources)
+	composition.acquisition = []acquisitionScope{cloneAcquisitionScope(groupScope)}
+	composition.importGraph = inclusionGraph.clone()
+	projectNestedEvidence(composition, groupScope)
+	projectAcquiredDestinations(composition, groupScope, sources)
 	for pointer, origin := range composition.origins {
 		origin.Layer = LayerUnderlay
 		composition.origins[pointer] = origin
@@ -225,3 +268,13 @@ func acquireUnderlayGroupWithPolicyVersion(group loadGroup, index int, rule *Ver
 	}
 	return composition, sources, contexts, nil
 }
+
+func nativeTokenForLoadInput(input loadInput) *namespaceToken {
+	if input.source.kind == sourceFilesystem && input.source.namespace != nil {
+		return input.source.namespace.token
+	}
+	if input.filesystem != nil && input.filesystem.FS != nil && input.filesystem.FS.namespace != nil {
+		return input.filesystem.FS.namespace.token
+	}
+	return nil
+}
--- a/internal/engine/load_request_test.go
+++ b/internal/engine/load_request_test.go
@@ -19,6 +19,186 @@ func testGroupedLoad[T any](request loadRequest[T]) (*testResult[T], error) {
 	return testLoaded(loaded), err
 }

+func TestUnderlayGroupSchedulesInterleavedImportBranches(t *testing.T) {
+	directory := t.TempDir()
+	zRoot := writeImportFile(t, directory, "z-root.yml", `configkit:
+  imports:
+    - path: d-parent.yml
+items: [{id: z}]
+`)
+	dParent := writeImportFile(t, directory, "d-parent.yml", `configkit:
+  imports:
+    - path: a-dependency.yml
+items: [{id: d}]
+`)
+	aDependency := writeImportFile(t, directory, "a-dependency.yml", "items: [{id: a}]\n")
+	bRoot := writeImportFile(t, directory, "b-root.yml", `configkit:
+  imports:
+    - path: c-helper.yml
+items: [{id: b}]
+`)
+	cHelper := writeImportFile(t, directory, "c-helper.yml", "items: [{id: c}]\n")
+
+	request := loadRequest[map[string]any]{
+		main: loadInput{source: NoSource()},
+		underlays: []loadGroup{{inputs: []loadInput{
+			{source: NoSource()},
+			{source: File(zRoot), imports: true},
+			{source: File(bRoot), imports: true},
+		}}},
+		definition: Definition[map[string]any]{sequenceRegistrations: []sequenceRegistration{{path: Keys("items"), identityField: "id"}}},
+	}
+	result, err := testGroupedLoad(request)
+	if err != nil {
+		t.Fatalf("grouped load: %v", err)
+	}
+	items := result.Config["items"].([]any)
+	wantIDs := []string{"a", "c", "b", "d", "z"}
+	if len(items) != len(wantIDs) {
+		t.Fatalf("scheduled items = %#v", items)
+	}
+	for index, want := range wantIDs {
+		if got := items[index].(map[string]any)["id"]; got != want {
+			t.Errorf("scheduled item %d ID = %#v, want %q", index, got, want)
+		}
+	}
+	wantPaths := []string{zRoot, dParent, aDependency, bRoot, cHelper}
+	for index, want := range wantPaths {
+		if got := result.Sources()[index].Path; got != canonicalTestPath(t, want) {
+			t.Errorf("acquired source %d path = %q, want %q", index, got, canonicalTestPath(t, want))
+		}
+	}
+	if got := result.Sources(); len(got) != len(wantPaths) {
+		t.Fatalf("NoSource added a source record: %#v", got)
+	}
+}
+
+func TestUnderlayNamespaceAnchorsUseInputPositionsAndFilesystemIdentity(t *testing.T) {
+	firstNamespace := NewFilesystem("same-label", fstest.MapFS{
+		"root.yml":  &fstest.MapFile{Data: []byte("items: [{id: first}]\n")},
+		"later.yml": &fstest.MapFile{Data: []byte("items: [{id: copied}]\n")},
+	})
+	secondNamespace := NewFilesystem("same-label", fstest.MapFS{
+		"root.yml": &fstest.MapFile{Data: []byte("items: [{id: second}]\n")},
+	})
+	request := loadRequest[map[string]any]{
+		main: loadInput{source: NoSource()},
+		underlays: []loadGroup{{inputs: []loadInput{
+			{source: firstNamespace.File("root.yml")},
+			{source: secondNamespace.File("root.yml")},
+			{source: firstNamespace.File("later.yml")},
+		}}},
+		definition: Definition[map[string]any]{sequenceRegistrations: []sequenceRegistration{{path: Keys("items"), identityField: "id"}}},
+	}
+	result, err := testGroupedLoad(request)
+	if err != nil {
+		t.Fatalf("grouped load with native namespaces: %v", err)
+	}
+	items := result.Config["items"].([]any)
+	wantIDs := []string{"copied", "first", "second"}
+	if len(items) != len(wantIDs) {
+		t.Fatalf("scheduled native items = %#v", items)
+	}
+	for index, want := range wantIDs {
+		if got := items[index].(map[string]any)["id"]; got != want {
+			t.Errorf("scheduled native item %d ID = %#v, want %q", index, got, want)
+		}
+	}
+	if sources := result.Sources(); len(sources) != len(wantIDs) {
+		t.Fatalf("source records = %#v", sources)
+	}
+}
+
+func TestUnderlayItemConflictUsesScheduledGroupIndex(t *testing.T) {
+	directory := t.TempDir()
+	zItem := writeImportFile(t, directory, "z-item.yml", "id: duplicate\n")
+	aItem := writeImportFile(t, directory, "a-item.yml", "id: duplicate\n")
+	request := loadRequest[map[string]any]{
+		main: loadInput{source: NoSource()},
+		underlays: []loadGroup{{inputs: []loadInput{
+			{source: File(zItem), destination: Keys("items"), placementMode: placementModeSequenceItem},
+			{source: File(aItem), destination: Keys("items"), placementMode: placementModeSequenceItem},
+		}}},
+		definition: Definition[map[string]any]{sequenceRegistrations: []sequenceRegistration{{path: Keys("items"), identityField: "id"}}},
+	}
+	_, err := testGroupedLoad(request)
+	loadError, ok := err.(*LoadError)
+	if !ok || len(loadError.Diagnostics()) != 1 {
+		t.Fatalf("expected item conflict: %v", err)
+	}
+	diagnostic := loadError.Diagnostics()[0]
+	if diagnostic.Code != CodeComposeConflict || diagnostic.Path.String() != "/items" {
+		t.Fatalf("group item conflict = %s %s, want compose.conflict /items", diagnostic.Code, diagnostic.Path.String())
+	}
+	files := make(map[string]string)
+	for _, source := range loadError.Sources() {
+		files[source.ID] = source.Path
+	}
+	locations := append([]Location(nil), diagnostic.Related...)
+	if diagnostic.Primary != nil {
+		locations = append(locations, *diagnostic.Primary)
+	}
+	wantLocations := map[string]Location{
+		canonicalTestPath(t, zItem): {Line: 1, Column: 5},
+		canonicalTestPath(t, aItem): {Line: 1, Column: 5},
+	}
+	gotPaths := make(map[string]bool, len(locations))
+	for _, location := range locations {
+		want := wantLocations[files[location.SourceID]]
+		gotPaths[files[location.SourceID]] = location.Line == want.Line && location.Column == want.Column
+	}
+	if len(locations) != 2 || !gotPaths[canonicalTestPath(t, zItem)] || !gotPaths[canonicalTestPath(t, aItem)] {
+		t.Fatalf("group item conflict sites = %#v; source paths = %#v", locations, files)
+	}
+}
+
+func TestUnderlayItemHelperConflictUsesScheduledGroupIndex(t *testing.T) {
+	directory := t.TempDir()
+	zBad := writeImportFile(t, directory, "z-bad.yml", "configkit: {imports: [{path: helper.yml}]}\nid: duplicate\n")
+	helper := writeImportFile(t, directory, "helper.yml", "id: duplicate\n")
+	aGood := writeImportFile(t, directory, "a-good.yml", "id: good\n")
+	request := loadRequest[map[string]any]{
+		main: loadInput{source: NoSource()},
+		underlays: []loadGroup{{inputs: []loadInput{
+			{source: File(zBad), imports: true, destination: Keys("items"), placementMode: placementModeSequenceItem},
+			{source: File(aGood), destination: Keys("items"), placementMode: placementModeSequenceItem},
+		}}},
+		definition: Definition[map[string]any]{sequenceRegistrations: []sequenceRegistration{{path: Keys("items"), identityField: "id"}}},
+	}
+	_, err := testGroupedLoad(request)
+	loadError, ok := err.(*LoadError)
+	if !ok || len(loadError.Diagnostics()) != 1 {
+		t.Fatalf("expected item helper conflict: %v", err)
+	}
+	diagnostic := loadError.Diagnostics()[0]
+	if diagnostic.Code != CodeComposeConflict || diagnostic.Path.String() != "/items/1/id" {
+		t.Fatalf("group item helper conflict = %s %s, want compose.conflict /items/1/id", diagnostic.Code, diagnostic.Path.String())
+	}
+	files := make(map[string]string)
+	for _, source := range loadError.Sources() {
+		files[source.ID] = source.Path
+	}
+	if len(files) != 3 {
+		t.Fatalf("group acquisition stopped before all sources were read: %#v", files)
+	}
+	locations := append([]Location(nil), diagnostic.Related...)
+	if diagnostic.Primary != nil {
+		locations = append(locations, *diagnostic.Primary)
+	}
+	wantLocations := map[string]Location{
+		canonicalTestPath(t, zBad):   {Line: 2, Column: 5},
+		canonicalTestPath(t, helper): {Line: 1, Column: 5},
+	}
+	gotPaths := make(map[string]bool, len(locations))
+	for _, location := range locations {
+		want := wantLocations[files[location.SourceID]]
+		gotPaths[files[location.SourceID]] = location.Line == want.Line && location.Column == want.Column
+	}
+	if len(locations) != 2 || !gotPaths[canonicalTestPath(t, zBad)] || !gotPaths[canonicalTestPath(t, helper)] {
+		t.Fatalf("group item helper conflict sites = %#v; source paths = %#v", locations, files)
+	}
+}
+
 func TestGroupedLoadPrecedenceSourceOrderAndSingleLifecycle(t *testing.T) {
 	filesystem := NewFilesystem("bundle", fstest.MapFS{
 		"main.yml":          &fstest.MapFile{Data: []byte("configkit: {imports: [{path: imported.yml}]}\nversion: v1\nport: 5\nsettings: {shared: main}\n")},