Fix pulling remote state snapshot in config-remote-sync (#4444)

## Changes

Currently, CLI stores the remote resource snapshot file to the
`/Workspace/Users/<user>.bundle/<bundle>/dev/state/resources-config-sync-snapshot.json`,
but during config-remote-sync it's never used; the command only relies
on the snapshot from the cache. This PR adds a step to the
config-remote-sync to ensure that the snapshot is loaded

 
## Why
<!-- Why are these changes needed? Provide the context that the reviewer
might be missing.
For example, were there any decisions behind the change that are not
reflected in the code itself? -->

In DABs in the Workspace, cache is flushed when compute is reattached,
which leads to unexpected behavior

## Tests
<!-- How have you tested the changes? -->

<!-- If your PR needs to be included in the release notes for next
release,
add a separate entry in NEXT_CHANGELOG.md as part of your PR. -->

Author: ilyakuz-db

acceptance/bundle/config-remote-sync/flushed_cache/databricks.yml.tmpl

@@ -0,0 +1,15 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+resources:
+  jobs:
+    test_job:
+      max_concurrent_runs: 1
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Users/{{workspace_user_name}}/notebook
+          new_cluster:
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1

acceptance/bundle/config-remote-sync/flushed_cache/out.test.toml

@@ -0,0 +1,22 @@
+
+=== Deploy bundle
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== Get job ID and modify remote resource
+Modify max_concurrent_runs to 5
+
+=== Flush local cache
+Removing .databricks directory to simulate flushed cache
+ls: .databricks: No such file or directory
+Cache directory removed
+
+=== Check for changes after cache flush
+Detected changes in 1 resource(s):
+
+Resource: resources.jobs.test_job
+  max_concurrent_runs: replace
+
+

acceptance/bundle/config-remote-sync/flushed_cache/output.txt

@@ -0,0 +1,25 @@
+#!/bin/bash
+
+envsubst < databricks.yml.tmpl > databricks.yml
+
+title "Deploy bundle"
+echo
+$CLI bundle deploy
+
+title "Get job ID and modify remote resource"
+echo
+job_id="$(read_id.py test_job)"
+echo "Modify max_concurrent_runs to 5"
+edit_resource.py jobs $job_id <<EOF
+r["max_concurrent_runs"] = 5
+EOF
+
+title "Flush local cache"
+echo
+echo "Removing .databricks directory to simulate flushed cache"
+rm -rf .databricks
+ls -la .databricks 2>&1 || echo "Cache directory removed"
+
+title "Check for changes after cache flush"
+echo
+$CLI bundle config-remote-sync

acceptance/bundle/config-remote-sync/flushed_cache/script

@@ -0,0 +1,15 @@
+Cloud = true
+
+RecordRequests = false
+Ignore = [".databricks", "databricks.yml"]
+
+[Env]
+DATABRICKS_BUNDLE_ENABLE_EXPERIMENTAL_YAML_SYNC = "true"
+
+[EnvMatrix]
+DATABRICKS_BUNDLE_ENGINE = ["direct", "terraform"]
+
+# Normalize ls error messages across platforms
+[[Repls]]
+Old = "ls: cannot access '\\.databricks': No such file or directory"
+New = "ls: .databricks: No such file or directory"

acceptance/bundle/config-remote-sync/flushed_cache/test.toml

@@ -2,10 +2,16 @@ package configsync
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config/engine"
+	"github.com/databricks/cli/bundle/deploy"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/direct"
 	"github.com/databricks/cli/libs/dyn"
@@ -120,6 +126,11 @@ func convertChangeDesc(path string, cd *deployplan.ChangeDesc) (*ConfigChangeDes
 func DetectChanges(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) (Changes, error) {
 	changes := make(Changes)
 
+	err := ensureSnapshotAvailable(ctx, b, engine)
+	if err != nil {
+		return nil, fmt.Errorf("state snapshot not available: %w", err)
+	}
+
 	deployBundle := &direct.DeploymentBundle{}
 	var statePath string
 	if engine.IsDirect() {
@@ -164,3 +175,52 @@ func DetectChanges(ctx context.Context, b *bundle.Bundle, engine engine.EngineTy
 
 	return changes, nil
 }
+
+func ensureSnapshotAvailable(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) error {
+	if engine.IsDirect() {
+		return nil
+	}
+
+	remotePathSnapshot, localPathSnapshot := b.StateFilenameConfigSnapshot(ctx)
+
+	if _, err := os.Stat(localPathSnapshot); err == nil {
+		return nil
+	} else if !errors.Is(err, fs.ErrNotExist) {
+		return fmt.Errorf("checking snapshot file: %w", err)
+	}
+
+	log.Debugf(ctx, "Resources state snapshot not found locally, pulling from remote")
+
+	f, err := deploy.StateFiler(b)
+	if err != nil {
+		return fmt.Errorf("getting state filer: %w", err)
+	}
+
+	r, err := f.Read(ctx, remotePathSnapshot)
+	if err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return fmt.Errorf("resources state snapshot not found remotely at %s", remotePathSnapshot)
+		}
+		return fmt.Errorf("reading remote snapshot: %w", err)
+	}
+	defer r.Close()
+
+	content, err := io.ReadAll(r)
+	if err != nil {
+		return fmt.Errorf("reading snapshot content: %w", err)
+	}
+
+	localStateDir := filepath.Dir(localPathSnapshot)
+	err = os.MkdirAll(localStateDir, 0o700)
+	if err != nil {
+		return fmt.Errorf("creating snapshot directory: %w", err)
+	}
+
+	err = os.WriteFile(localPathSnapshot, content, 0o600)
+	if err != nil {
+		return fmt.Errorf("writing snapshot file: %w", err)
+	}
+
+	log.Debugf(ctx, "Pulled config snapshot from remote to %s", localPathSnapshot)
+	return nil
+}