diff --git a/internal/engine/glob.go b/internal/engine/glob.go
index 3216fbc..29649e1 100644
--- a/internal/engine/glob.go
+++ b/internal/engine/glob.go
@@ -5,13 +5,12 @@ import (
 	"io/fs"
 	"path"
 	"slices"
+	"sort"
 	"strings"
 )
 
 const globReadBatch = 128
 
-var errInvalidImportGlobPattern = errors.New("invalid import glob filename pattern")
-
 type globCandidate struct {
 	absolute string
 	relative string
@@ -58,9 +57,6 @@ func (s *importTraversal) expandImportGlob(basePath string, declaration importDe
 				}
 				if last {
 					next = append(next, globCandidate{absolute: canonical, relative: relative})
-					if s.accounting.sources+len(next) > s.limits.sources {
-						return nil, []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "glob matches exceed the remaining source-count limit", &location)}
-					}
 					continue
 				}
 				info, err := s.access.stat(canonical)
@@ -86,44 +82,212 @@ func (s *importTraversal) expandImportGlob(basePath string, declaration importDe
 	return matches, nil
 }
 
-func (s *importTraversal) readGlobMatches(directory, pattern string, location Location) ([]string, []Diagnostic) {
-	canonical, err := s.access.canonical(directory)
+// selectGlobImportCandidates expands path components while retaining the
+// lexical spelling used for filtering and later source admission.
+func selectGlobImportCandidates(access *filesystemAccess, basePath string, spec importDeclarationSpec, policy resolvedImportPolicy, limits importLimits, accounting *importAccounting) (importSelection, []Diagnostic) {
+	site := spec.fieldSites["glob"]
+	segments := spec.globSegments
+	if len(segments) == 0 {
+		var valid bool
+		segments, valid = validGlobPattern(spec.glob)
+		if !valid {
+			return importSelection{}, []Diagnostic{*importSelectionDiagnostic(site, CodeImportInvalid, "import glob pattern is invalid")}
+		}
+	}
+	if accounting == nil {
+		accounting = &importAccounting{}
+	}
+	rootPath := normalizedImportCandidatePath(access, basePath)
+	candidates := []selectedImportCandidate{{path: rootPath, accessPath: basePath, site: site}}
+	for index, segment := range segments {
+		last := index == len(segments)-1
+		literal, patterned := globLiteralSegment(segment)
+		nextByPath := make(map[string]selectedImportCandidate)
+		for _, parent := range candidates {
+			if !patterned {
+				childPath := access.join(parent.accessPath, literal)
+				candidate, missing, diagnostics := explicitGlobCandidate(access, childPath, site, last)
+				if len(diagnostics) != 0 {
+					return importSelection{}, diagnostics
+				}
+				if !missing {
+					nextByPath[candidate.path] = candidate
+				}
+				continue
+			}
+
+			entries, missing, diagnostics := readGlobEntries(access, parent.accessPath, site.location, limits, accounting)
+			if len(diagnostics) != 0 {
+				return importSelection{}, diagnostics
+			}
+			if missing {
+				continue
+			}
+			for _, entry := range entries {
+				matched, err := path.Match(segment, entry.Name())
+				if err != nil {
+					return importSelection{}, []Diagnostic{*importSelectionDiagnostic(site, CodeImportInvalid, "import glob pattern is invalid")}
+				}
+				if !matched || !policy.hidden && strings.HasPrefix(entry.Name(), ".") {
+					continue
+				}
+				accessPath := access.join(parent.accessPath, entry.Name())
+				if !access.contains(accessPath) {
+					return importSelection{}, []Diagnostic{*importSelectionDiagnostic(site, CodeImportOutsideRoot, "glob traversal is outside the configuration root")}
+				}
+				candidate := selectedImportCandidate{path: normalizedImportCandidatePath(access, accessPath), accessPath: accessPath, site: site}
+				target, err := access.importEntryTarget(accessPath, entry, policy.follow)
+				if err != nil {
+					return importSelection{}, []Diagnostic{globCandidateDiagnostic(err, site.location, "glob match could not be inspected")}
+				}
+				if target.symlink && !policy.follow {
+					continue
+				}
+				if !last && !target.dir {
+					continue
+				}
+				nextByPath[candidate.path] = candidate
+			}
+		}
+		candidates = candidatesFromMap(nextByPath)
+		if len(candidates) == 0 {
+			break
+		}
+	}
+
+	selected := make([]selectedImportCandidate, 0, len(candidates))
+	for _, candidate := range candidates {
+		missing, _, diagnostic := validateNamedImportTarget(access, candidate, false, false)
+		if diagnostic != nil {
+			return importSelection{}, []Diagnostic{*diagnostic}
+		}
+		if missing || excludedBasename(path.Base(candidate.path), policy.exclude) {
+			continue
+		}
+		selected = append(selected, candidate)
+	}
+	if policy.required && len(selected) == 0 {
+		return importSelection{}, []Diagnostic{*importSelectionDiagnostic(site, CodeImportMissing, "required import glob selected no files")}
+	}
+	// Source allowance is charged by source admission, after this filtered set
+	// reaches the traversal; that preserves repeated-source diagnostics first.
+	selection := importSelection{candidates: selected}
+	if len(selected) != 0 {
+		selection.outcome = importSelectionHasCandidates
+	}
+	return selection, nil
+}
+
+func explicitGlobCandidate(access *filesystemAccess, candidatePath string, site importFieldSite, last bool) (selectedImportCandidate, bool, []Diagnostic) {
+	if !access.contains(candidatePath) {
+		return selectedImportCandidate{}, false, []Diagnostic{*importSelectionDiagnostic(site, CodeImportOutsideRoot, "glob traversal is outside the configuration root")}
+	}
+	target, err := access.importEntryTarget(candidatePath, nil, true)
 	if err != nil {
 		if errors.Is(err, fs.ErrNotExist) {
-			return nil, nil
+			return selectedImportCandidate{}, true, nil
 		}
-		return nil, []Diagnostic{globFilesystemDiagnostic(err, "glob directory could not be canonicalized", location)}
+		return selectedImportCandidate{}, false, []Diagnostic{globCandidateDiagnostic(err, site.location, "glob match could not be inspected")}
 	}
-	if !s.access.contains(canonical) {
-		return nil, []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "glob directory resolves outside the configuration root", &location)}
+	canonical, err := access.canonical(candidatePath)
+	if err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return selectedImportCandidate{}, true, nil
+		}
+		return selectedImportCandidate{}, false, []Diagnostic{globFilesystemDiagnostic(err, "glob match could not be canonicalized", site.location)}
 	}
-	if _, err := s.access.relative(canonical); err != nil {
-		return nil, []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "glob directory could not be addressed inside the configuration root", &location)}
+	if !access.contains(canonical) {
+		return selectedImportCandidate{}, false, []Diagnostic{*importSelectionDiagnostic(site, CodeImportOutsideRoot, "glob traversal resolves outside the configuration root")}
+	}
+	if !last {
+		if !target.dir {
+			return selectedImportCandidate{}, true, nil
+		}
+	}
+	return selectedImportCandidate{
+		path: normalizedImportCandidatePath(access, candidatePath), accessPath: candidatePath, site: site,
+	}, false, nil
+}
+
+func readGlobEntries(access *filesystemAccess, directory string, location Location, limits importLimits, accounting *importAccounting) ([]fs.DirEntry, bool, []Diagnostic) {
+	if _, err := access.importEntryTarget(directory, nil, true); err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil, true, nil
+		}
+		return nil, false, []Diagnostic{globCandidateDiagnostic(err, location, "glob directory could not be inspected")}
+	}
+	canonical, err := access.canonical(directory)
+	if err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil, true, nil
+		}
+		return nil, false, []Diagnostic{globFilesystemDiagnostic(err, "glob directory could not be canonicalized", location)}
+	}
+	if !access.contains(canonical) {
+		return nil, false, []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "glob directory resolves outside the configuration root", &location)}
+	}
+	if _, err := access.relative(canonical); err != nil {
+		return nil, false, []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "glob directory could not be addressed inside the configuration root", &location)}
+	}
+	entries := make([]fs.DirEntry, 0)
+	err = visitBoundedImportDirectory(access, directory, limits.entries, accounting, func(entry fs.DirEntry) error {
+		entries = append(entries, entry)
+		return nil
+	})
+	if errors.Is(err, errImportDirectoryEntryLimit) {
+		return nil, false, []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "glob traversal exceeds the directory-entry visit limit", &location)}
+	}
+	if errors.Is(err, fs.ErrNotExist) {
+		return nil, true, nil
+	}
+	if err != nil {
+		return nil, false, []Diagnostic{globFilesystemDiagnostic(err, "glob directory could not be traversed", location)}
+	}
+	sort.Slice(entries, func(left, right int) bool { return entries[left].Name() < entries[right].Name() })
+	return entries, false, nil
+}
+
+func globCandidateDiagnostic(err error, location Location, message string) Diagnostic {
+	if errors.Is(err, errImportDirectoryLinkCycle) {
+		return importDiagnostic(CodeImportCycle, "directory traversal cycle detected", &location)
+	}
+	if errors.Is(err, errImportPathOutsideRoot) {
+		return importDiagnostic(CodeImportOutsideRoot, "glob traversal resolves outside the configuration root", &location)
+	}
+	return globFilesystemDiagnostic(err, message, location)
+}
+
+func candidatesFromMap(candidates map[string]selectedImportCandidate) []selectedImportCandidate {
+	paths := make([]string, 0, len(candidates))
+	for candidatePath := range candidates {
+		paths = append(paths, candidatePath)
+	}
+	slices.Sort(paths)
+	result := make([]selectedImportCandidate, 0, len(paths))
+	for _, candidatePath := range paths {
+		result = append(result, candidates[candidatePath])
+	}
+	return result
+}
+
+func (s *importTraversal) readGlobMatches(directory, pattern string, location Location) ([]string, []Diagnostic) {
+	entries, missing, diagnostics := readGlobEntries(s.access, directory, location, s.limits, s.accounting)
+	if len(diagnostics) != 0 {
+		return nil, diagnostics
+	}
+	if missing {
+		return nil, nil
 	}
 	var matches []string
-	err = visitBoundedImportDirectory(s.access, canonical, s.limits.entries, s.accounting, func(entry fs.DirEntry) error {
+	for _, entry := range entries {
 		matched, matchErr := path.Match(pattern, entry.Name())
 		if matchErr != nil {
-			return errInvalidImportGlobPattern
+			return nil, []Diagnostic{importDiagnostic(CodeImportInvalid, "import glob pattern is invalid", &location)}
 		}
 		if matched {
 			matches = append(matches, entry.Name())
 		}
-		return nil
-	})
-	if errors.Is(err, errImportDirectoryEntryLimit) {
-		return nil, []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "glob traversal exceeds the directory-entry visit limit", &location)}
 	}
-	if errors.Is(err, errInvalidImportGlobPattern) {
-		return nil, []Diagnostic{importDiagnostic(CodeImportInvalid, "import glob pattern is invalid", &location)}
-	}
-	if errors.Is(err, fs.ErrNotExist) {
-		return nil, nil
-	}
-	if err != nil {
-		return nil, []Diagnostic{globFilesystemDiagnostic(err, "glob directory could not be traversed", location)}
-	}
-	slices.Sort(matches)
 	return matches, nil
 }
 
diff --git a/internal/engine/glob_selection_test.go b/internal/engine/glob_selection_test.go
new file mode 100644
index 0000000..3e73ef1
--- /dev/null
+++ b/internal/engine/glob_selection_test.go
@@ -0,0 +1,203 @@
+package engine
+
+import (
+	"io/fs"
+	"os"
+	"path/filepath"
+	"testing"
+	"testing/fstest"
+)
+
+func TestSelectGlobFiltersOnlyWildcardDiscoveredEntries(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	for name, content := range map[string]string{
+		".hidden/secret.yaml": "value: hidden\n",
+		"visible/item.yaml":   "value: visible\n",
+		"visible/source.txt":  "value: source\n",
+	} {
+		filename := filepath.Join(root, filepath.FromSlash(name))
+		if err := os.MkdirAll(filepath.Dir(filename), 0o700); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(filename, []byte(content), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for alias, target := range map[string]string{
+		"visible-alias":  "visible",
+		"selected.yaml":  "visible/source.txt",
+		"unselected.txt": "visible/item.yaml",
+		".secret.yaml":   "visible/source.txt",
+	} {
+		if err := os.Symlink(target, filepath.Join(root, filepath.FromSlash(alias))); err != nil {
+			t.Skipf("symlinks are unavailable: %v", err)
+		}
+	}
+	access, err := openHostAccess(root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer access.close()
+
+	selection := selectGlobForTest(t, access, root, `    - glob: '*/*.yaml'
+      follow: false
+      hidden: false
+`)
+	if got, want := selectedPathsRelativeTo(selection, root), []string{"visible/item.yaml"}; !equalStrings(got, want) {
+		t.Fatalf("filtered glob candidates = %#v, want %#v", got, want)
+	}
+
+	selection = selectGlobForTest(t, access, root, `    - glob: '.hidden/*.yaml'
+      follow: false
+      hidden: false
+`)
+	if got, want := selectedPathsRelativeTo(selection, root), []string{".hidden/secret.yaml"}; !equalStrings(got, want) {
+		t.Fatalf("literal hidden component candidates = %#v, want %#v", got, want)
+	}
+
+	selection = selectGlobForTest(t, access, root, `    - glob: 'visible-alias/*.yaml'
+      follow: false
+      hidden: false
+`)
+	if got, want := selectedPathsRelativeTo(selection, root), []string{"visible-alias/item.yaml"}; !equalStrings(got, want) {
+		t.Fatalf("literal symlink component candidates = %#v, want %#v", got, want)
+	}
+
+	selection = selectGlobForTest(t, access, root, `    - glob: '*.yaml'
+      follow: false
+      hidden: true
+`)
+	if len(selection.candidates) != 0 {
+		t.Fatalf("follow:false selected a discovered file symlink: %#v", selection.candidates)
+	}
+
+	selection = selectGlobForTest(t, access, root, `    - glob: '*.yaml'
+      follow: true
+      hidden: false
+`)
+	if got, want := selectedPathsRelativeTo(selection, root), []string{"selected.yaml"}; !equalStrings(got, want) {
+		t.Fatalf("filename aliases were not filtered by candidate spelling: %#v, want %#v", got, want)
+	}
+}
+
+func TestSelectGlobAppliesExclusionsBeforeSourceAdmissionWithoutOpeningFiles(t *testing.T) {
+	t.Parallel()
+
+	filesystem := &observedGlobFilesystem{files: fstest.MapFS{
+		"parts":                 &fstest.MapFile{Mode: fs.ModeDir | 0o700},
+		"parts/keep.yaml":       &fstest.MapFile{Data: []byte("not parsed: [\n")},
+		"parts/skip-one.yaml":   &fstest.MapFile{Data: []byte("not parsed: [\n")},
+		"parts/skip-two.yaml":   &fstest.MapFile{Data: []byte("not parsed: [\n")},
+		"parts/skip-three.yaml": &fstest.MapFile{Data: []byte("not parsed: [\n")},
+	}}
+	access := newNativeAccess(filesystem, ".", &namespaceToken{})
+	specs, diagnostics := parseImportSpecsYAML(t, `    - glob: 'parts/*.yaml'
+      exclude: 'skip-*.yaml'
+`)
+	if len(diagnostics) != 0 {
+		t.Fatal(diagnostics)
+	}
+	policy, diagnostics := resolveImportPolicy(specs[0], importDefaults{})
+	if len(diagnostics) != 0 {
+		t.Fatal(diagnostics)
+	}
+	limits := defaultImportLimits()
+	limits.sources = 2
+	accounting := &importAccounting{sources: 1}
+	selection, diagnostics := selectGlobImportCandidates(access, ".", specs[0], policy, limits, accounting)
+	if len(diagnostics) != 0 || !equalStrings(selectedPaths(selection), []string{"parts/keep.yaml"}) {
+		t.Fatalf("post-exclusion selection, diagnostics = %#v, %#v", selection, diagnostics)
+	}
+	if len(filesystem.opened) != 1 || filesystem.opened[0] != "parts" {
+		t.Fatalf("glob opened paths = %#v, want only the parts directory", filesystem.opened)
+	}
+	if accounting.sources+len(selection.candidates) > limits.sources {
+		t.Fatalf("final candidate set exceeds the private source limit: accounting=%#v selection=%#v", accounting, selection.candidates)
+	}
+}
+
+func TestSelectGlobRejectsEscapingWildcardLinks(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	outside := t.TempDir()
+	if err := os.WriteFile(filepath.Join(outside, "file.yaml"), []byte("value: outside\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Symlink(outside, filepath.Join(root, "escape")); err != nil {
+		t.Skipf("symlinks are unavailable: %v", err)
+	}
+	access, err := openHostAccess(root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer access.close()
+
+	_, diagnostics := selectGlobForTestWithDiagnostics(t, access, root, `    - glob: '*/file.yaml'
+      follow: true
+`)
+	if len(diagnostics) != 1 || diagnostics[0].Code != CodeImportOutsideRoot {
+		t.Fatalf("escaping glob diagnostics = %#v, want import.outside_root", diagnostics)
+	}
+	selection := selectGlobForTest(t, access, root, `    - glob: '*/file.yaml'
+      follow: false
+`)
+	if len(selection.candidates) != 0 {
+		t.Fatalf("follow:false selected an escaping alias: %#v", selection.candidates)
+	}
+}
+
+func TestSelectGlobPreservesMatchedDirectoryValidityBeforeExclusion(t *testing.T) {
+	t.Parallel()
+
+	filesystem := fstest.MapFS{
+		"parts":                &fstest.MapFile{Mode: fs.ModeDir | 0o700},
+		"parts/matched.yaml":   &fstest.MapFile{Mode: fs.ModeDir | 0o700},
+		"parts/matched.yaml/x": &fstest.MapFile{Data: []byte("value: x\n")},
+	}
+	access := newNativeAccess(filesystem, ".", &namespaceToken{})
+	_, diagnostics := selectGlobForTestWithDiagnostics(t, access, ".", `    - glob: 'parts/*.yaml'
+      exclude: '*.yaml'
+`)
+	if len(diagnostics) != 1 || diagnostics[0].Code != CodeSourceNotRegular {
+		t.Fatalf("excluded directory diagnostics = %#v, want source.not_regular", diagnostics)
+	}
+}
+
+func selectGlobForTest(t *testing.T, access *filesystemAccess, base, declaration string) importSelection {
+	t.Helper()
+	selection, diagnostics := selectGlobForTestWithDiagnostics(t, access, base, declaration)
+	if len(diagnostics) != 0 {
+		t.Fatalf("glob selection diagnostics = %#v", diagnostics)
+	}
+	return selection
+}
+
+func selectGlobForTestWithDiagnostics(t *testing.T, access *filesystemAccess, base, declaration string) (importSelection, []Diagnostic) {
+	t.Helper()
+	specs, diagnostics := parseImportSpecsYAML(t, declaration)
+	if len(diagnostics) != 0 {
+		t.Fatalf("parse diagnostics = %#v", diagnostics)
+	}
+	policy, diagnostics := resolveImportPolicy(specs[0], importDefaults{})
+	if len(diagnostics) != 0 {
+		t.Fatalf("policy diagnostics = %#v", diagnostics)
+	}
+	return selectGlobImportCandidates(access, base, specs[0], policy, defaultImportLimits(), &importAccounting{})
+}
+
+type observedGlobFilesystem struct {
+	files  fstest.MapFS
+	opened []string
+}
+
+func (filesystem *observedGlobFilesystem) Open(name string) (fs.File, error) {
+	filesystem.opened = append(filesystem.opened, name)
+	return filesystem.files.Open(name)
+}
+
+func (filesystem *observedGlobFilesystem) Stat(name string) (fs.FileInfo, error) {
+	return fs.Stat(filesystem.files, name)
+}