diff --git a/bundle/deploy/metadata/compute.go b/bundle/deploy/metadata/compute.go
index 6fb5336ac45..d7a30605d44 100644
--- a/bundle/deploy/metadata/compute.go
+++ b/bundle/deploy/metadata/compute.go
@@ -8,6 +8,7 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/metadata"
 	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/log"
 )
 
 type compute struct{}
@@ -20,7 +21,7 @@ func (m *compute) Name() string {
 	return "metadata.Compute"
 }
 
-func (m *compute) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
+func (m *compute) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	b.Metadata = metadata.Metadata{
 		Version: metadata.Version,
 		Config:  metadata.Config{},
@@ -40,10 +41,17 @@ func (m *compute) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
 		// Compute config file path the job is defined in, relative to the bundle
 		// root
 		l := b.Config.GetLocation("resources.jobs." + name)
+		if l.File == "" {
+			// b.Config.Resources.Jobs may include a job that only exists in state but not in config
+			continue
+		}
+
 		relativePath, err := filepath.Rel(b.BundleRootPath, l.File)
 		if err != nil {
-			return diag.Errorf("failed to compute relative path for job %s: %v", name, err)
+			log.Warnf(ctx, "failed to compute relative path for job %q: %s", name, err)
+			relativePath = ""
 		}
+
 		// Metadata for the job
 		jobsMetadata[name] = &metadata.Resource{
 			ID:           job.ID,
diff --git a/bundle/deployplan/plan.go b/bundle/deployplan/plan.go
index 9c509e82d65..0da0743d35b 100644
--- a/bundle/deployplan/plan.go
+++ b/bundle/deployplan/plan.go
@@ -97,3 +97,9 @@ func (p *Plan) UnlockEntry(resourceKey string) {
 	defer p.mutex.Unlock()
 	p.locks[resourceKey] = false
 }
+
+func (p *Plan) RemoveEntry(resourceKey string) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	delete(p.Plan, resourceKey)
+}
diff --git a/bundle/direct/apply.go b/bundle/direct/apply.go
index f0dc0ad4c51..e3dc5fcb84b 100644
--- a/bundle/direct/apply.go
+++ b/bundle/direct/apply.go
@@ -93,7 +93,7 @@ func (d *DeploymentUnit) Create(ctx context.Context, db *dstate.DeploymentState,
 
 func (d *DeploymentUnit) Recreate(ctx context.Context, db *dstate.DeploymentState, oldID string, newState any) error {
 	err := d.Adapter.DoDelete(ctx, oldID)
-	if err != nil {
+	if err != nil && !isResourceGone(err) {
 		return fmt.Errorf("deleting old id=%s: %w", oldID, err)
 	}
 
@@ -172,9 +172,8 @@ func (d *DeploymentUnit) UpdateWithID(ctx context.Context, db *dstate.Deployment
 }
 
 func (d *DeploymentUnit) Delete(ctx context.Context, db *dstate.DeploymentState, oldID string) error {
-	// TODO: recognize 404 and 403 as "deleted" and proceed to removing state
 	err := d.Adapter.DoDelete(ctx, oldID)
-	if err != nil {
+	if err != nil && !isResourceGone(err) {
 		return fmt.Errorf("deleting id=%s: %w", oldID, err)
 	}
 
diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index 0d9c369a64a..5f306dd8038 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -161,7 +161,7 @@ func (b *DeploymentBundle) LookupReferenceRemote(ctx context.Context, path *stru
 	return structaccess.Get(remoteState, fieldPath)
 }
 
-func jsonDump(obj map[string]string) string {
+func jsonDump(obj any) string {
 	bytes, err := json.MarshalIndent(obj, "", "  ")
 	if err != nil {
 		return err.Error()
diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go
index f1366316707..756f5b8663a 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"reflect"
 	"slices"
 	"strings"
@@ -11,6 +12,7 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/direct/dresources"
+	"github.com/databricks/cli/bundle/direct/dstate"
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/dynvar"
 	"github.com/databricks/cli/libs/log"
@@ -21,7 +23,6 @@ import (
 	"github.com/databricks/cli/libs/structs/structvar"
 	"github.com/databricks/cli/libs/utils"
 	"github.com/databricks/databricks-sdk-go"
-	"github.com/databricks/databricks-sdk-go/apierr"
 )
 
 var errDelayed = errors.New("must be resolved after apply")
@@ -43,14 +44,14 @@ func (b *DeploymentBundle) Init(client *databricks.WorkspaceClient) error {
 	return err
 }
 
-func (b *DeploymentBundle) CalculatePlanForDeploy(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root) (*deployplan.Plan, error) {
+func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root) (*deployplan.Plan, error) {
 	b.StateDB.AssertOpened()
 	err := b.Init(client)
 	if err != nil {
 		return nil, err
 	}
 
-	plan, err := b.makePlan(ctx, configRoot)
+	plan, err := b.makePlan(ctx, configRoot, &b.StateDB.Data)
 	if err != nil {
 		return nil, fmt.Errorf("reading config: %w", err)
 	}
@@ -67,7 +68,7 @@ func (b *DeploymentBundle) CalculatePlanForDeploy(ctx context.Context, client *d
 		return nil, err
 	}
 
-	// We're processing resources in DAG order because we're resolving refernces (that can be resolved at plan stage).
+	// We're processing resources in DAG order because we're resolving references (that can be resolved at plan stage).
 	g.Run(defaultParallelism, func(resourceKey string, failedDependency *string) bool {
 		errorPrefix := "cannot plan " + resourceKey
 
@@ -90,6 +91,27 @@ func (b *DeploymentBundle) CalculatePlanForDeploy(ctx context.Context, client *d
 			return false
 		}
 
+		if entry.Action == deployplan.ActionTypeDelete.String() {
+			dbentry, hasEntry := b.StateDB.GetResourceEntry(resourceKey)
+			if !hasEntry {
+				logdiag.LogError(ctx, fmt.Errorf("%s: internal error, missing in state", errorPrefix))
+				return false
+			}
+
+			_, err := adapter.DoRefresh(ctx, dbentry.ID)
+			if err != nil {
+				if isResourceGone(err) {
+					// no such resource
+					plan.RemoveEntry(resourceKey)
+				} else {
+					log.Warnf(ctx, "cannot read %s id=%q: %s", resourceKey, dbentry.ID, err)
+					return false
+				}
+			}
+
+			return true
+		}
+
 		// Process all references in the resource using Refs map
 		// Refs maps path inside resource to references e.g. "${resources.jobs.foo.id} ${resources.jobs.foo.name}"
 		if !b.resolveReferences(ctx, entry, errorPrefix, true) {
@@ -139,7 +161,7 @@ func (b *DeploymentBundle) CalculatePlanForDeploy(ctx context.Context, client *d
 
 		remoteState, err := adapter.DoRefresh(ctx, dbentry.ID)
 		if err != nil {
-			if errors.Is(err, apierr.ErrResourceDoesNotExist) || errors.Is(err, apierr.ErrNotFound) {
+			if isResourceGone(err) {
 				remoteState = nil
 			} else {
 				logdiag.LogError(ctx, fmt.Errorf("%s: failed to read id=%q: %w (localAction=%q)", errorPrefix, dbentry.ID, err, localAction.String()))
@@ -188,28 +210,6 @@ func (b *DeploymentBundle) CalculatePlanForDeploy(ctx context.Context, client *d
 		return nil, errors.New("planning failed")
 	}
 
-	// Note, we cannot simply remove 'skip' entries here as then we'd need to ensure there are no edges to them
-
-	state := b.StateDB.ExportState(ctx)
-
-	// Remained in state are resources that no longer present in the config
-	for _, group := range utils.SortedKeys(state) {
-		_, ok := b.Adapters[group]
-
-		if !ok {
-			log.Warnf(ctx, "%s: resource type not supported on direct backend", group)
-			continue
-		}
-		for _, key := range utils.SortedKeys(state[group]) {
-			n := "resources." + group + "." + key
-			_, exists := plan.Plan[n]
-			if exists {
-				continue
-			}
-			plan.Plan[n] = &deployplan.PlanEntry{Action: deployplan.ActionTypeDelete.String()}
-		}
-	}
-
 	for _, entry := range plan.Plan {
 		if entry.Action == deployplan.ActionTypeSkipString {
 			entry.NewState = nil
@@ -271,32 +271,6 @@ func interpretOldStateVsRemoteState(ctx context.Context, adapter *dresources.Ada
 	return action, m
 }
 
-func (b *DeploymentBundle) CalculatePlanForDestroy(ctx context.Context, client *databricks.WorkspaceClient) (*deployplan.Plan, error) {
-	b.StateDB.AssertOpened()
-
-	err := b.Init(client)
-	if err != nil {
-		return nil, err
-	}
-
-	plan := deployplan.NewPlan()
-
-	state := b.StateDB.ExportState(ctx)
-	for group, groupData := range state {
-		_, ok := b.Adapters[group]
-		if !ok {
-			logdiag.LogError(ctx, fmt.Errorf("cannot destroy %s: resource type not supported on direct backend", group))
-			continue
-		}
-		for key := range groupData {
-			n := "resources." + group + "." + key
-			plan.Plan[n] = &deployplan.PlanEntry{Action: deployplan.ActionTypeDelete.String()}
-		}
-	}
-
-	return plan, nil
-}
-
 func (b *DeploymentBundle) LookupReferenceLocal(ctx context.Context, path *structpath.PathNode) (any, error) {
 	targetResourceKey := path.Prefix(3).String()
 	fieldPath := path.SkipPrefix(3)
@@ -432,35 +406,41 @@ func (b *DeploymentBundle) resolveReferences(ctx context.Context, entry *deployp
 	return true
 }
 
-func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root) (*deployplan.Plan, error) {
+func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root, db *dstate.Database) (*deployplan.Plan, error) {
 	p := deployplan.NewPlan()
 
 	// Collect and sort nodes first, because MapByPattern gives them in randomized order
 	var nodes []string
 
-	// Walk?
-	_, err := dyn.MapByPattern(
-		configRoot.Value(),
-		dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey()),
-		func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
-			group := p[1].Key()
+	existingKeys := maps.Clone(db.State)
 
-			_, ok := dresources.SupportedResources[group]
-			if !ok {
-				return v, fmt.Errorf("unsupported resource: %s", group)
-			}
+	// Walk?
+	if configRoot != nil {
+		_, err := dyn.MapByPattern(
+			configRoot.Value(),
+			dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey()),
+			func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+				group := p[1].Key()
+
+				_, ok := dresources.SupportedResources[group]
+				if !ok {
+					return v, fmt.Errorf("unsupported resource: %s", group)
+				}
 
-			nodes = append(nodes, p.String())
-			return dyn.InvalidValue, nil
-		},
-	)
-	if err != nil {
-		return nil, fmt.Errorf("reading config: %w", err)
+				nodes = append(nodes, p.String())
+				return dyn.InvalidValue, nil
+			},
+		)
+		if err != nil {
+			return nil, fmt.Errorf("reading config: %w", err)
+		}
 	}
 
 	slices.Sort(nodes)
 
 	for _, node := range nodes {
+		delete(existingKeys, node)
+
 		prefix := "cannot plan " + node
 		inputConfig, ok := configRoot.GetResourceConfig(node)
 		if !ok {
@@ -541,6 +521,15 @@ func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root
 		p.Plan[node] = &e
 	}
 
+	for n := range existingKeys {
+		if p.Plan[n] != nil {
+			panic("unexpected node " + n)
+		}
+		p.Plan[n] = &deployplan.PlanEntry{
+			Action: deployplan.ActionTypeDelete.String(),
+		}
+	}
+
 	return p, nil
 }
 
diff --git a/bundle/direct/util.go b/bundle/direct/util.go
new file mode 100644
index 00000000000..43c39c9a67f
--- /dev/null
+++ b/bundle/direct/util.go
@@ -0,0 +1,11 @@
+package direct
+
+import (
+	"errors"
+
+	"github.com/databricks/databricks-sdk-go/apierr"
+)
+
+func isResourceGone(err error) bool {
+	return errors.Is(err, apierr.ErrResourceDoesNotExist) || errors.Is(err, apierr.ErrNotFound)
+}
diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go
index ea012d057e9..30421dfa7dd 100644
--- a/bundle/phases/deploy.go
+++ b/bundle/phases/deploy.go
@@ -209,7 +209,7 @@ func planWithoutPrepare(ctx context.Context, b *bundle.Bundle) *deployplan.Plan
 			logdiag.LogError(ctx, err)
 			return nil
 		}
-		plan, err := b.DeploymentBundle.CalculatePlanForDeploy(ctx, b.WorkspaceClient(), &b.Config)
+		plan, err := b.DeploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(), &b.Config)
 		if err != nil {
 			logdiag.LogError(ctx, err)
 			return nil
diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go
index 2ecb6fedd86..d75e7271ae6 100644
--- a/bundle/phases/destroy.go
+++ b/bundle/phases/destroy.go
@@ -156,7 +156,7 @@ func Destroy(ctx context.Context, b *bundle.Bundle) {
 			logdiag.LogError(ctx, err)
 			return
 		}
-		plan, err = b.DeploymentBundle.CalculatePlanForDestroy(ctx, b.WorkspaceClient())
+		plan, err = b.DeploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(), nil)
 		if err != nil {
 			logdiag.LogError(ctx, err)
 			return