diff --git a/.github/workflows/push.yml b/.github/workflows/push.yml
index b4c5238326b..5e6988d3ca0 100644
--- a/.github/workflows/push.yml
+++ b/.github/workflows/push.yml
@@ -602,8 +602,9 @@ jobs:
             -f commit_sha=${{ github.event.pull_request.head.sha }}
 
       # Dispatch the nightly workflow; as a workflow_dispatch event it runs
-      # integration-short (CloudSlow still skips via -short, but locally-covered
-      # tests run against real clouds, unlike the PR task's skiplocal variant).
+      # integration-short (CloudSlow still skips via -short, but the whole acceptance
+      # suite runs against real clouds, unlike the PR task which only runs the
+      # acceptance tests the PR touches).
       - name: Trigger integration tests (push to main)
         if: ${{ github.event_name == 'push' }}
         env:
diff --git a/Taskfile.yml b/Taskfile.yml
index 0a2030181f6..a82c83689b6 100644
--- a/Taskfile.yml
+++ b/Taskfile.yml
@@ -550,6 +550,15 @@ tasks:
           --packages ./acceptance/... \
           -- -timeout=${LOCAL_TIMEOUT:-60m}{{if .ACCEPTANCE_TEST_FILTER}} -run "{{.ACCEPTANCE_TEST_FILTER}}"{{end}}
 
+  local-select:
+    desc: Run selection of local tests (tests changed in this branch)
+    # Local counterpart of cloud-select: the same DATABRICKS_TEST_SELECT_CHANGED selection,
+    # against the testserver instead of a real workspace. -v to show the selection summary
+    # and the tests that ran; no sources/generates because the selection depends on the git
+    # diff, which Task cannot checksum.
+    cmds:
+      - "DATABRICKS_TEST_SELECT_CHANGED=${DATABRICKS_TEST_SELECT_CHANGED:-50} go test ./acceptance -run '^TestAccept$' -v -timeout=${LOCAL_TIMEOUT:-60m}"
+
   test-update:
     desc: Update acceptance test output (local)
     # Excludes out* because the task rewrites them; keeping them in sources would
@@ -759,12 +768,16 @@ tasks:
           --packages "./acceptance ./integration/..." \
           -- -parallel 4 -timeout=12h -short
 
-  integration-short-skiplocal:
-    desc: Run integration tests for PRs (skips CloudSlow and tests with testserver coverage)
+  cloud-select:
+    desc: Run selection of integration tests (tests changed in this branch)
+    # The alias is the name the eng-dev-ecosystem CI job dispatches (cli-isolated-pr.yml
+    # passes it as integration_task, and cli-isolated-tests.yml greps Taskfile.yml for it
+    # and falls back to integration-short, the full cloud suite, if it is missing).
+    aliases: [integration-short-skiplocal]
     deps: [install-pythons]
     cmds:
       - |
-        DATABRICKS_TEST_SKIPLOCAL=withchanged VERBOSE_TEST=1 \
+        DATABRICKS_TEST_SELECT_CHANGED=50 VERBOSE_TEST=1 \
         go run -modfile=tools/go.mod ./tools/testrunner/main.go \
           {{.GO_TOOL}} gotestsum \
           --format github-actions \
diff --git a/acceptance/acceptance_test.go b/acceptance/acceptance_test.go
index 16fd53a4785..12764d90150 100644
--- a/acceptance/acceptance_test.go
+++ b/acceptance/acceptance_test.go
@@ -30,6 +30,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/databricks/cli/acceptance/internal"
+	"github.com/databricks/cli/acceptance/internal/selection"
 	"github.com/databricks/cli/internal/build"
 	"github.com/databricks/cli/internal/testutil"
 	"github.com/databricks/cli/libs/auth"
@@ -89,7 +90,7 @@ func init() {
 }
 
 const (
-	EntryPointScript = "script"
+	EntryPointScript = selection.EntryPointScript
 	CleanupScript    = "script.cleanup"
 	PrepareScript    = "script.prepare"
 	MaxFileSize      = 1_000_000
@@ -141,6 +142,11 @@ func TestInprocessMode(t *testing.T) {
 	if os.Getenv("CLOUD_ENV") != "" {
 		t.Skip("No need to run this as integration test.")
 	}
+	if os.Getenv(selection.EnvVar) != "" {
+		// The two selftests below only run if this branch changed them, so the
+		// assertions on the returned count do not hold under test selection.
+		t.Skip("Disabled via " + selection.EnvVar)
+	}
 
 	// Uncomment to load  ~/.databricks/debug-env.json to debug integration tests
 	// testutil.LoadDebugEnvIfRunFromIDE(t, "workspace")
@@ -445,25 +451,39 @@ func testAccept(t *testing.T, inprocessMode bool, singleTest string) int {
 		testDirsSet[d] = true
 	}
 
-	skipLocalMode := os.Getenv(SkipLocalEnvVar)
 	subset := newSubsetSelector(t, testdiff.OverwriteMode, Forcerun)
 
-	switch skipLocalMode {
-	case "", SkipLocalWithChanged:
-	default:
-		t.Fatalf("Unsupported %s=%q, expected %q", SkipLocalEnvVar, skipLocalMode, SkipLocalWithChanged)
+	changedLimit, err := selection.ParseLimit(os.Getenv(selection.EnvVar))
+	require.NoError(t, err)
+	selectChanged := changedLimit > 0
+	if !selectChanged && subset.enabled {
+		changedLimit = subsetChangedLimit
 	}
-	skipLocalWithChanged := skipLocalMode == SkipLocalWithChanged
 
-	// changedTests maps test dir to extra env filters for added/modified tests; nil
-	// filters means all variants of that dir changed. Both SkipLocalWithChanged and the
-	// subset selector keep these tests, so detect them at most once here.
+	// changedTests maps test dir to extra env filters for changed tests; nil filters
+	// means all variants of that dir changed. Both selection.EnvVar and the subset
+	// selector keep these tests, so detect them at most once here.
 	var changedTests map[string][]string
-	if skipLocalWithChanged || subset.enabled {
-		changedTests = selectChangedLocalTests(t, testDirsSet)
+	if changedLimit > 0 {
+		// A failed selection (e.g. no origin/main in a shallow checkout) must fail the
+		// run: treating it as "nothing changed" would silently skip new tests.
+		result, err := selection.FromGit(testDirsSet, changedLimit)
+		require.NoError(t, err)
+		t.Log(result.Summary())
+		changedTests = result.Tests()
 	}
 	subset.changed = changedTests
 
+	// Drop the tests that were not selected instead of skipping them per dir: a skip
+	// per dir buries the run in a thousand SKIP lines and hides the selection summary.
+	// Their out.test.toml is left alone, which is what a partial run should do.
+	if selectChanged {
+		testDirs = slices.DeleteFunc(testDirs, func(dir string) bool {
+			_, ok := changedTests[dir]
+			return !ok
+		})
+	}
+
 	if singleTest != "" {
 		testDirs = slices.DeleteFunc(testDirs, func(n string) bool {
 			return n != singleTest
@@ -518,7 +538,7 @@ func testAccept(t *testing.T, inprocessMode bool, singleTest string) int {
 				t.Skip("Skipping test execution (only regenerating out.test.toml)")
 			}
 
-			skipReason := getSkipReason(&config, configPath, dir, skipLocalMode, changedTests)
+			skipReason := getSkipReason(&config, configPath)
 			if skipReason != "" {
 				skippedDirs += 1
 				t.Skip(skipReason)
@@ -572,9 +592,9 @@ func testAccept(t *testing.T, inprocessMode bool, singleTest string) int {
 						if runParallel {
 							t.Parallel()
 						}
-						// Under SkipLocalWithChanged, an invariant dir re-enabled by a
+						// Under selection.EnvVar, an invariant dir re-enabled by a
 						// specific config change runs only its matching variants.
-						if skipLocalWithChanged {
+						if selectChanged {
 							if variantFilters := changedTests[dir]; variantFilters != nil {
 								checkEnvFilters(t, envset, variantFilters)
 							}
@@ -625,23 +645,9 @@ func getEnvFilters(t *testing.T) []string {
 }
 
 func getTests(t *testing.T) []string {
-	testDirs := make([]string, 0, 128)
-
-	err := filepath.Walk(".", func(path string, info os.FileInfo, err error) error {
-		if err != nil {
-			return err
-		}
-		name := filepath.Base(path)
-		if name == EntryPointScript {
-			// Presence of 'script' marks a test case in this directory
-			testName := filepath.ToSlash(filepath.Dir(path))
-			testDirs = append(testDirs, testName)
-		}
-		return nil
-	})
+	// Tests are discovered relative to the acceptance dir, which is the working directory.
+	testDirs, err := selection.FindTestDirs(".")
 	require.NoError(t, err)
-
-	slices.Sort(testDirs)
 	return testDirs
 }
 
@@ -654,15 +660,7 @@ func validateTestPhase(phase int) error {
 }
 
 // Return a reason to skip the test. Empty string means "don't skip".
-// skipLocalMode is the value of DATABRICKS_TEST_SKIPLOCAL read once at startup.
-// changedTests maps test dirs to extra env filters; nil map means feature is off.
-func getSkipReason(config *internal.TestConfig, configPath, dir, skipLocalMode string, changedTests map[string][]string) string {
-	if skipLocalMode == SkipLocalWithChanged {
-		if _, ok := changedTests[dir]; !ok {
-			return "Disabled via DATABRICKS_TEST_SKIPLOCAL=" + SkipLocalWithChanged + " in " + configPath
-		}
-	}
-
+func getSkipReason(config *internal.TestConfig, configPath string) string {
 	if Forcerun {
 		return ""
 	}
diff --git a/acceptance/internal/selection/cmd/main.go b/acceptance/internal/selection/cmd/main.go
new file mode 100644
index 00000000000..559eb1d952d
--- /dev/null
+++ b/acceptance/internal/selection/cmd/main.go
@@ -0,0 +1,126 @@
+// Command cmd prints the acceptance tests that DATABRICKS_TEST_SELECT_CHANGED would run,
+// either for the current branch or for the paths given as arguments, together with the
+// score each test was picked by. It is a way to see what a change selects without running
+// the suite. Run it from the repo root:
+//
+//	go run ./acceptance/internal/selection/cmd
+//	go run ./acceptance/internal/selection/cmd -limit 5
+//	go run ./acceptance/internal/selection/cmd acceptance/bundle/invariant/configs/job.yml.tmpl
+//
+// A bare path takes the status git reports for it: added for a path git does not track,
+// modified otherwise. Prefix a path to force one ("A:path", "M:path", "R100:old:new").
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+
+	"github.com/databricks/cli/acceptance/internal/selection"
+)
+
+func main() {
+	limit := flag.Int("limit", 50, "maximum number of changed tests to select")
+	root := flag.String("root", "acceptance", "path to the acceptance directory")
+	flag.Parse()
+
+	if *limit <= 0 {
+		fatalf("-limit must be a positive integer, got %d", *limit)
+	}
+
+	dirs, err := selection.FindTestDirs(*root)
+	if err != nil {
+		fatalf("cannot list test dirs in %s: %s", *root, err)
+	}
+	testDirs := make(map[string]bool, len(dirs))
+	for _, dir := range dirs {
+		testDirs[dir] = true
+	}
+
+	var result selection.Result
+	if args := flag.Args(); len(args) > 0 {
+		result = selection.FromDiff(diffFromArgs(args), testDirs, *limit)
+	} else {
+		result, err = selection.FromGit(testDirs, *limit)
+		if err != nil {
+			fatalf("%s", err)
+		}
+	}
+
+	fmt.Println(result.Counts())
+	for _, test := range result.Selected {
+		fmt.Printf("  %3d  %s\n", test.Score, test.Name())
+	}
+	if len(result.Selected) == 0 {
+		fmt.Println("       none of the changed files belong to a test dir")
+	}
+}
+
+// diffFromArgs renders command line arguments as `git diff --name-status` lines, so the
+// same selection runs on them as on a real diff.
+func diffFromArgs(args []string) string {
+	changed := changedLines()
+
+	lines := make([]string, 0, len(args))
+	for _, arg := range args {
+		status, rest, ok := strings.Cut(arg, ":")
+		if !ok {
+			status, rest = "", arg
+		}
+		// A rename carries both paths ("R100:old:new"), which git separates by tabs.
+		paths := strings.Split(rest, ":")
+		for i, path := range paths {
+			paths[i] = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(path)), "./")
+		}
+		if status != "" {
+			lines = append(lines, status+"\t"+strings.Join(paths, "\t"))
+			continue
+		}
+		// A path given without a status takes the line git reports for it, kept whole so a
+		// rename keeps both its status and its source path. A path git does not report as
+		// changed stands for a hypothetical change: modified when tracked, added when not.
+		if line, ok := changed[paths[0]]; ok {
+			lines = append(lines, line)
+			continue
+		}
+		lines = append(lines, gitStatus(paths[0])+"\t"+paths[0])
+	}
+	return strings.Join(lines, "\n")
+}
+
+// changedLines maps each path this branch changed to its whole `git diff --name-status`
+// line, keyed the way the diff names it now: the destination path of a rename, and the
+// path itself otherwise. A path filter cannot be used for this, because git only pairs a
+// rename when both of its paths are in the diff — asking about the destination alone
+// reports an addition.
+func changedLines() map[string]string {
+	out, err := exec.Command("git", "diff", "--name-status", "--merge-base", "-M", "origin/main").Output()
+	if err != nil {
+		fatalf("git diff --merge-base origin/main failed: %s", err)
+	}
+
+	lines := map[string]string{}
+	for line := range strings.SplitSeq(strings.TrimSpace(string(out)), "\n") {
+		if fields := strings.Split(line, "\t"); len(fields) >= 2 {
+			lines[fields[len(fields)-1]] = line
+		}
+	}
+	return lines
+}
+
+// gitStatus is the status to assume for a path git does not report as changed: "M" for a
+// tracked path and "A" for one git does not know.
+func gitStatus(path string) string {
+	if exec.Command("git", "ls-files", "--error-unmatch", "--", path).Run() == nil {
+		return "M"
+	}
+	return "A"
+}
+
+func fatalf(format string, args ...any) {
+	fmt.Fprintf(os.Stderr, format+"\n", args...)
+	os.Exit(1)
+}
diff --git a/acceptance/internal/selection/selection.go b/acceptance/internal/selection/selection.go
new file mode 100644
index 00000000000..d3e1d18e4f0
--- /dev/null
+++ b/acceptance/internal/selection/selection.go
@@ -0,0 +1,371 @@
+// Package selection picks the acceptance tests a branch changed, so a run can cover what a
+// PR touches instead of the whole suite. The acceptance harness uses it for
+// DATABRICKS_TEST_SELECT_CHANGED; the cmd subpackage exposes the same selection as a
+// command, to inspect what a given change would run.
+package selection
+
+import (
+	"errors"
+	"fmt"
+	"io/fs"
+	"maps"
+	"os/exec"
+	"path/filepath"
+	"slices"
+	"strconv"
+	"strings"
+)
+
+const (
+	// EnvVar holds the number of changed tests to select. Unset means no selection.
+	EnvVar = "DATABRICKS_TEST_SELECT_CHANGED"
+
+	// EntryPointScript is the file whose presence marks a directory as a test case.
+	EntryPointScript = "script"
+
+	// acceptanceDirPrefix is where the test dirs live relative to the repo root, since
+	// git reports repo-relative paths while test dirs are named relative to acceptance/.
+	acceptanceDirPrefix = "acceptance/"
+
+	invariantConfigsPrefix = acceptanceDirPrefix + "bundle/invariant/configs/"
+	invariantDirPrefix     = "bundle/invariant/"
+)
+
+// Test is one test the selection picked.
+type Test struct {
+	// Dir is the test dir, relative to acceptance/.
+	Dir string
+
+	// Filters restricts the run to the variants matching these KEY=value filters. A nil
+	// slice means every variant of the dir runs.
+	Filters []string
+
+	// Score is why the test was picked; see the score constants.
+	Score int
+}
+
+// Name is the test dir with its variant filters, as the log and the command print it.
+func (t Test) Name() string {
+	if t.Filters == nil {
+		return t.Dir
+	}
+	return t.Dir + "[" + strings.Join(t.Filters, ",") + "]"
+}
+
+// Result is the outcome of a selection.
+type Result struct {
+	// Selected lists the picked tests, highest score first and alphabetical within a
+	// score.
+	Selected []Test
+
+	// Dropped counts the changed tests that did not fit Limit.
+	Dropped int
+
+	// Limit is the cap this selection was made with.
+	Limit int
+}
+
+// Tests maps each selected test dir to the variant filters it runs with, the form the
+// acceptance harness looks tests up by.
+func (r Result) Tests() map[string][]string {
+	tests := make(map[string][]string, len(r.Selected))
+	for _, test := range r.Selected {
+		tests[test.Dir] = test.Filters
+	}
+	return tests
+}
+
+// Counts says how many tests were selected and how many the limit cut.
+func (r Result) Counts() string {
+	return fmt.Sprintf("Selected %d changed tests (limit=%d, %d not selected)", len(r.Selected), r.Limit, r.Dropped)
+}
+
+// Summary is the whole outcome on one line, for the test log.
+func (r Result) Summary() string {
+	names := make([]string, 0, len(r.Selected))
+	for _, test := range r.Selected {
+		names = append(names, test.Name())
+	}
+	return r.Counts() + ": " + strings.Join(names, " ")
+}
+
+// ParseLimit reads the number of tests to select from a raw EnvVar value. An empty value
+// yields 0, which means no selection.
+func ParseLimit(raw string) (int, error) {
+	if raw == "" {
+		return 0, nil
+	}
+
+	limit, err := strconv.Atoi(raw)
+	if err != nil || limit <= 0 {
+		return 0, fmt.Errorf("invalid %s=%q, expected a positive integer", EnvVar, raw)
+	}
+
+	return limit, nil
+}
+
+// FindTestDirs returns every test dir under root, named relative to root with forward
+// slashes, sorted.
+func FindTestDirs(root string) ([]string, error) {
+	var dirs []string
+
+	err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if entry.IsDir() || entry.Name() != EntryPointScript {
+			return nil
+		}
+		rel, err := filepath.Rel(root, filepath.Dir(path))
+		if err != nil {
+			return err
+		}
+		dirs = append(dirs, filepath.ToSlash(rel))
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	slices.Sort(dirs)
+	return dirs, nil
+}
+
+// FromGit selects among testDirs the tests this branch changed, at most limit of them.
+//
+// --merge-base diffs the working tree against the merge base of HEAD and
+// origin/main. This covers committed, staged, and unstaged changes alike —
+// the working tree reflects all three. Untracked files (not yet git-added)
+// are not visible to git diff and will not be re-enabled until staged or
+// committed. The three-dot form origin/main...HEAD only covers committed
+// changes and misses unstaged edits, which breaks the "touch a config, run
+// the test" local dev workflow (same reason lintdiff.py uses --merge-base).
+func FromGit(testDirs map[string]bool, limit int) (Result, error) {
+	out, err := exec.Command("git", "diff", "--name-status", "--merge-base", "-M", "origin/main").Output()
+	if err != nil {
+		// A failed diff (most commonly a missing origin/main in a shallow CI checkout)
+		// must not be silently treated as "nothing changed": that disables change
+		// detection and lets newly added tests skip. Every caller (push.yml PR cells,
+		// integration runs) fetches origin/main, so the caller should fail loudly.
+		if exitErr, ok := errors.AsType[*exec.ExitError](err); ok {
+			if stderr := strings.TrimSpace(string(exitErr.Stderr)); stderr != "" {
+				return Result{}, fmt.Errorf("git diff --merge-base origin/main failed: %w: %s", err, stderr)
+			}
+		}
+		return Result{}, fmt.Errorf("git diff --merge-base origin/main failed: %w", err)
+	}
+
+	return FromDiff(strings.TrimSpace(string(out)), testDirs, limit), nil
+}
+
+// testDirForFile maps a repo-relative changed file (e.g. acceptance/bundle/foo/script)
+// to its owning test dir relative to acceptance/ (e.g. bundle/foo), or "" if the file
+// is outside acceptance/ or not under any known test dir.
+func testDirForFile(repoRelPath string, testDirs map[string]bool) string {
+	parts := strings.Split(filepath.ToSlash(repoRelPath), "/")
+	if len(parts) < 2 || parts[0]+"/" != acceptanceDirPrefix {
+		return ""
+	}
+	// Longest ancestor first so nested tests map to the innermost test dir.
+	for depth := len(parts); depth > 1; depth-- {
+		candidate := strings.Join(parts[1:depth], "/")
+		if testDirs[candidate] {
+			return candidate
+		}
+	}
+	return ""
+}
+
+// changedDir records how one test dir changed and which of its variants should run.
+type changedDir struct {
+	// filters restricts the run to the variants matching these KEY=value filters.
+	// Empty means every variant of the dir runs.
+	filters []string
+
+	// allVariants is set by a change to the dir itself, as opposed to a change to an
+	// invariant config the dir is generated from. It clears filters and keeps a later
+	// config change from narrowing the dir back down to one config.
+	allVariants bool
+
+	// newTest is set when the test itself is new: the dir's script is new, or a new
+	// invariant config adds a variant of the dir. moved is set when the script arrived as
+	// a rename, so the test only changed location. score treats them as exclusive.
+	newTest bool
+	moved   bool
+
+	// fixture is set when a file the test is made of changed, generated when a file the
+	// test produces changed (out*). A dir with only generated changes is one whose golden
+	// output was regenerated.
+	fixture   bool
+	generated bool
+}
+
+// The cap takes the highest scoring dirs, and the scores add up, so a dir that changed in
+// several ways outranks one that changed in a single way. A new test is the most likely to
+// be broken, while a dir where nothing but the golden output changed scores lowest: that
+// usually follows a change elsewhere in the tree and lands on hundreds of dirs at once,
+// which would otherwise fill the quota with tests this branch never edited.
+const (
+	scoreNewTest   = 5
+	scoreChange    = 5
+	scoreGenerated = 1
+	scoreMoved     = 1
+)
+
+func (d *changedDir) score() int {
+	score := 0
+	switch {
+	case d.newTest:
+		score += scoreNewTest
+	case d.moved:
+		// The files of a moved dir all arrive as renames. Moving a test does not change
+		// what it does, so those renames do not also count as changes. A dir that is new
+		// and moved at once (a renamed invariant dir picking up a new config) is scored as
+		// new, since being new says more about it than the move does.
+		return scoreMoved
+	}
+	if d.fixture {
+		score += scoreChange
+	}
+	if d.generated {
+		score += scoreGenerated
+	}
+	return score
+}
+
+// changedDirs maps a test dir, relative to acceptance/, to how it changed.
+type changedDirs map[string]*changedDir
+
+func (c changedDirs) get(dir string) *changedDir {
+	if d, ok := c[dir]; ok {
+		return d
+	}
+	d := &changedDir{}
+	c[dir] = d
+	return d
+}
+
+// invariantConfigName returns the config a changed file under acceptance/bundle/invariant/
+// configs/ belongs to (job.yml.tmpl for both job.yml.tmpl and job.yml.tmpl-init.sh), or ""
+// for any other path.
+func invariantConfigName(path string) string {
+	if !strings.HasPrefix(path, invariantConfigsPrefix) {
+		return ""
+	}
+	name := strings.TrimPrefix(path, invariantConfigsPrefix)
+	// Strip -init.sh / -cleanup.sh suffixes to get the base config name.
+	if i := strings.Index(name, "-"); i > 0 && strings.HasSuffix(name, ".sh") {
+		name = name[:i]
+	}
+	if !strings.HasSuffix(name, ".yml.tmpl") {
+		return ""
+	}
+	return name
+}
+
+// isGeneratedFile reports whether path is a file the test generates rather than a fixture
+// the test is made of. A file counts as generated when its path relative to the test dir
+// starts with "out" (output.txt, out.requests.txt, out.test.toml) — the same rule the
+// harness uses to split inputs from outputs when it copies a test dir, so a nested file
+// such as subdir/outer.py stays a fixture.
+func isGeneratedFile(path, dir string) bool {
+	return strings.HasPrefix(strings.TrimPrefix(path, acceptanceDirPrefix+dir+"/"), "out")
+}
+
+// FromDiff selects among testDirs the tests touched by `git diff --name-status` output,
+// keeping at most limit of them in the order documented on the score constants.
+func FromDiff(diff string, testDirs map[string]bool, limit int) Result {
+	dirs := changedDirs{}
+
+	for line := range strings.SplitSeq(diff, "\n") {
+		// A rename line carries both paths ("R100\told\tnew"); the last field is the
+		// path that exists now.
+		fields := strings.Split(line, "\t")
+		if len(fields) < 2 {
+			continue
+		}
+		status := fields[0]
+		path := fields[len(fields)-1]
+
+		// A changed invariant config re-enables every invariant subdir, but only for the
+		// variants generated from that config.
+		if configName := invariantConfigName(path); configName != "" {
+			for dir := range testDirs {
+				if !strings.HasPrefix(dir, invariantDirPrefix) {
+					continue
+				}
+				d := dirs.get(dir)
+				// The config is the fixture these dirs are generated from, and a new
+				// config adds a variant of each of them. Its -init.sh / -cleanup.sh
+				// companions change how an existing variant runs, so only the config
+				// itself counts as a new test.
+				d.fixture = true
+				if status == "A" && strings.HasSuffix(path, configName) {
+					d.newTest = true
+				}
+				filter := "INPUT_CONFIG=" + configName
+				switch {
+				case d.allVariants:
+					// A change to the dir itself already runs every variant.
+				case len(d.filters) == 0:
+					d.filters = []string{filter}
+				case d.filters[0] != filter:
+					// The harness requires every filter to match (see checkEnvFilters), so
+					// two different INPUT_CONFIG values would skip every variant and the
+					// dir would run nothing. Run all of its variants instead.
+					d.allVariants = true
+					d.filters = nil
+				}
+			}
+			continue
+		}
+
+		// test.toml and out.test.toml under the invariant tree regenerate
+		// automatically when INPUT_CONFIG changes; ignore them so they don't
+		// unlock all variants of every invariant subdir.
+		if strings.HasPrefix(path, acceptanceDirPrefix+invariantDirPrefix) {
+			if name := filepath.Base(path); name == "test.toml" || name == "out.test.toml" {
+				continue
+			}
+		}
+
+		dir := testDirForFile(path, testDirs)
+		if dir == "" {
+			continue
+		}
+
+		d := dirs.get(dir)
+		d.allVariants = true
+		d.filters = nil
+		if isGeneratedFile(path, dir) {
+			d.generated = true
+		} else {
+			d.fixture = true
+		}
+		// The status of the dir's script says how the dir itself changed.
+		if strings.HasSuffix(path, "/script") {
+			switch {
+			case status == "A":
+				d.newTest = true
+			case strings.HasPrefix(status, "R"):
+				d.moved = true
+			}
+		}
+	}
+
+	// Sort by name first, then stably by descending score, so dirs that score the same
+	// stay alphabetical.
+	selected := slices.Sorted(maps.Keys(dirs))
+	slices.SortStableFunc(selected, func(a, b string) int {
+		return dirs[b].score() - dirs[a].score()
+	})
+
+	dropped := max(len(selected)-limit, 0)
+	selected = selected[:len(selected)-dropped]
+
+	tests := make([]Test, 0, len(selected))
+	for _, dir := range selected {
+		tests = append(tests, Test{Dir: dir, Filters: dirs[dir].filters, Score: dirs[dir].score()})
+	}
+	return Result{Selected: tests, Dropped: dropped, Limit: limit}
+}
diff --git a/acceptance/internal/selection/selection_test.go b/acceptance/internal/selection/selection_test.go
new file mode 100644
index 00000000000..044f0e2c97a
--- /dev/null
+++ b/acceptance/internal/selection/selection_test.go
@@ -0,0 +1,220 @@
+package selection_test
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/databricks/cli/acceptance/internal/selection"
+)
+
+var testDirs = map[string]bool{
+	"bundle/added":           true,
+	"bundle/modified":        true,
+	"bundle/regenerated":     true,
+	"bundle/moved":           true,
+	"bundle/untouched":       true,
+	"bundle/invariant/jobs":  true,
+	"bundle/invariant/apps":  true,
+	"cmd/sync/nested":        true,
+	"cmd/sync/nested/deeper": true,
+}
+
+// diffLines joins name-status records the way `git diff --name-status` prints them.
+func diffLines(lines ...string) string {
+	return strings.Join(lines, "\n")
+}
+
+func TestFromDiffStatuses(t *testing.T) {
+	diff := diffLines(
+		"A\tacceptance/bundle/added/script",
+		"M\tacceptance/bundle/modified/databricks.yml",
+		"R100\tacceptance/bundle/old/script\tacceptance/bundle/moved/script",
+		"M\tlibs/dyn/value.go",
+	)
+	result := selection.FromDiff(diff, testDirs, 10)
+	assert.Zero(t, result.Dropped)
+	assert.Equal(t, map[string][]string{
+		"bundle/added":    nil,
+		"bundle/modified": nil,
+		"bundle/moved":    nil,
+	}, result.Tests())
+}
+
+func TestFromDiffPriority(t *testing.T) {
+	diff := diffLines(
+		"R090\tacceptance/bundle/old/script\tacceptance/bundle/moved/script",
+		"M\tacceptance/bundle/regenerated/output.txt",
+		"M\tacceptance/bundle/modified/script",
+		"A\tacceptance/bundle/added/script",
+	)
+	// The cap keeps added first, then a changed fixture, then a moved dir, and last a dir
+	// where only the golden output was regenerated.
+	scored := []string{"bundle/added", "bundle/modified", "bundle/moved", "bundle/regenerated"}
+	for limit := 1; limit <= len(scored); limit++ {
+		result := selection.FromDiff(diff, testDirs, limit)
+		assert.Len(t, result.Selected, limit)
+		assert.Equal(t, len(scored)-limit, result.Dropped, "limit=%d", limit)
+		for _, dir := range scored[:limit] {
+			assert.Contains(t, result.Tests(), dir, "limit=%d", limit)
+		}
+	}
+}
+
+func TestFromDiffScores(t *testing.T) {
+	// Scores add up: a new dir counts as new (5) plus its fixtures (5) plus its goldens
+	// (1); a dir whose script and golden both changed counts 5+1. A moved dir scores the
+	// move alone, since the files it brings along are renames rather than changes.
+	diff := diffLines(
+		"A\tacceptance/bundle/added/script",
+		"A\tacceptance/bundle/added/output.txt",
+		"M\tacceptance/bundle/modified/script",
+		"M\tacceptance/bundle/modified/output.txt",
+		"R090\tacceptance/bundle/old/script\tacceptance/bundle/moved/script",
+		"R090\tacceptance/bundle/old/output.txt\tacceptance/bundle/moved/output.txt",
+		"M\tacceptance/bundle/regenerated/output.txt",
+		"M\tacceptance/bundle/untouched/databricks.yml",
+	)
+	result := selection.FromDiff(diff, testDirs, 10)
+	scores := map[string]int{}
+	for _, test := range result.Selected {
+		scores[test.Dir] = test.Score
+	}
+	assert.Equal(t, map[string]int{
+		"bundle/added":       11,
+		"bundle/modified":    6,
+		"bundle/untouched":   5,
+		"bundle/moved":       1,
+		"bundle/regenerated": 1,
+	}, scores)
+}
+
+func TestFromDiffFixtureBeatsOutputInSameDir(t *testing.T) {
+	// A dir with both a fixture and an output change ranks as a fixture change.
+	diff := diffLines(
+		"M\tacceptance/bundle/modified/output.txt",
+		"M\tacceptance/bundle/modified/databricks.yml",
+		"M\tacceptance/bundle/regenerated/out.requests.txt",
+	)
+	result := selection.FromDiff(diff, testDirs, 1)
+	assert.Equal(t, map[string][]string{"bundle/modified": nil}, result.Tests())
+	assert.Equal(t, 1, result.Dropped)
+}
+
+func TestFromDiffNestedFixture(t *testing.T) {
+	// "out" is matched against the path relative to the test dir, so a file in a
+	// subdirectory is a fixture even when its own name starts with "out".
+	diff := diffLines(
+		"M\tacceptance/bundle/modified/subdir/outer.py",
+		"M\tacceptance/bundle/regenerated/output.txt",
+	)
+	result := selection.FromDiff(diff, testDirs, 1)
+	assert.Equal(t, map[string][]string{"bundle/modified": nil}, result.Tests())
+	assert.Equal(t, 1, result.Dropped)
+}
+
+func TestFromDiffInvariantConfigRanksAsFixture(t *testing.T) {
+	// The invariant config is the fixture its dirs are generated from, so it outranks
+	// a dir whose output was regenerated.
+	diff := diffLines(
+		"M\tacceptance/bundle/regenerated/output.txt",
+		"M\tacceptance/bundle/invariant/configs/job.yml.tmpl",
+	)
+	result := selection.FromDiff(diff, testDirs, 2)
+	assert.NotContains(t, result.Tests(), "bundle/regenerated")
+	assert.Equal(t, 1, result.Dropped)
+}
+
+func TestFromDiffNewInvariantConfig(t *testing.T) {
+	// A new invariant config adds a variant of every invariant dir, so it scores as a new
+	// test on top of the fixture change. Changing an existing config, or adding one of its
+	// -init.sh companions, only changes how an existing variant runs.
+	for status, want := range map[string]int{"A": 10, "M": 5} {
+		result := selection.FromDiff(diffLines(status+"\tacceptance/bundle/invariant/configs/job.yml.tmpl"), testDirs, 10)
+		assert.Equal(t, want, result.Selected[0].Score, "status=%s", status)
+	}
+
+	result := selection.FromDiff(diffLines("A\tacceptance/bundle/invariant/configs/job.yml.tmpl-init.sh"), testDirs, 10)
+	assert.Equal(t, 5, result.Selected[0].Score)
+}
+
+func TestFromDiffTwoInvariantConfigs(t *testing.T) {
+	// The harness requires every filter to match, so two INPUT_CONFIG filters would skip
+	// every variant. Two changed configs run all variants of the invariant dirs instead.
+	diff := diffLines(
+		"M\tacceptance/bundle/invariant/configs/job.yml.tmpl",
+		"M\tacceptance/bundle/invariant/configs/pipeline.yml.tmpl",
+	)
+	result := selection.FromDiff(diff, testDirs, 10)
+	assert.Equal(t, map[string][]string{
+		"bundle/invariant/jobs": nil,
+		"bundle/invariant/apps": nil,
+	}, result.Tests())
+}
+
+func TestFromDiffConfigAndItsInitScript(t *testing.T) {
+	// A config and its setup script name the same variant, so the filter is kept.
+	diff := diffLines(
+		"M\tacceptance/bundle/invariant/configs/job.yml.tmpl",
+		"M\tacceptance/bundle/invariant/configs/job.yml.tmpl-init.sh",
+	)
+	result := selection.FromDiff(diff, testDirs, 10)
+	assert.Equal(t, map[string][]string{
+		"bundle/invariant/jobs": {"INPUT_CONFIG=job.yml.tmpl"},
+		"bundle/invariant/apps": {"INPUT_CONFIG=job.yml.tmpl"},
+	}, result.Tests())
+}
+
+func TestFromDiffNewAndMovedAreExclusive(t *testing.T) {
+	// A moved invariant dir that also picks up a new config scores as new, not as new plus
+	// moved.
+	diff := diffLines(
+		"A\tacceptance/bundle/invariant/configs/job.yml.tmpl",
+		"R100\tacceptance/bundle/invariant/old/script\tacceptance/bundle/invariant/jobs/script",
+	)
+	result := selection.FromDiff(diff, testDirs, 10)
+	scores := map[string]int{}
+	for _, test := range result.Selected {
+		scores[test.Dir] = test.Score
+	}
+	assert.Equal(t, 10, scores["bundle/invariant/jobs"])
+}
+
+func TestFromDiffNestedDir(t *testing.T) {
+	// A file maps to the innermost test dir that owns it.
+	diff := diffLines("M\tacceptance/cmd/sync/nested/deeper/script")
+	result := selection.FromDiff(diff, testDirs, 10)
+	assert.Equal(t, map[string][]string{"cmd/sync/nested/deeper": nil}, result.Tests())
+}
+
+func TestFromDiffInvariantConfig(t *testing.T) {
+	// A changed invariant config re-enables every invariant dir, restricted to that config.
+	diff := diffLines("M\tacceptance/bundle/invariant/configs/job.yml.tmpl")
+	result := selection.FromDiff(diff, testDirs, 10)
+	assert.Equal(t, map[string][]string{
+		"bundle/invariant/jobs": {"INPUT_CONFIG=job.yml.tmpl"},
+		"bundle/invariant/apps": {"INPUT_CONFIG=job.yml.tmpl"},
+	}, result.Tests())
+}
+
+func TestFromDiffInvariantConfigAndDir(t *testing.T) {
+	// A non-config change to an invariant dir unlocks all of its variants; the
+	// regenerated test.toml files are ignored.
+	diff := diffLines(
+		"M\tacceptance/bundle/invariant/configs/job.yml.tmpl",
+		"M\tacceptance/bundle/invariant/jobs/script",
+		"M\tacceptance/bundle/invariant/apps/out.test.toml",
+	)
+	result := selection.FromDiff(diff, testDirs, 10)
+	assert.Equal(t, map[string][]string{
+		"bundle/invariant/jobs": nil,
+		"bundle/invariant/apps": {"INPUT_CONFIG=job.yml.tmpl"},
+	}, result.Tests())
+}
+
+func TestFromDiffEmptyDiff(t *testing.T) {
+	result := selection.FromDiff("", testDirs, 10)
+	assert.Empty(t, result.Tests())
+	assert.Zero(t, result.Dropped)
+}
diff --git a/acceptance/skiplocal_test.go b/acceptance/skiplocal_test.go
deleted file mode 100644
index 4f3c7af22da..00000000000
--- a/acceptance/skiplocal_test.go
+++ /dev/null
@@ -1,151 +0,0 @@
-package acceptance_test
-
-import (
-	"errors"
-	"os/exec"
-	"path/filepath"
-	"slices"
-	"strings"
-	"testing"
-)
-
-// Cloud PR runs set DATABRICKS_TEST_SKIPLOCAL=withchanged to skip acceptance
-// tests that already run locally, except those this branch touches.
-const (
-	SkipLocalEnvVar = "DATABRICKS_TEST_SKIPLOCAL"
-
-	SkipLocalWithChanged = "withchanged"
-
-	// Cap re-enabled tests so cloud PR runs stay bounded; prefer added over modified.
-	maxChangedLocalTests = 50
-
-	invariantConfigsPrefix = "acceptance/bundle/invariant/configs/"
-	invariantDirPrefix     = "bundle/invariant/"
-)
-
-// testDirForFile maps a repo-relative changed file (e.g. acceptance/bundle/foo/script)
-// to its owning test dir relative to acceptance/ (e.g. bundle/foo), or "" if the file
-// is outside acceptance/ or not under any known test dir.
-func testDirForFile(repoRelPath string, testDirs map[string]bool) string {
-	parts := strings.Split(filepath.ToSlash(repoRelPath), "/")
-	if len(parts) < 2 || parts[0] != "acceptance" {
-		return ""
-	}
-	// Longest ancestor first so nested tests map to the innermost test dir.
-	for depth := len(parts); depth > 1; depth-- {
-		candidate := strings.Join(parts[1:depth], "/")
-		if testDirs[candidate] {
-			return candidate
-		}
-	}
-	return ""
-}
-
-// selectChangedLocalTests returns a map of test dir → extra env filters for
-// re-enabling under SkipLocalWithChanged. A nil filter slice means all variants
-// of that dir run; a non-nil slice restricts to variants matching those filters
-// (applied by the caller via checkEnvFilters in the variant loop).
-// Added dirs come before modified ones; the total is capped at maxChangedLocalTests.
-//
-// A changed invariant config (acceptance/bundle/invariant/configs/*.yml.tmpl)
-// maps to all invariant subdirs with an INPUT_CONFIG= filter, so touching
-// job.yml.tmpl re-enables all subdirs but only for their job.yml.tmpl variants.
-//
-// --merge-base diffs the working tree against the merge base of HEAD and
-// origin/main. This covers committed, staged, and unstaged changes alike —
-// the working tree reflects all three. Untracked files (not yet git-added)
-// are not visible to git diff and will not be re-enabled until staged or
-// committed. The three-dot form origin/main...HEAD only covers committed
-// changes and misses unstaged edits, which breaks the "touch a config, run
-// the test" local dev workflow (same reason lintdiff.py uses --merge-base).
-func selectChangedLocalTests(t *testing.T, testDirs map[string]bool) map[string][]string {
-	out, err := exec.Command("git", "diff", "--name-status", "--merge-base", "-M", "origin/main").Output()
-	if err != nil {
-		// A failed diff (most commonly a missing origin/main in a shallow CI
-		// checkout) must not be silently treated as "nothing changed": that
-		// disables change detection and lets newly added tests skip. Fail loudly.
-		// Every caller (push.yml PR cells, integration runs) now fetches origin/main.
-		stderr := ""
-		if exitErr, ok := errors.AsType[*exec.ExitError](err); ok {
-			stderr = strings.TrimSpace(string(exitErr.Stderr))
-		}
-		t.Fatalf("git diff --merge-base origin/main failed: %v\n%s", err, stderr)
-	}
-	diff := strings.TrimSpace(string(out))
-
-	// result accumulates dirs with their filters; added tracks brand-new dirs.
-	// nil filter slice = all variants run; non-nil = restricted to those filters.
-	result := map[string][]string{}
-	added := map[string]bool{}
-
-	for line := range strings.SplitSeq(diff, "\n") {
-		fields := strings.Split(line, "\t")
-		if len(fields) < 2 {
-			continue
-		}
-		status := fields[0]
-		path := fields[len(fields)-1]
-
-		// A changed invariant config re-enables all invariant subdirs with an
-		// INPUT_CONFIG filter, unless a subdir was already unlocked by a non-config change.
-		if strings.HasPrefix(path, invariantConfigsPrefix) {
-			configName := path[len(invariantConfigsPrefix):]
-			// Strip -init.sh / -cleanup.sh suffixes to get the base config name.
-			if i := strings.Index(configName, "-"); i > 0 && strings.HasSuffix(configName, ".sh") {
-				configName = configName[:i]
-			}
-			if strings.HasSuffix(configName, ".yml.tmpl") {
-				for dir := range testDirs {
-					if strings.HasPrefix(dir, invariantDirPrefix) {
-						if existing, ok := result[dir]; !ok || existing != nil {
-							result[dir] = append(result[dir], "INPUT_CONFIG="+configName)
-						}
-					}
-				}
-			}
-			continue
-		}
-
-		// test.toml and out.test.toml under the invariant tree regenerate
-		// automatically when INPUT_CONFIG changes; ignore them so they don't
-		// unlock all variants of every invariant subdir.
-		if strings.HasPrefix(path, "acceptance/"+invariantDirPrefix) {
-			if name := filepath.Base(path); name == "test.toml" || name == "out.test.toml" {
-				continue
-			}
-		}
-
-		dir := testDirForFile(path, testDirs)
-		if dir == "" {
-			continue
-		}
-		result[dir] = nil // nil = all variants; overrides any prior config-scoped filter
-		// A script file with status A means the test dir is brand new.
-		// Renames (R) land here as the destination path but are not "added".
-		if status == "A" && strings.HasSuffix(path, "/script") {
-			added[dir] = true
-		}
-	}
-
-	var addedDirs, modifiedDirs []string
-	for dir := range result {
-		if added[dir] {
-			addedDirs = append(addedDirs, dir)
-		} else {
-			modifiedDirs = append(modifiedDirs, dir)
-		}
-	}
-	slices.Sort(addedDirs)
-	slices.Sort(modifiedDirs)
-
-	selected := append(addedDirs, modifiedDirs...)
-	if len(selected) > maxChangedLocalTests {
-		selected = selected[:maxChangedLocalTests]
-	}
-
-	out2 := make(map[string][]string, len(selected))
-	for _, dir := range selected {
-		out2[dir] = result[dir]
-	}
-	return out2
-}
diff --git a/acceptance/subset_test.go b/acceptance/subset_test.go
index 22eda164b3d..77ab97c9b3d 100644
--- a/acceptance/subset_test.go
+++ b/acceptance/subset_test.go
@@ -23,13 +23,19 @@ const (
 	// a new commit reshuffles the subset while a retry of the same commit repeats it.
 	// If unset, a random seed is generated and logged so the run can be reproduced.
 	SubsetSeedEnvVar = "TESTS_SELECT_SUBSET_SEED"
+
+	// subsetChangedLimit caps how many changed tests the subset selector keeps on top of
+	// its hash-selected fraction. Unlike selection.EnvVar it carries no count of its own,
+	// and a PR that edits hundreds of test dirs must not turn the subset cells back into
+	// a full run.
+	subsetChangedLimit = 50
 )
 
 // subsetSelector decides, per subtest, whether it runs under TESTS_SELECT_SUBSET_PCT.
-// A subtest runs if it is an added/modified test on this branch (always kept, reusing
-// the same change detection as SkipLocalWithChanged), or if its seeded hash falls
-// under the percentage. The decision is independent per subtest, so added/modified
-// tests run on top of the hash-selected subset rather than displacing anything.
+// A subtest runs if it is a changed test on this branch (always kept, reusing the same
+// change detection as DATABRICKS_TEST_SELECT_CHANGED), or if its seeded hash falls
+// under the percentage. The decision is independent per subtest, so changed tests run
+// on top of the hash-selected subset rather than displacing anything.
 type subsetSelector struct {
 	enabled bool
 	pct     int
@@ -42,8 +48,8 @@ type subsetSelector struct {
 
 // newSubsetSelector reads the subset env vars. Subsetting is disabled in update mode
 // and under -forcerun so that every output is regenerated and forced runs are honored.
-// The caller assigns .changed (the added/modified tests to always keep) so that the
-// change detection is shared with SkipLocalWithChanged and runs at most once.
+// The caller assigns .changed (the changed tests to always keep) so that the change
+// detection is shared with DATABRICKS_TEST_SELECT_CHANGED and runs at most once.
 func newSubsetSelector(t *testing.T, overwrite, forcerun bool) subsetSelector {
 	raw := os.Getenv(SubsetPctEnvVar)
 	if raw == "" || overwrite || forcerun {
@@ -88,7 +94,7 @@ func (s subsetSelector) skipReason(dir string, envset []string) string {
 	return "Skipped by " + SubsetPctEnvVar
 }
 
-// isChanged reports whether the subtest belongs to an added/modified test dir. For an
+// isChanged reports whether the subtest belongs to a changed test dir. For an
 // invariant dir re-enabled by a specific config change, only the matching variants
 // count as changed.
 func (s subsetSelector) isChanged(dir string, envset []string) bool {