Move direct engine state lifecycle to top-level callers (#4928)

## Changes
- Move state lifecycle (`Open`/`Finalize`) from internal methods to
top-level command callers, so state is opened exactly once per command.
Double-open panics.
- Script-based `bundle run` no longer pulls remote state it doesn't
need.
- Skip the `/api/2.0/preview/scim/v2/Me` call during `bundle run` when
only `ErrorOnEmptyState` is set.
- Skip state file creation when deploy/destroy plans are empty.

## Why
The previous pattern hid `Open`/`Finalize` inside low-level methods,
making it hard to reason about when state was opened, whether it would
be written, and whether double-opens could occur. Lifting the lifecycle
to callers makes the invariant obvious and enforceable, and avoids
unnecessary state I/O and API calls in read-only flows.

Preparation for WAL & metadata service.

## Tests
- Unit tests for `DeploymentState`: open/save/finalize round-trip, panic
on double open, delete tracking.
- Existing acceptance tests cover deploy, destroy, `bundle run`, and
config-remote-sync flows.

Author: denik

bundle/configsync/diff.go

@@ -127,15 +127,19 @@ func DetectChanges(ctx context.Context, b *bundle.Bundle, engine engine.EngineTy
 		return nil, fmt.Errorf("state snapshot not available: %w", err)
 	}
 
-	deployBundle := &direct.DeploymentBundle{}
-	var statePath string
+	var deployBundle *direct.DeploymentBundle
 	if engine.IsDirect() {
-		_, statePath = b.StateFilenameDirect(ctx)
+		// For direct engine, state is already opened by the caller (process.go).
+		deployBundle = &b.DeploymentBundle
 	} else {
-		_, statePath = b.StateFilenameConfigSnapshot(ctx)
+		deployBundle = &direct.DeploymentBundle{}
+		_, statePath := b.StateFilenameConfigSnapshot(ctx)
+		if err := deployBundle.StateDB.Open(statePath); err != nil {
+			return nil, fmt.Errorf("failed to open state: %w", err)
+		}
 	}
 
-	plan, err := deployBundle.CalculatePlan(ctx, b.WorkspaceClient(), &b.Config, statePath)
+	plan, err := deployBundle.CalculatePlan(ctx, b.WorkspaceClient(), &b.Config)
 	if err != nil {
 		return nil, fmt.Errorf("failed to calculate plan: %w", err)
 	}

bundle/deploy/terraform/check_dashboards_modified_remotely.go

@@ -21,13 +21,12 @@ func collectDashboardsFromState(ctx context.Context, b *bundle.Bundle, directDep
 	var state ExportedResourcesMap
 	var err error
 	if directDeployment {
-		_, localPath := b.StateFilenameDirect(ctx)
-		state, err = b.DeploymentBundle.ExportState(ctx, localPath)
+		state = b.DeploymentBundle.ExportState(ctx)
 	} else {
 		state, err = ParseResourcesState(ctx, b)
-	}
-	if err != nil {
-		return nil, err
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	var dashboards []dashboardState

bundle/direct/bind.go

@@ -95,7 +95,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 		return nil, err
 	}
 
-	// Finalize to write temp state to disk so CalculatePlan can read it
+	// Finalize to persist temp state to disk
 	err = b.StateDB.Finalize()
 	if err != nil {
 		os.Remove(tmpStatePath)
@@ -105,7 +105,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 	log.Infof(ctx, "Bound %s to id=%s (in temp state)", resourceKey, resourceID)
 
 	// First plan + update: populate state with resolved config
-	plan, err := b.CalculatePlan(ctx, client, configRoot, tmpStatePath)
+	plan, err := b.CalculatePlan(ctx, client, configRoot)
 	if err != nil {
 		os.Remove(tmpStatePath)
 		return nil, err
@@ -146,7 +146,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 	}
 
 	// Second plan: this is the plan to present to the user (change between remote resource and config)
-	plan, err = b.CalculatePlan(ctx, client, configRoot, tmpStatePath)
+	plan, err = b.CalculatePlan(ctx, client, configRoot)
 	if err != nil {
 		os.Remove(tmpStatePath)
 		return nil, err

bundle/direct/bundle_apply.go

@@ -151,12 +151,6 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 
 		return true
 	})
-
-	// This must run even if deploy failed:
-	err = b.StateDB.Finalize()
-	if err != nil {
-		logdiag.LogError(ctx, err)
-	}
 }
 
 func (b *DeploymentBundle) LookupReferencePostDeploy(ctx context.Context, path *structpath.PathNode) (any, error) {

bundle/direct/bundle_plan.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"maps"
-	"os"
 	"reflect"
 	"slices"
 	"strings"
@@ -42,22 +41,14 @@ func (b *DeploymentBundle) init(client *databricks.WorkspaceClient) error {
 // ValidatePlanAgainstState validates that a plan's lineage and serial match the current state.
 // This should be called early in the deployment process, before any file operations.
 // If the plan has no lineage (first deployment), validation is skipped.
-func ValidatePlanAgainstState(statePath string, plan *deployplan.Plan) error {
+func ValidatePlanAgainstState(stateDB *dstate.DeploymentState, plan *deployplan.Plan) error {
 	// If plan has no lineage, this is a first deployment before any state exists
 	// No validation needed
 	if plan.Lineage == "" {
 		return nil
 	}
 
-	var stateDB dstate.DeploymentState
-	err := stateDB.Open(statePath)
-	if err != nil {
-		// If state file doesn't exist but plan has lineage, something is wrong
-		if os.IsNotExist(err) {
-			return fmt.Errorf("plan has lineage %q but state file does not exist at %s; the state may have been deleted", plan.Lineage, statePath)
-		}
-		return fmt.Errorf("reading state from %s: %w", statePath, err)
-	}
+	stateDB.AssertOpened()
 
 	// Validate that the plan's lineage matches the current state's lineage
 	if plan.Lineage != stateDB.Data.Lineage {
@@ -74,13 +65,10 @@ func ValidatePlanAgainstState(statePath string, plan *deployplan.Plan) error {
 
 // InitForApply initializes the DeploymentBundle for applying a pre-computed plan.
 // This is used when --plan is specified to skip the planning phase.
-func (b *DeploymentBundle) InitForApply(ctx context.Context, client *databricks.WorkspaceClient, statePath string, plan *deployplan.Plan) error {
-	err := b.StateDB.Open(statePath)
-	if err != nil {
-		return fmt.Errorf("reading state from %s: %w", statePath, err)
-	}
+func (b *DeploymentBundle) InitForApply(ctx context.Context, client *databricks.WorkspaceClient, plan *deployplan.Plan) error {
+	b.StateDB.AssertOpened()
 
-	err = b.init(client)
+	err := b.init(client)
 	if err != nil {
 		return err
 	}
@@ -110,13 +98,10 @@ func (b *DeploymentBundle) InitForApply(ctx context.Context, client *databricks.
 	return nil
 }
 
-func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root, statePath string) (*deployplan.Plan, error) {
-	err := b.StateDB.Open(statePath)
-	if err != nil {
-		return nil, fmt.Errorf("reading state from %s: %w", statePath, err)
-	}
+func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root) (*deployplan.Plan, error) {
+	b.StateDB.AssertOpened()
 
-	err = b.init(client)
+	err := b.init(client)
 	if err != nil {
 		return nil, err
 	}

bundle/direct/dstate/state.go

@@ -113,10 +113,7 @@ func (db *DeploymentState) Open(path string) error {
 	defer db.mu.Unlock()
 
 	if db.Path != "" {
-		if db.Path == path {
-			return nil
-		}
-		return fmt.Errorf("already read state %v, cannot open %v", db.Path, path)
+		panic(fmt.Sprintf("state already opened: %v, cannot open %v", db.Path, path))
 	}
 
 	data, err := os.ReadFile(path)

bundle/direct/dstate/state_test.go

@@ -0,0 +1,53 @@
+package dstate
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestOpenSaveFinalizeRoundTrip(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.json")
+
+	var db DeploymentState
+	require.NoError(t, db.Open(path))
+
+	require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{"key": "val"}, nil))
+	require.NoError(t, db.Finalize())
+
+	// Re-open and verify persisted data.
+	var db2 DeploymentState
+	require.NoError(t, db2.Open(path))
+	assert.Equal(t, 1, db2.Data.Serial)
+	assert.Equal(t, "123", db2.GetResourceID("jobs.my_job"))
+}
+
+func TestPanicOnDoubleOpen(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.json")
+
+	var db DeploymentState
+	require.NoError(t, db.Open(path))
+
+	assert.Panics(t, func() {
+		_ = db.Open(path)
+	})
+}
+
+func TestDeleteState(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.json")
+
+	var db DeploymentState
+	require.NoError(t, db.Open(path))
+	require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil))
+	require.NoError(t, db.Finalize())
+
+	require.NoError(t, db.DeleteState("jobs.my_job"))
+	require.NoError(t, db.Finalize())
+
+	var db2 DeploymentState
+	require.NoError(t, db2.Open(path))
+	assert.Equal(t, 2, db2.Data.Serial)
+	assert.Equal(t, "", db2.GetResourceID("jobs.my_job"))
+}

bundle/direct/pkg.go

@@ -64,10 +64,7 @@ func (d *DeploymentUnit) SetRemoteState(remoteState any) error {
 	return nil
 }
 
-func (b *DeploymentBundle) ExportState(ctx context.Context, path string) (resourcestate.ExportedResourcesMap, error) {
-	err := b.StateDB.Open(path)
-	if err != nil {
-		return nil, err
-	}
-	return b.StateDB.ExportState(ctx), nil
+func (b *DeploymentBundle) ExportState(ctx context.Context) resourcestate.ExportedResourcesMap {
+	b.StateDB.AssertOpened()
+	return b.StateDB.ExportState(ctx)
 }

bundle/phases/deploy.go

@@ -104,6 +104,13 @@ func deployCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, ta
 
 	if targetEngine.IsDirect() {
 		b.DeploymentBundle.Apply(ctx, b.WorkspaceClient(), plan, direct.MigrateMode(false))
+		// Finalize state: write to disk even if deploy failed, so partial progress is saved.
+		// Skip for empty plans to avoid creating a state file when nothing was deployed.
+		if len(plan.Plan) > 0 {
+			if err := b.DeploymentBundle.StateDB.Finalize(); err != nil {
+				logdiag.LogError(ctx, err)
+			}
+		}
 	} else {
 		bundle.ApplyContext(ctx, b, terraform.Apply())
 	}
@@ -178,8 +185,7 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 
 	if plan != nil {
 		// Initialize DeploymentBundle for applying the loaded plan
-		_, localPath := b.StateFilenameDirect(ctx)
-		err := b.DeploymentBundle.InitForApply(ctx, b.WorkspaceClient(), localPath, plan)
+		err := b.DeploymentBundle.InitForApply(ctx, b.WorkspaceClient(), plan)
 		if err != nil {
 			logdiag.LogError(ctx, err)
 			return
@@ -213,8 +219,7 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 
 func RunPlan(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) *deployplan.Plan {
 	if engine.IsDirect() {
-		_, localPath := b.StateFilenameDirect(ctx)
-		plan, err := b.DeploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(), &b.Config, localPath)
+		plan, err := b.DeploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(), &b.Config)
 		if err != nil {
 			logdiag.LogError(ctx, err)
 			return nil

bundle/phases/destroy.go

@@ -97,6 +97,12 @@ func approvalForDestroy(ctx context.Context, b *bundle.Bundle, plan *deployplan.
 func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, engine engine.EngineType) {
 	if engine.IsDirect() {
 		b.DeploymentBundle.Apply(ctx, b.WorkspaceClient(), plan, direct.MigrateMode(false))
+		// Skip Finalize for empty plans to avoid creating a state file when nothing was destroyed.
+		if len(plan.Plan) > 0 {
+			if err := b.DeploymentBundle.StateDB.Finalize(); err != nil {
+				logdiag.LogError(ctx, err)
+			}
+		}
 	} else {
 		// Core destructive mutators for destroy. These require informed user consent.
 		bundle.ApplyContext(ctx, b, terraform.Apply())
@@ -157,8 +163,7 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) {
 
 	var plan *deployplan.Plan
 	if engine.IsDirect() {
-		_, localPath := b.StateFilenameDirect(ctx)
-		plan, err = b.DeploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(), nil, localPath)
+		plan, err = b.DeploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(), nil)
 		if err != nil {
 			logdiag.LogError(ctx, err)
 			return