From 4834069a81633e60e01ac7327bd66af7f11da0da Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Tue, 18 Aug 2026 11:21:06 +0200
Subject: [PATCH 01/10] direct: add DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT to cap
 resource waits

Co-authored-by: Isaac
---
 .nextchanges/bundles/resource-max-wait.md     |   1 +
 .../deploy/resource-max-wait/databricks.yml   |  16 ++
 .../deploy/resource-max-wait/out.test.toml    |   2 +
 .../deploy/resource-max-wait/output.txt       |  34 ++++
 .../bundle/deploy/resource-max-wait/script    |  11 ++
 .../bundle/deploy/resource-max-wait/test.toml |   2 +
 bundle/direct/apply.go                        |  12 +-
 bundle/direct/bundle_apply.go                 |  16 ++
 bundle/direct/maxwait.go                      |  88 ++++++++++
 bundle/direct/maxwait_test.go                 | 151 ++++++++++++++++++
 bundle/direct/pkg.go                          |   6 +
 bundle/env/resource_max_wait.go               |  13 ++
 12 files changed, 349 insertions(+), 3 deletions(-)
 create mode 100644 .nextchanges/bundles/resource-max-wait.md
 create mode 100644 acceptance/bundle/deploy/resource-max-wait/databricks.yml
 create mode 100644 acceptance/bundle/deploy/resource-max-wait/out.test.toml
 create mode 100644 acceptance/bundle/deploy/resource-max-wait/output.txt
 create mode 100644 acceptance/bundle/deploy/resource-max-wait/script
 create mode 100644 acceptance/bundle/deploy/resource-max-wait/test.toml
 create mode 100644 bundle/direct/maxwait.go
 create mode 100644 bundle/direct/maxwait_test.go
 create mode 100644 bundle/env/resource_max_wait.go

diff --git a/.nextchanges/bundles/resource-max-wait.md b/.nextchanges/bundles/resource-max-wait.md
new file mode 100644
index 00000000000..e6a7d7e6c47
--- /dev/null
+++ b/.nextchanges/bundles/resource-max-wait.md
@@ -0,0 +1 @@
+Added `DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT` (in seconds) to cap how long `bundle deploy` and `bundle destroy` wait for a resource to become ready or finish deleting. When the cap expires the wait is abandoned with a warning instead of failing — the resource is already recorded in state, so the next deployment reconciles it. Resources that others depend on keep their full wait, since a dependent must not act on a resource that is not ready. Direct engine only.
diff --git a/acceptance/bundle/deploy/resource-max-wait/databricks.yml b/acceptance/bundle/deploy/resource-max-wait/databricks.yml
new file mode 100644
index 00000000000..2d641214acb
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait/databricks.yml
@@ -0,0 +1,16 @@
+bundle:
+  name: test-bundle
+
+# Nothing to sync: keeps the "Files:" line out of the output, which would otherwise
+# count whatever the harness has written into the test directory.
+sync:
+  paths: []
+
+resources:
+  jobs:
+    my_job:
+      name: test-job
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/notebook
diff --git a/acceptance/bundle/deploy/resource-max-wait/out.test.toml b/acceptance/bundle/deploy/resource-max-wait/out.test.toml
new file mode 100644
index 00000000000..0938e678987
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/deploy/resource-max-wait/output.txt b/acceptance/bundle/deploy/resource-max-wait/output.txt
new file mode 100644
index 00000000000..52e43bdde11
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait/output.txt
@@ -0,0 +1,34 @@
+
+=== A malformed value is rejected instead of falling back to the default wait
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=1m musterr [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Error: invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="1m": expected a non-negative number of seconds
+
+Files: 0 uploaded, 0 deleted
+
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=-5 musterr [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Error: invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="-5": expected a non-negative number of seconds
+
+Files: 0 uploaded, 0 deleted
+
+=== A valid value deploys normally
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Created jobs.my_job
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Zero means do not wait at all
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Files: 0 uploaded, 0 deleted
+Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
+
+Destroy: 1 deleted
diff --git a/acceptance/bundle/deploy/resource-max-wait/script b/acceptance/bundle/deploy/resource-max-wait/script
new file mode 100644
index 00000000000..a457a311c6e
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait/script
@@ -0,0 +1,11 @@
+title "A malformed value is rejected instead of falling back to the default wait"
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=1m musterr $CLI bundle deploy
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=-5 musterr $CLI bundle deploy
+
+title "A valid value deploys normally"
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 $CLI bundle deploy
+
+title "Zero means do not wait at all"
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 $CLI bundle deploy
+
+trace $CLI bundle destroy --auto-approve
diff --git a/acceptance/bundle/deploy/resource-max-wait/test.toml b/acceptance/bundle/deploy/resource-max-wait/test.toml
new file mode 100644
index 00000000000..427bdd03f18
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait/test.toml
@@ -0,0 +1,2 @@
+# DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT is read by the direct engine only.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/bundle/direct/apply.go b/bundle/direct/apply.go
index 3ed9d233f60..15986bac362 100644
--- a/bundle/direct/apply.go
+++ b/bundle/direct/apply.go
@@ -80,8 +80,10 @@ func (d *DeploymentUnit) Create(ctx context.Context, db *dstate.DeploymentState,
 		return fmt.Errorf("saving state after creating id=%s: %w", newID, err)
 	}
 
-	waitRemoteState, err := retryOnTransient(ctx, func() (any, error) {
-		return d.Adapter.WaitAfterCreate(ctx, newID, newState)
+	waitRemoteState, err := waitCapped(ctx, d.MaxWait, "creation of "+d.ResourceKey, func(ctx context.Context) (any, error) {
+		return retryOnTransient(ctx, func() (any, error) {
+			return d.Adapter.WaitAfterCreate(ctx, newID, newState)
+		})
 	})
 	if err != nil {
 		return fmt.Errorf("waiting after creating id=%s: %w", newID, err)
@@ -258,7 +260,11 @@ func (d *DeploymentUnit) Delete(ctx context.Context, db *dstate.DeploymentState,
 	// Wait for asynchronous teardown after dropping state. Mirrors Recreate so
 	// the contract is the same regardless of whether the user triggered
 	// `bundle destroy` or a recreate.
-	err = d.Adapter.WaitAfterDelete(ctx, oldID)
+	// The two diverge once MaxWait is set: this wait is capped, Recreate's is not,
+	// because only Recreate needs the name released for the create that follows.
+	_, err = waitCapped(ctx, d.MaxWait, "deletion of "+d.ResourceKey, func(ctx context.Context) (struct{}, error) {
+		return struct{}{}, d.Adapter.WaitAfterDelete(ctx, oldID)
+	})
 	if err != nil {
 		return fmt.Errorf("waiting after deleting id=%s: %w", oldID, err)
 	}
diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index c4178c4e601..12205677f2d 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -9,6 +9,7 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/terraform_dabs_map"
+	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
 	"github.com/databricks/cli/libs/structs/structaccess"
 	"github.com/databricks/cli/libs/structs/structpath"
@@ -20,6 +21,14 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 		panic("Planning is not done")
 	}
 
+	// Read before the early return below so a malformed value is reported even when there is
+	// nothing to deploy.
+	maxWait, err := resourceMaxWait(ctx)
+	if err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+
 	if len(plan.Plan) == 0 {
 		// Avoid creating state file if nothing to deploy
 		return
@@ -70,10 +79,17 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 			return false
 		}
 
+		// g.Adj holds the edges out of this node, i.e. the resources that run after it.
+		unitWait := unitMaxWait(maxWait, action, len(g.Adj[resourceKey]))
+		if maxWait != maxWaitUnset && unitWait == maxWaitUnset {
+			log.Debugf(ctx, "Not capping wait for %s: other resources depend on it", resourceKey)
+		}
+
 		d := &DeploymentUnit{
 			ResourceKey: resourceKey,
 			Adapter:     adapter,
 			DependsOn:   entry.DependsOn,
+			MaxWait:     unitWait,
 		}
 
 		if action == deployplan.Delete {
diff --git a/bundle/direct/maxwait.go b/bundle/direct/maxwait.go
new file mode 100644
index 00000000000..ae58ed042a4
--- /dev/null
+++ b/bundle/direct/maxwait.go
@@ -0,0 +1,88 @@
+package direct
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/databricks/cli/bundle/deployplan"
+	bundleenv "github.com/databricks/cli/bundle/env"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go/retries"
+)
+
+// maxWaitUnset means "no cap configured", which must stay distinguishable from an explicit
+// 0 ("do not wait at all").
+const maxWaitUnset = time.Duration(-1)
+
+// resourceMaxWait returns the cap on waiting for a resource to reach its target state, or
+// maxWaitUnset when the environment variable is absent. Unlike retryInterval, a malformed
+// value is an error rather than a silent fallback: ignoring a typo would restore the
+// multi-hour default wait that the user was trying to shorten.
+func resourceMaxWait(ctx context.Context) (time.Duration, error) {
+	v, ok := bundleenv.ResourceMaxWait(ctx)
+	if !ok {
+		return maxWaitUnset, nil
+	}
+	seconds, err := strconv.Atoi(v)
+	if err != nil || seconds < 0 {
+		return maxWaitUnset, fmt.Errorf("invalid %s=%q: expected a non-negative number of seconds", bundleenv.ResourceMaxWaitVariable, v)
+	}
+	return time.Duration(seconds) * time.Second, nil
+}
+
+// unitMaxWait returns the cap to apply to a single resource, given how many resources run
+// after it in the deployment graph.
+//
+// Deletes are capped regardless of dependents. The trade-off is accepted deliberately: state
+// is dropped before the wait, so a cut-short delete leaves the resource untracked while it is
+// still tearing down, and the dependency deleted after it may then be rejected for still
+// having a child. Recreate's internal delete-wait is excluded structurally — it is the one
+// wait never routed through here, because it releases the name for the following create.
+//
+// Every other action is capped only when nothing depends on this resource, since a dependent
+// would otherwise act on a resource that has not reached its target state.
+func unitMaxWait(maxWait time.Duration, action deployplan.ActionType, dependents int) time.Duration {
+	if action == deployplan.Delete || dependents == 0 {
+		return maxWait
+	}
+	return maxWaitUnset
+}
+
+// waitCapped runs wait under maxWait. When the cap expires the wait is abandoned with a
+// warning instead of failing the deployment: state is written before the wait, so the
+// resource stays tracked and the next plan reconciles it. Genuine failures still propagate,
+// since retries reports those without a timeout error.
+func waitCapped[T any](ctx context.Context, maxWait time.Duration, description string, wait func(context.Context) (T, error)) (T, error) {
+	if maxWait == maxWaitUnset {
+		return wait(ctx)
+	}
+
+	waitCtx, cancel := context.WithTimeout(ctx, maxWait)
+	defer cancel()
+
+	result, err := wait(waitCtx)
+
+	// waitCtx expired but ctx did not: the cap fired rather than the whole deployment being
+	// cancelled, which must keep failing so an interrupt is not swallowed.
+	if err != nil && waitCtx.Err() != nil && ctx.Err() == nil && isWaitTimeout(err) {
+		log.Warnf(ctx, "Stopped waiting for %s after %s (%s); it may still be in progress", description, maxWait, bundleenv.ResourceMaxWaitVariable)
+		var zero T
+		return zero, nil
+	}
+
+	return result, err
+}
+
+// isWaitTimeout reports whether err is a wait that ran out of time rather than a resource
+// that failed. Two shapes reach here: retries.Poll reports a deadline as ErrTimedOut wrapping
+// the last poll message, while retryWith returns a bare context error when the deadline lands
+// while it sleeps between transient-error retries.
+func isWaitTimeout(err error) bool {
+	if _, ok := errors.AsType[*retries.ErrTimedOut](err); ok {
+		return true
+	}
+	return errors.Is(err, context.DeadlineExceeded)
+}
diff --git a/bundle/direct/maxwait_test.go b/bundle/direct/maxwait_test.go
new file mode 100644
index 00000000000..ebe2356402a
--- /dev/null
+++ b/bundle/direct/maxwait_test.go
@@ -0,0 +1,151 @@
+package direct
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/databricks/cli/bundle/deployplan"
+	bundleenv "github.com/databricks/cli/bundle/env"
+	"github.com/databricks/cli/libs/env"
+	"github.com/databricks/databricks-sdk-go/retries"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestResourceMaxWait(t *testing.T) {
+	tests := []struct {
+		name    string
+		value   string
+		set     bool
+		want    time.Duration
+		wantErr string
+	}{
+		{name: "unset", want: maxWaitUnset},
+		{name: "seconds", value: "90", set: true, want: 90 * time.Second},
+		{name: "zero does not wait", value: "0", set: true, want: 0},
+		{name: "typo", value: "6O", set: true, wantErr: `invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="6O"`},
+		{name: "negative", value: "-5", set: true, wantErr: `invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="-5"`},
+		{name: "duration syntax is not accepted", value: "1m", set: true, wantErr: `invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="1m"`},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := t.Context()
+			if tc.set {
+				ctx = env.Set(ctx, bundleenv.ResourceMaxWaitVariable, tc.value)
+			}
+
+			got, err := resourceMaxWait(ctx)
+			if tc.wantErr != "" {
+				require.ErrorContains(t, err, tc.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}
+
+func TestUnitMaxWait(t *testing.T) {
+	const configured = 30 * time.Second
+
+	tests := []struct {
+		name       string
+		maxWait    time.Duration
+		action     deployplan.ActionType
+		dependents int
+		want       time.Duration
+	}{
+		{name: "create without dependents is capped", maxWait: configured, action: deployplan.Create, want: configured},
+		{name: "create with dependents keeps full wait", maxWait: configured, action: deployplan.Create, dependents: 1, want: maxWaitUnset},
+		{name: "recreate with dependents keeps full wait", maxWait: configured, action: deployplan.Recreate, dependents: 1, want: maxWaitUnset},
+		{name: "delete ignores dependents", maxWait: configured, action: deployplan.Delete, dependents: 2, want: configured},
+		{name: "unset stays unset", maxWait: maxWaitUnset, action: deployplan.Create, want: maxWaitUnset},
+		{name: "unset stays unset for delete", maxWait: maxWaitUnset, action: deployplan.Delete, dependents: 2, want: maxWaitUnset},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, unitMaxWait(tc.maxWait, tc.action, tc.dependents))
+		})
+	}
+}
+
+// TestWaitCappedAbandonsPoll covers the shape produced by retries.Poll, which every resource
+// wait goes through: the parent deadline wins over the resource's own timeout and surfaces as
+// ErrTimedOut rather than a context error.
+func TestWaitCappedAbandonsPoll(t *testing.T) {
+	polls := 0
+	_, err := waitCapped(t.Context(), 50*time.Millisecond, "test resource", func(ctx context.Context) (any, error) {
+		return retries.Poll(ctx, time.Hour, func() (*struct{}, *retries.Err) {
+			polls++
+			return nil, retries.Continues("still provisioning")
+		})
+	})
+
+	require.NoError(t, err)
+	assert.Positive(t, polls)
+}
+
+// TestWaitCappedAbandonsBareContextError covers the other shape: retryWith returns ctx.Err()
+// directly when the cap lands while it sleeps between transient-error retries.
+func TestWaitCappedAbandonsBareContextError(t *testing.T) {
+	_, err := waitCapped(t.Context(), 20*time.Millisecond, "test resource", waitForCtx)
+	require.NoError(t, err)
+}
+
+func TestWaitCappedPropagatesFailure(t *testing.T) {
+	sentinel := errors.New("index failed to provision")
+
+	_, err := waitCapped(t.Context(), time.Minute, "test resource", func(ctx context.Context) (any, error) {
+		return retries.Poll(ctx, time.Hour, func() (*struct{}, *retries.Err) {
+			return nil, retries.Halt(sentinel)
+		})
+	})
+
+	require.ErrorIs(t, err, sentinel)
+}
+
+// TestWaitCappedPropagatesCancellation asserts the cap does not swallow the deployment being
+// cancelled or timing out as a whole, which would carry on past the user's interrupt.
+func TestWaitCappedPropagatesCancellation(t *testing.T) {
+	t.Run("cancelled", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel()
+
+		_, err := waitCapped(ctx, time.Minute, "test resource", waitForCtx)
+		require.ErrorIs(t, err, context.Canceled)
+	})
+
+	t.Run("parent deadline", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+		defer cancel()
+
+		// Cap is longer than the parent deadline, so the parent is what expires.
+		_, err := waitCapped(ctx, time.Minute, "test resource", waitForCtx)
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+}
+
+func TestWaitCappedUnsetAddsNoDeadline(t *testing.T) {
+	_, err := waitCapped(t.Context(), maxWaitUnset, "test resource", func(ctx context.Context) (struct{}, error) {
+		_, ok := ctx.Deadline()
+		assert.False(t, ok, "unset cap must not impose a deadline")
+		return struct{}{}, nil
+	})
+	require.NoError(t, err)
+}
+
+func TestWaitCappedZeroDoesNotWait(t *testing.T) {
+	_, err := waitCapped(t.Context(), 0, "test resource", waitForCtx)
+	require.NoError(t, err)
+}
+
+// waitForCtx blocks until the context is done and reports its error, standing in for a wait
+// that never observes its resource becoming ready.
+func waitForCtx(ctx context.Context) (struct{}, error) {
+	<-ctx.Done()
+	return struct{}{}, ctx.Err()
+}
diff --git a/bundle/direct/pkg.go b/bundle/direct/pkg.go
index 48a9c5a2ff7..ea7ae1e51b6 100644
--- a/bundle/direct/pkg.go
+++ b/bundle/direct/pkg.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"time"
 
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/direct/dresources"
@@ -35,6 +36,11 @@ type DeploymentUnit struct {
 
 	// DependsOn lists resources this resource depends on (persisted in state).
 	DependsOn []deployplan.DependsOnEntry
+
+	// MaxWait caps how long to wait for this resource to reach its target state. Must be set
+	// to maxWaitUnset to fall back to the resource's own timeout: the zero value is a valid
+	// cap meaning "do not wait at all". See resourceMaxWait and unitMaxWait.
+	MaxWait time.Duration
 }
 
 // DeploymentBundle holds everything needed to deploy a bundle
diff --git a/bundle/env/resource_max_wait.go b/bundle/env/resource_max_wait.go
new file mode 100644
index 00000000000..b836a0ab8d1
--- /dev/null
+++ b/bundle/env/resource_max_wait.go
@@ -0,0 +1,13 @@
+package env
+
+import "context"
+
+// ResourceMaxWaitVariable names the environment variable that caps how long deployment waits
+// for a resource to reach its target state.
+const ResourceMaxWaitVariable = "DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT"
+
+// ResourceMaxWait returns the cap (in seconds) on waiting for a resource to reach its target
+// state.
+func ResourceMaxWait(ctx context.Context) (string, bool) {
+	return get(ctx, []string{ResourceMaxWaitVariable})
+}

From 003673812fcaaf86a3101283959632cd43b381f6 Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Tue, 18 Aug 2026 11:21:08 +0200
Subject: [PATCH 02/10] acc: cap the wait in the vector search index basic test

Co-authored-by: Isaac
---
 .../vector_search_indexes/basic/out.test.toml   |  2 +-
 .../vector_search_indexes/basic/test.toml       | 17 +++++++++++++++++
 2 files changed, 18 insertions(+), 1 deletion(-)
 create mode 100644 acceptance/bundle/resources/vector_search_indexes/basic/test.toml

diff --git a/acceptance/bundle/resources/vector_search_indexes/basic/out.test.toml b/acceptance/bundle/resources/vector_search_indexes/basic/out.test.toml
index d4ab7a4a156..7f5765d7d4a 100644
--- a/acceptance/bundle/resources/vector_search_indexes/basic/out.test.toml
+++ b/acceptance/bundle/resources/vector_search_indexes/basic/out.test.toml
@@ -1,4 +1,4 @@
 Cloud = true
-CloudSlow = true
+CloudSlow = false
 RequiresUnityCatalog = true
 EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/resources/vector_search_indexes/basic/test.toml b/acceptance/bundle/resources/vector_search_indexes/basic/test.toml
new file mode 100644
index 00000000000..17aee8dafd4
--- /dev/null
+++ b/acceptance/bundle/resources/vector_search_indexes/basic/test.toml
@@ -0,0 +1,17 @@
+# Index provisioning takes ~30 minutes, which exceeded the test timeout and made this the
+# slowest test in the cloud job. Nothing here depends on the index being queryable — the
+# assertions only need it to exist — so cap the wait instead. The cap is what this test's
+# runtime now consists of (~2x cap: once on create, once on delete), so keep it small.
+Env.DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT = "30"
+
+# With the wait capped this is no longer slow, so run it on every PR rather than only in the
+# nightly job. The parent sets CloudSlow=true for the tests that still wait ~30 minutes.
+CloudSlow = false
+Cloud = true
+
+# The cap makes deploy warn that it stopped waiting. Drop the line rather than baking it into
+# the golden: it is a property of this test's configuration, not of the behaviour under test.
+# Trailing \n so the whole line goes, leaving no blank line behind.
+[[Repls]]
+Old = 'Warn: [^\n]*Stopped waiting[^\n]*\n'
+New = ''

From 8de468b013fc278314b994ccb88ba15de2c342f2 Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Tue, 18 Aug 2026 11:59:57 +0200
Subject: [PATCH 03/10] direct: do not let a permissions/grants child block the
 wait cap

A .permissions or .grants child references only its parent's id, which DoCreate
returns before the wait starts, so it does not need a provisioned parent. Exclude
child nodes when deciding whether a resource has a dependent, and cap the wait in
the vector search index grants test.

Co-authored-by: Isaac
---
 .../grants/select/out.test.toml               |  2 +-
 .../grants/select/test.toml                   | 13 +++++++
 bundle/direct/bundle_apply.go                 |  3 +-
 bundle/direct/maxwait.go                      | 20 ++++++++++
 bundle/direct/maxwait_test.go                 | 38 +++++++++++++++++++
 5 files changed, 73 insertions(+), 3 deletions(-)

diff --git a/acceptance/bundle/resources/vector_search_indexes/grants/select/out.test.toml b/acceptance/bundle/resources/vector_search_indexes/grants/select/out.test.toml
index d4ab7a4a156..7f5765d7d4a 100644
--- a/acceptance/bundle/resources/vector_search_indexes/grants/select/out.test.toml
+++ b/acceptance/bundle/resources/vector_search_indexes/grants/select/out.test.toml
@@ -1,4 +1,4 @@
 Cloud = true
-CloudSlow = true
+CloudSlow = false
 RequiresUnityCatalog = true
 EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/resources/vector_search_indexes/grants/select/test.toml b/acceptance/bundle/resources/vector_search_indexes/grants/select/test.toml
index 0b0ea585e09..760b10789ec 100644
--- a/acceptance/bundle/resources/vector_search_indexes/grants/select/test.toml
+++ b/acceptance/bundle/resources/vector_search_indexes/grants/select/test.toml
@@ -1,2 +1,15 @@
 RequiresUnityCatalog = true
 RecordRequests = true
+
+# The index's only dependent is its own grants child, which needs the index id but not a
+# provisioned index, so the wait is capped here too. Runtime becomes ~2x the cap.
+Env.DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT = "30"
+
+# No longer slow once the wait is capped, so run it on every PR.
+CloudSlow = false
+Cloud = true
+
+# See basic/test.toml: the cap makes deploy warn that it stopped waiting.
+[[Repls]]
+Old = 'Warn: [^\n]*Stopped waiting[^\n]*\n'
+New = ''
diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index 12205677f2d..a99bcb260ff 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -79,8 +79,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 			return false
 		}
 
-		// g.Adj holds the edges out of this node, i.e. the resources that run after it.
-		unitWait := unitMaxWait(maxWait, action, len(g.Adj[resourceKey]))
+		unitWait := unitMaxWait(maxWait, action, countBlockingDependents(g, resourceKey))
 		if maxWait != maxWaitUnset && unitWait == maxWaitUnset {
 			log.Debugf(ctx, "Not capping wait for %s: other resources depend on it", resourceKey)
 		}
diff --git a/bundle/direct/maxwait.go b/bundle/direct/maxwait.go
index ae58ed042a4..5ef58b062e6 100644
--- a/bundle/direct/maxwait.go
+++ b/bundle/direct/maxwait.go
@@ -5,10 +5,12 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/databricks/cli/bundle/deployplan"
 	bundleenv "github.com/databricks/cli/bundle/env"
+	"github.com/databricks/cli/libs/dagrun"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/databricks-sdk-go/retries"
 )
@@ -33,6 +35,24 @@ func resourceMaxWait(ctx context.Context) (time.Duration, error) {
 	return time.Duration(seconds) * time.Second, nil
 }
 
+// countBlockingDependents returns how many nodes that run after resourceKey need it to have
+// reached its target state.
+//
+// Child nodes (.permissions, .grants) are excluded: they reference nothing but the parent's id
+// (see PrepareGrantsInputConfig and PreparePermissionsInputConfig), which DoCreate returns
+// before the wait even starts, so they attach to a resource that exists but is not yet
+// provisioned. Only a 4-segment key can have a 3-segment resource key as its prefix, so the
+// prefix test cannot match a sibling.
+func countBlockingDependents(g *dagrun.Graph, resourceKey string) int {
+	n := 0
+	for _, edge := range g.Adj[resourceKey] {
+		if !strings.HasPrefix(edge.To, resourceKey+".") {
+			n++
+		}
+	}
+	return n
+}
+
 // unitMaxWait returns the cap to apply to a single resource, given how many resources run
 // after it in the deployment graph.
 //
diff --git a/bundle/direct/maxwait_test.go b/bundle/direct/maxwait_test.go
index ebe2356402a..388467a8d78 100644
--- a/bundle/direct/maxwait_test.go
+++ b/bundle/direct/maxwait_test.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/databricks/cli/bundle/deployplan"
 	bundleenv "github.com/databricks/cli/bundle/env"
+	"github.com/databricks/cli/libs/dagrun"
 	"github.com/databricks/cli/libs/env"
 	"github.com/databricks/databricks-sdk-go/retries"
 	"github.com/stretchr/testify/assert"
@@ -149,3 +150,40 @@ func waitForCtx(ctx context.Context) (struct{}, error) {
 	<-ctx.Done()
 	return struct{}{}, ctx.Err()
 }
+
+func TestCountBlockingDependents(t *testing.T) {
+	const index = "resources.vector_search_indexes.foo"
+
+	tests := []struct {
+		name  string
+		edges []string
+		want  int
+	}{
+		{name: "no dependents", want: 0},
+		{name: "grants child does not block", edges: []string{index + ".grants"}, want: 0},
+		{name: "permissions child does not block", edges: []string{index + ".permissions"}, want: 0},
+		{name: "another resource blocks", edges: []string{"resources.jobs.bar"}, want: 1},
+		{
+			name:  "child and resource together",
+			edges: []string{index + ".grants", "resources.jobs.bar"},
+			want:  1,
+		},
+		{
+			// A sibling sharing a name prefix is not a child, so it must still block.
+			name:  "name-prefixed sibling blocks",
+			edges: []string{"resources.vector_search_indexes.foobar"},
+			want:  1,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			g := dagrun.NewGraph()
+			g.AddNode(index)
+			for _, to := range tc.edges {
+				g.AddDirectedEdge(index, to, "label")
+			}
+			assert.Equal(t, tc.want, countBlockingDependents(g, index))
+		})
+	}
+}

From ce787db4f9e18fb8fb185e2156e215ad855cc5df Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Tue, 18 Aug 2026 14:14:15 +0200
Subject: [PATCH 04/10] direct: hasBlockingDependents instead of a count

Co-authored-by: Isaac
---
 bundle/direct/bundle_apply.go |  2 +-
 bundle/direct/maxwait.go      | 17 ++++++++--------
 bundle/direct/maxwait_test.go | 38 +++++++++++++++++------------------
 3 files changed, 28 insertions(+), 29 deletions(-)

diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index a99bcb260ff..f1244c56f7c 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -79,7 +79,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 			return false
 		}
 
-		unitWait := unitMaxWait(maxWait, action, countBlockingDependents(g, resourceKey))
+		unitWait := unitMaxWait(maxWait, action, hasBlockingDependents(g, resourceKey))
 		if maxWait != maxWaitUnset && unitWait == maxWaitUnset {
 			log.Debugf(ctx, "Not capping wait for %s: other resources depend on it", resourceKey)
 		}
diff --git a/bundle/direct/maxwait.go b/bundle/direct/maxwait.go
index 5ef58b062e6..93b18e268e6 100644
--- a/bundle/direct/maxwait.go
+++ b/bundle/direct/maxwait.go
@@ -35,7 +35,7 @@ func resourceMaxWait(ctx context.Context) (time.Duration, error) {
 	return time.Duration(seconds) * time.Second, nil
 }
 
-// countBlockingDependents returns how many nodes that run after resourceKey need it to have
+// hasBlockingDependents reports whether any node that runs after resourceKey needs it to have
 // reached its target state.
 //
 // Child nodes (.permissions, .grants) are excluded: they reference nothing but the parent's id
@@ -43,18 +43,17 @@ func resourceMaxWait(ctx context.Context) (time.Duration, error) {
 // before the wait even starts, so they attach to a resource that exists but is not yet
 // provisioned. Only a 4-segment key can have a 3-segment resource key as its prefix, so the
 // prefix test cannot match a sibling.
-func countBlockingDependents(g *dagrun.Graph, resourceKey string) int {
-	n := 0
+func hasBlockingDependents(g *dagrun.Graph, resourceKey string) bool {
 	for _, edge := range g.Adj[resourceKey] {
 		if !strings.HasPrefix(edge.To, resourceKey+".") {
-			n++
+			return true
 		}
 	}
-	return n
+	return false
 }
 
-// unitMaxWait returns the cap to apply to a single resource, given how many resources run
-// after it in the deployment graph.
+// unitMaxWait returns the cap to apply to a single resource, given whether anything that runs
+// after it in the deployment graph needs it provisioned.
 //
 // Deletes are capped regardless of dependents. The trade-off is accepted deliberately: state
 // is dropped before the wait, so a cut-short delete leaves the resource untracked while it is
@@ -64,8 +63,8 @@ func countBlockingDependents(g *dagrun.Graph, resourceKey string) int {
 //
 // Every other action is capped only when nothing depends on this resource, since a dependent
 // would otherwise act on a resource that has not reached its target state.
-func unitMaxWait(maxWait time.Duration, action deployplan.ActionType, dependents int) time.Duration {
-	if action == deployplan.Delete || dependents == 0 {
+func unitMaxWait(maxWait time.Duration, action deployplan.ActionType, hasDependents bool) time.Duration {
+	if action == deployplan.Delete || !hasDependents {
 		return maxWait
 	}
 	return maxWaitUnset
diff --git a/bundle/direct/maxwait_test.go b/bundle/direct/maxwait_test.go
index 388467a8d78..04f1e6974c9 100644
--- a/bundle/direct/maxwait_test.go
+++ b/bundle/direct/maxwait_test.go
@@ -53,23 +53,23 @@ func TestUnitMaxWait(t *testing.T) {
 	const configured = 30 * time.Second
 
 	tests := []struct {
-		name       string
-		maxWait    time.Duration
-		action     deployplan.ActionType
-		dependents int
-		want       time.Duration
+		name          string
+		maxWait       time.Duration
+		action        deployplan.ActionType
+		hasDependents bool
+		want          time.Duration
 	}{
 		{name: "create without dependents is capped", maxWait: configured, action: deployplan.Create, want: configured},
-		{name: "create with dependents keeps full wait", maxWait: configured, action: deployplan.Create, dependents: 1, want: maxWaitUnset},
-		{name: "recreate with dependents keeps full wait", maxWait: configured, action: deployplan.Recreate, dependents: 1, want: maxWaitUnset},
-		{name: "delete ignores dependents", maxWait: configured, action: deployplan.Delete, dependents: 2, want: configured},
+		{name: "create with dependents keeps full wait", maxWait: configured, action: deployplan.Create, hasDependents: true, want: maxWaitUnset},
+		{name: "recreate with dependents keeps full wait", maxWait: configured, action: deployplan.Recreate, hasDependents: true, want: maxWaitUnset},
+		{name: "delete ignores dependents", maxWait: configured, action: deployplan.Delete, hasDependents: true, want: configured},
 		{name: "unset stays unset", maxWait: maxWaitUnset, action: deployplan.Create, want: maxWaitUnset},
-		{name: "unset stays unset for delete", maxWait: maxWaitUnset, action: deployplan.Delete, dependents: 2, want: maxWaitUnset},
+		{name: "unset stays unset for delete", maxWait: maxWaitUnset, action: deployplan.Delete, hasDependents: true, want: maxWaitUnset},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			assert.Equal(t, tc.want, unitMaxWait(tc.maxWait, tc.action, tc.dependents))
+			assert.Equal(t, tc.want, unitMaxWait(tc.maxWait, tc.action, tc.hasDependents))
 		})
 	}
 }
@@ -151,28 +151,28 @@ func waitForCtx(ctx context.Context) (struct{}, error) {
 	return struct{}{}, ctx.Err()
 }
 
-func TestCountBlockingDependents(t *testing.T) {
+func TestHasBlockingDependents(t *testing.T) {
 	const index = "resources.vector_search_indexes.foo"
 
 	tests := []struct {
 		name  string
 		edges []string
-		want  int
+		want  bool
 	}{
-		{name: "no dependents", want: 0},
-		{name: "grants child does not block", edges: []string{index + ".grants"}, want: 0},
-		{name: "permissions child does not block", edges: []string{index + ".permissions"}, want: 0},
-		{name: "another resource blocks", edges: []string{"resources.jobs.bar"}, want: 1},
+		{name: "no dependents", want: false},
+		{name: "grants child does not block", edges: []string{index + ".grants"}, want: false},
+		{name: "permissions child does not block", edges: []string{index + ".permissions"}, want: false},
+		{name: "another resource blocks", edges: []string{"resources.jobs.bar"}, want: true},
 		{
 			name:  "child and resource together",
 			edges: []string{index + ".grants", "resources.jobs.bar"},
-			want:  1,
+			want:  true,
 		},
 		{
 			// A sibling sharing a name prefix is not a child, so it must still block.
 			name:  "name-prefixed sibling blocks",
 			edges: []string{"resources.vector_search_indexes.foobar"},
-			want:  1,
+			want:  true,
 		},
 	}
 
@@ -183,7 +183,7 @@ func TestCountBlockingDependents(t *testing.T) {
 			for _, to := range tc.edges {
 				g.AddDirectedEdge(index, to, "label")
 			}
-			assert.Equal(t, tc.want, countBlockingDependents(g, index))
+			assert.Equal(t, tc.want, hasBlockingDependents(g, index))
 		})
 	}
 }

From d737365dac02e96998272efa831518eadd215a8b Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Tue, 18 Aug 2026 14:29:03 +0200
Subject: [PATCH 05/10] direct: inline unitMaxWait, make a zero cap skip the
 wait

A zero cap built an already-expired context and still spent one poll learning what
the caller said it did not care about. Returning early makes it observable: the new
acceptance test deploys a cluster, which the test server only advances from PENDING
to RUNNING on read, so the absent poll is what proves the resource was left unwaited.

Co-authored-by: Isaac
---
 .../resource-max-wait-zero/databricks.yml     | 15 +++++
 .../resource-max-wait-zero/out.test.toml      |  2 +
 .../deploy/resource-max-wait-zero/output.txt  | 64 +++++++++++++++++++
 .../deploy/resource-max-wait-zero/script      | 22 +++++++
 .../deploy/resource-max-wait-zero/test.toml   |  9 +++
 bundle/direct/bundle_apply.go                 | 14 +++-
 bundle/direct/maxwait.go                      | 27 +++-----
 bundle/direct/maxwait_test.go                 | 33 ++--------
 8 files changed, 137 insertions(+), 49 deletions(-)
 create mode 100644 acceptance/bundle/deploy/resource-max-wait-zero/databricks.yml
 create mode 100644 acceptance/bundle/deploy/resource-max-wait-zero/out.test.toml
 create mode 100644 acceptance/bundle/deploy/resource-max-wait-zero/output.txt
 create mode 100644 acceptance/bundle/deploy/resource-max-wait-zero/script
 create mode 100644 acceptance/bundle/deploy/resource-max-wait-zero/test.toml

diff --git a/acceptance/bundle/deploy/resource-max-wait-zero/databricks.yml b/acceptance/bundle/deploy/resource-max-wait-zero/databricks.yml
new file mode 100644
index 00000000000..dd5b5c0bf57
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait-zero/databricks.yml
@@ -0,0 +1,15 @@
+bundle:
+  name: test-bundle
+
+# Nothing to sync: keeps the "Files:" line out of the output, which would otherwise
+# count whatever the harness has written into the test directory.
+sync:
+  paths: []
+
+resources:
+  clusters:
+    my_cluster:
+      cluster_name: test-cluster
+      spark_version: 13.3.x-scala2.12
+      node_type_id: i3.xlarge
+      num_workers: 1
diff --git a/acceptance/bundle/deploy/resource-max-wait-zero/out.test.toml b/acceptance/bundle/deploy/resource-max-wait-zero/out.test.toml
new file mode 100644
index 00000000000..0938e678987
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait-zero/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/deploy/resource-max-wait-zero/output.txt b/acceptance/bundle/deploy/resource-max-wait-zero/output.txt
new file mode 100644
index 00000000000..8fbb8e94852
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait-zero/output.txt
@@ -0,0 +1,64 @@
+
+=== Zero cap: the create wait is skipped entirely
+>>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Warn: deploying resources.clusters.my_cluster: Not waiting for creation of resources.clusters.my_cluster (DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0); it may still be in progress
+Created clusters.my_cluster
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py --get --unique //clusters
+{
+  "method": "POST",
+  "path": "/api/2.1/clusters/create",
+  "body": {
+    "autotermination_minutes": 60,
+    "cluster_name": "test-cluster",
+    "node_type_id": "[NODE_TYPE_ID]",
+    "num_workers": 1,
+    "spark_version": "13.3.x-scala2.12"
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.my_cluster
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
+
+Destroy: 1 deleted
+
+=== No cap: the deployment polls until the cluster is RUNNING
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Created clusters.my_cluster
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py --get --unique //clusters
+{
+  "method": "POST",
+  "path": "/api/2.1/clusters/create",
+  "body": {
+    "autotermination_minutes": 60,
+    "cluster_name": "test-cluster",
+    "node_type_id": "[NODE_TYPE_ID]",
+    "num_workers": 1,
+    "spark_version": "13.3.x-scala2.12"
+  }
+}
+{
+  "method": "GET",
+  "path": "/api/2.1/clusters/get",
+  "q": {
+    "cluster_id": "[UUID]"
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.my_cluster
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
+
+Destroy: 1 deleted
diff --git a/acceptance/bundle/deploy/resource-max-wait-zero/script b/acceptance/bundle/deploy/resource-max-wait-zero/script
new file mode 100644
index 00000000000..da0294581f1
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait-zero/script
@@ -0,0 +1,22 @@
+cleanup() {
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+title "Zero cap: the create wait is skipped entirely"
+rm -f out.requests.txt
+trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 $CLI bundle deploy
+
+# Only the create POST is recorded. No GET means the cluster was left PENDING: the
+# testserver advances PENDING -> RUNNING on read, so a wait would have shown up here.
+trace print_requests.py --get --unique //clusters
+
+trace $CLI bundle destroy --auto-approve
+
+title "No cap: the deployment polls until the cluster is RUNNING"
+rm -f out.requests.txt
+trace $CLI bundle deploy
+
+trace print_requests.py --get --unique //clusters
+
+trace $CLI bundle destroy --auto-approve
diff --git a/acceptance/bundle/deploy/resource-max-wait-zero/test.toml b/acceptance/bundle/deploy/resource-max-wait-zero/test.toml
new file mode 100644
index 00000000000..3c8a65fabca
--- /dev/null
+++ b/acceptance/bundle/deploy/resource-max-wait-zero/test.toml
@@ -0,0 +1,9 @@
+# The cap is read by the direct engine only.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+RecordRequests = true
+
+# A cluster is created PENDING and only reaches RUNNING when something polls it
+# (see ClustersGet in libs/testserver), so the poll GET is what proves whether the
+# deployment waited.
+Ignore = [".databricks"]
diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index f1244c56f7c..424ae2bdec9 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -79,9 +79,17 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 			return false
 		}
 
-		unitWait := unitMaxWait(maxWait, action, hasBlockingDependents(g, resourceKey))
-		if maxWait != maxWaitUnset && unitWait == maxWaitUnset {
-			log.Debugf(ctx, "Not capping wait for %s: other resources depend on it", resourceKey)
+		// Deletes are capped even with dependents: state is dropped before the wait, so a
+		// cut-short delete leaves the resource untracked while it tears down, and a dependency
+		// deleted after it may be rejected for still having a child. Accepted deliberately.
+		// Recreate's internal delete-wait is never routed through the cap at all, because it
+		// releases the name for the create that follows.
+		unitWait := maxWait
+		if action != deployplan.Delete && hasBlockingDependents(g, resourceKey) {
+			unitWait = maxWaitUnset
+			if maxWait != maxWaitUnset {
+				log.Debugf(ctx, "Not capping wait for %s: other resources depend on it", resourceKey)
+			}
 		}
 
 		d := &DeploymentUnit{
diff --git a/bundle/direct/maxwait.go b/bundle/direct/maxwait.go
index 93b18e268e6..a13a5faf61d 100644
--- a/bundle/direct/maxwait.go
+++ b/bundle/direct/maxwait.go
@@ -8,7 +8,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/databricks/cli/bundle/deployplan"
 	bundleenv "github.com/databricks/cli/bundle/env"
 	"github.com/databricks/cli/libs/dagrun"
 	"github.com/databricks/cli/libs/log"
@@ -52,24 +51,6 @@ func hasBlockingDependents(g *dagrun.Graph, resourceKey string) bool {
 	return false
 }
 
-// unitMaxWait returns the cap to apply to a single resource, given whether anything that runs
-// after it in the deployment graph needs it provisioned.
-//
-// Deletes are capped regardless of dependents. The trade-off is accepted deliberately: state
-// is dropped before the wait, so a cut-short delete leaves the resource untracked while it is
-// still tearing down, and the dependency deleted after it may then be rejected for still
-// having a child. Recreate's internal delete-wait is excluded structurally — it is the one
-// wait never routed through here, because it releases the name for the following create.
-//
-// Every other action is capped only when nothing depends on this resource, since a dependent
-// would otherwise act on a resource that has not reached its target state.
-func unitMaxWait(maxWait time.Duration, action deployplan.ActionType, hasDependents bool) time.Duration {
-	if action == deployplan.Delete || !hasDependents {
-		return maxWait
-	}
-	return maxWaitUnset
-}
-
 // waitCapped runs wait under maxWait. When the cap expires the wait is abandoned with a
 // warning instead of failing the deployment: state is written before the wait, so the
 // resource stays tracked and the next plan reconciles it. Genuine failures still propagate,
@@ -79,6 +60,14 @@ func waitCapped[T any](ctx context.Context, maxWait time.Duration, description s
 		return wait(ctx)
 	}
 
+	if maxWait == 0 {
+		// Skip the call rather than starting a poll that is already out of time, which would
+		// spend one request to learn what the caller has already said it does not care about.
+		log.Warnf(ctx, "Not waiting for %s (%s=0); it may still be in progress", description, bundleenv.ResourceMaxWaitVariable)
+		var zero T
+		return zero, nil
+	}
+
 	waitCtx, cancel := context.WithTimeout(ctx, maxWait)
 	defer cancel()
 
diff --git a/bundle/direct/maxwait_test.go b/bundle/direct/maxwait_test.go
index 04f1e6974c9..4b3f8e2b546 100644
--- a/bundle/direct/maxwait_test.go
+++ b/bundle/direct/maxwait_test.go
@@ -6,7 +6,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/databricks/cli/bundle/deployplan"
 	bundleenv "github.com/databricks/cli/bundle/env"
 	"github.com/databricks/cli/libs/dagrun"
 	"github.com/databricks/cli/libs/env"
@@ -49,31 +48,6 @@ func TestResourceMaxWait(t *testing.T) {
 	}
 }
 
-func TestUnitMaxWait(t *testing.T) {
-	const configured = 30 * time.Second
-
-	tests := []struct {
-		name          string
-		maxWait       time.Duration
-		action        deployplan.ActionType
-		hasDependents bool
-		want          time.Duration
-	}{
-		{name: "create without dependents is capped", maxWait: configured, action: deployplan.Create, want: configured},
-		{name: "create with dependents keeps full wait", maxWait: configured, action: deployplan.Create, hasDependents: true, want: maxWaitUnset},
-		{name: "recreate with dependents keeps full wait", maxWait: configured, action: deployplan.Recreate, hasDependents: true, want: maxWaitUnset},
-		{name: "delete ignores dependents", maxWait: configured, action: deployplan.Delete, hasDependents: true, want: configured},
-		{name: "unset stays unset", maxWait: maxWaitUnset, action: deployplan.Create, want: maxWaitUnset},
-		{name: "unset stays unset for delete", maxWait: maxWaitUnset, action: deployplan.Delete, hasDependents: true, want: maxWaitUnset},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			assert.Equal(t, tc.want, unitMaxWait(tc.maxWait, tc.action, tc.hasDependents))
-		})
-	}
-}
-
 // TestWaitCappedAbandonsPoll covers the shape produced by retries.Poll, which every resource
 // wait goes through: the parent deadline wins over the resource's own timeout and surfaces as
 // ErrTimedOut rather than a context error.
@@ -140,8 +114,13 @@ func TestWaitCappedUnsetAddsNoDeadline(t *testing.T) {
 }
 
 func TestWaitCappedZeroDoesNotWait(t *testing.T) {
-	_, err := waitCapped(t.Context(), 0, "test resource", waitForCtx)
+	called := false
+	_, err := waitCapped(t.Context(), 0, "test resource", func(ctx context.Context) (struct{}, error) {
+		called = true
+		return struct{}{}, nil
+	})
 	require.NoError(t, err)
+	assert.False(t, called, "a zero cap must not call the wait at all")
 }
 
 // waitForCtx blocks until the context is done and reports its error, standing in for a wait

From 86d1d3ecbccabe3e524ea1930763b2c43900f566 Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Wed, 19 Aug 2026 12:45:42 +0200
Subject: [PATCH 06/10] acc: run the vector search index invariants on cloud
 again

They were excluded from cloud for taking 10-20 minutes, all of it waiting for the
index to provision. None of the invariants need a queryable index, so cap the wait
per config instead of skipping the coverage.

Co-authored-by: Isaac
---
 .../configs/vector_search_index.yml.tmpl-init.sh       | 10 ++++++++++
 acceptance/bundle/invariant/test.toml                  |  6 ------
 2 files changed, 10 insertions(+), 6 deletions(-)
 create mode 100644 acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh

diff --git a/acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh b/acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh
new file mode 100644
index 00000000000..3f8358be248
--- /dev/null
+++ b/acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh
@@ -0,0 +1,10 @@
+#!/bin/bash
+
+# Index provisioning takes 15-30 minutes, which is why this config used to be excluded from
+# cloud runs entirely. None of the invariants need a queryable index -- they deploy, re-plan,
+# delete and re-delete -- so cap the wait instead of skipping the coverage.
+#
+# Sourced by invariant_render, so the export applies to every $CLI call in the script. It has
+# to live here rather than in test.toml, which is per-directory and would cap the waits of all
+# the other invariant configs too.
+export DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=30
diff --git a/acceptance/bundle/invariant/test.toml b/acceptance/bundle/invariant/test.toml
index faa2872a3b0..85690c3d45c 100644
--- a/acceptance/bundle/invariant/test.toml
+++ b/acceptance/bundle/invariant/test.toml
@@ -112,12 +112,6 @@ no_external_location_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=external_loc
 # External volumes reference external locations; excluded from cloud for the same reason
 no_external_volume_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=volume_external.yml.tmpl"]
 
-# A vector search index takes 10-20 min to create, and these tests deploy it twice
-# (deploy, then re-plan), making it by far the slowest variant on cloud. The dedicated
-# vector_search_indexes resource test already covers it on cloud (CloudSlow), so exclude
-# it here to keep the cloud invariant runs from timing out. Still exercised locally.
-no_vector_search_index_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=vector_search_index.yml.tmpl"]
-
 # Telemetry requires a real model, which cloud invariant tests do not provision.
 no_model_serving_endpoint_telemetry_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=model_serving_endpoint_telemetry.yml.tmpl"]
 

From d207d10a4388320697d4a78269cfff8353acee19 Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Thu, 20 Aug 2026 11:26:12 +0200
Subject: [PATCH 07/10] acc: say why the vector search index cap is per config

Co-authored-by: Isaac
---
 .../configs/vector_search_index.yml.tmpl-init.sh          | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh b/acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh
index 3f8358be248..7a077fa9132 100644
--- a/acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh
+++ b/acceptance/bundle/invariant/configs/vector_search_index.yml.tmpl-init.sh
@@ -4,7 +4,9 @@
 # cloud runs entirely. None of the invariants need a queryable index -- they deploy, re-plan,
 # delete and re-delete -- so cap the wait instead of skipping the coverage.
 #
-# Sourced by invariant_render, so the export applies to every $CLI call in the script. It has
-# to live here rather than in test.toml, which is per-directory and would cap the waits of all
-# the other invariant configs too.
+# Sourced by invariant_render, so the export applies to every $CLI call in the script.
+#
+# Deliberately per-config rather than Env in the directory's test.toml: this index is three
+# orders of magnitude slower than the next slowest config (cluster, ~100s), and capping every
+# config would leave nothing exercising a real WaitAfterCreate against a real backend.
 export DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=30

From 87b27d66ff61916d493b5fe07d965a0adf5c8938 Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Thu, 20 Aug 2026 12:41:00 +0200
Subject: [PATCH 08/10] acc: drop the no-op zero case from the max-wait parsing
 test

Jobs have no WaitAfterCreate, so a zero cap there asserted nothing -- the step passed
whether or not the zero path worked. resource-max-wait-zero covers it on a cluster,
which does wait.

Co-authored-by: Isaac
---
 acceptance/bundle/deploy/resource-max-wait/output.txt | 8 +-------
 acceptance/bundle/deploy/resource-max-wait/script     | 5 +----
 acceptance/bundle/deploy/resource-max-wait/test.toml  | 3 +++
 3 files changed, 5 insertions(+), 11 deletions(-)

diff --git a/acceptance/bundle/deploy/resource-max-wait/output.txt b/acceptance/bundle/deploy/resource-max-wait/output.txt
index 52e43bdde11..9da7b2c014a 100644
--- a/acceptance/bundle/deploy/resource-max-wait/output.txt
+++ b/acceptance/bundle/deploy/resource-max-wait/output.txt
@@ -12,19 +12,13 @@ Error: invalid DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT="-5": expected a non-negative
 
 Files: 0 uploaded, 0 deleted
 
-=== A valid value deploys normally
+=== A valid value is accepted
 >>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
 Created jobs.my_job
 Files: 0 uploaded, 0 deleted
 Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
 
-=== Zero means do not wait at all
->>> DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 [CLI] bundle deploy
-Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Files: 0 uploaded, 0 deleted
-Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
-
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
   delete resources.jobs.my_job
diff --git a/acceptance/bundle/deploy/resource-max-wait/script b/acceptance/bundle/deploy/resource-max-wait/script
index a457a311c6e..937b27e936a 100644
--- a/acceptance/bundle/deploy/resource-max-wait/script
+++ b/acceptance/bundle/deploy/resource-max-wait/script
@@ -2,10 +2,7 @@ title "A malformed value is rejected instead of falling back to the default wait
 trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=1m musterr $CLI bundle deploy
 trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=-5 musterr $CLI bundle deploy
 
-title "A valid value deploys normally"
+title "A valid value is accepted"
 trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=60 $CLI bundle deploy
 
-title "Zero means do not wait at all"
-trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 $CLI bundle deploy
-
 trace $CLI bundle destroy --auto-approve
diff --git a/acceptance/bundle/deploy/resource-max-wait/test.toml b/acceptance/bundle/deploy/resource-max-wait/test.toml
index 427bdd03f18..cb44a5e0409 100644
--- a/acceptance/bundle/deploy/resource-max-wait/test.toml
+++ b/acceptance/bundle/deploy/resource-max-wait/test.toml
@@ -1,2 +1,5 @@
+# Covers parsing of DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT only. What the cap does to a wait
+# is covered by ../resource-max-wait-zero, whose resource actually has one: jobs have no
+# WaitAfterCreate, so a cap would be a no-op here.
 # DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT is read by the direct engine only.
 EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]

From 4bbcf1ed62d38f5042f203e4c39da35e707d00a8 Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Thu, 20 Aug 2026 14:00:46 +0200
Subject: [PATCH 09/10] acc: print only method and path in the max-wait zero
 test

The assertion is whether the poll GET happened, so the cluster body was churn: any
change to the test server's cluster defaults would have broken it.

Co-authored-by: Isaac
---
 .../deploy/resource-max-wait-zero/output.txt  | 36 +++----------------
 .../deploy/resource-max-wait-zero/script      |  5 +--
 2 files changed, 8 insertions(+), 33 deletions(-)

diff --git a/acceptance/bundle/deploy/resource-max-wait-zero/output.txt b/acceptance/bundle/deploy/resource-max-wait-zero/output.txt
index 8fbb8e94852..1f379a5692e 100644
--- a/acceptance/bundle/deploy/resource-max-wait-zero/output.txt
+++ b/acceptance/bundle/deploy/resource-max-wait-zero/output.txt
@@ -7,18 +7,8 @@ Created clusters.my_cluster
 Files: 0 uploaded, 0 deleted
 Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
 
->>> print_requests.py --get --unique //clusters
-{
-  "method": "POST",
-  "path": "/api/2.1/clusters/create",
-  "body": {
-    "autotermination_minutes": 60,
-    "cluster_name": "test-cluster",
-    "node_type_id": "[NODE_TYPE_ID]",
-    "num_workers": 1,
-    "spark_version": "13.3.x-scala2.12"
-  }
-}
+>>> print_requests.py --get --unique --oneline --del-field body,q //clusters
+{"method": "POST", "path": "/api/2.1/clusters/create"}
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
@@ -35,25 +25,9 @@ Created clusters.my_cluster
 Files: 0 uploaded, 0 deleted
 Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
 
->>> print_requests.py --get --unique //clusters
-{
-  "method": "POST",
-  "path": "/api/2.1/clusters/create",
-  "body": {
-    "autotermination_minutes": 60,
-    "cluster_name": "test-cluster",
-    "node_type_id": "[NODE_TYPE_ID]",
-    "num_workers": 1,
-    "spark_version": "13.3.x-scala2.12"
-  }
-}
-{
-  "method": "GET",
-  "path": "/api/2.1/clusters/get",
-  "q": {
-    "cluster_id": "[UUID]"
-  }
-}
+>>> print_requests.py --get --unique --oneline --del-field body,q //clusters
+{"method": "POST", "path": "/api/2.1/clusters/create"}
+{"method": "GET", "path": "/api/2.1/clusters/get"}
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
diff --git a/acceptance/bundle/deploy/resource-max-wait-zero/script b/acceptance/bundle/deploy/resource-max-wait-zero/script
index da0294581f1..7f1e8ce3b9a 100644
--- a/acceptance/bundle/deploy/resource-max-wait-zero/script
+++ b/acceptance/bundle/deploy/resource-max-wait-zero/script
@@ -9,7 +9,8 @@ trace DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT=0 $CLI bundle deploy
 
 # Only the create POST is recorded. No GET means the cluster was left PENDING: the
 # testserver advances PENDING -> RUNNING on read, so a wait would have shown up here.
-trace print_requests.py --get --unique //clusters
+# Method and path are the whole assertion; the body would only add churn.
+trace print_requests.py --get --unique --oneline --del-field body,q //clusters
 
 trace $CLI bundle destroy --auto-approve
 
@@ -17,6 +18,6 @@ title "No cap: the deployment polls until the cluster is RUNNING"
 rm -f out.requests.txt
 trace $CLI bundle deploy
 
-trace print_requests.py --get --unique //clusters
+trace print_requests.py --get --unique --oneline --del-field body,q //clusters
 
 trace $CLI bundle destroy --auto-approve

From 165db3c84f9183e892ec8c05d50852e6c926c81b Mon Sep 17 00:00:00 2001
From: Denis Bilenko <denis.bilenko@databricks.com>
Date: Thu, 20 Aug 2026 14:38:34 +0200
Subject: [PATCH 10/10] Shorten the changelog entry

Co-authored-by: Isaac
---
 .nextchanges/bundles/resource-max-wait.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.nextchanges/bundles/resource-max-wait.md b/.nextchanges/bundles/resource-max-wait.md
index e6a7d7e6c47..d4ced0c4c51 100644
--- a/.nextchanges/bundles/resource-max-wait.md
+++ b/.nextchanges/bundles/resource-max-wait.md
@@ -1 +1 @@
-Added `DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT` (in seconds) to cap how long `bundle deploy` and `bundle destroy` wait for a resource to become ready or finish deleting. When the cap expires the wait is abandoned with a warning instead of failing — the resource is already recorded in state, so the next deployment reconciles it. Resources that others depend on keep their full wait, since a dependent must not act on a resource that is not ready. Direct engine only.
+Added `DATABRICKS_BUNDLE_RESOURCE_MAX_WAIT` (in seconds) to cap how long deploy and destroy wait for a resource. Direct engine only.