config-remote-sync: record which state the run read in telemetry

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

@@ -105,7 +105,16 @@ Resource: resources.jobs.my_job
     }
   ],
   "files_changed_count": 1,
-  "files_written_count": 1
+  "files_written_count": 1,
+  "state_serial": 1,
+  "state_lineage": "[UUID]",
+  "state_source": "local",
+  "states_available_count": 2,
+  "state_resource_ids": {
+    "resource_job_ids": [
+      "[MY_JOB_ID]"
+    ]
+  }
 }
 
 >>> [CLI] bundle destroy --auto-approve

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

@@ -134,7 +134,19 @@ Resource: resources.pipelines.my_pipeline
   "files_changed_count": 1,
   "files_written_count": 1,
   "refs_retargeted": 1,
-  "refs_from_siblings": 7
+  "refs_from_siblings": 7,
+  "state_serial": [SERIAL],
+  "state_lineage": "[UUID]",
+  "state_source": "local",
+  "states_available_count": 2,
+  "state_resource_ids": {
+    "resource_job_ids": [
+      "[MY_JOB_ID]"
+    ],
+    "resource_pipeline_ids": [
+      "[MY_PIPELINE_ID]"
+    ]
+  }
 }
 
 >>> [CLI] bundle destroy --auto-approve

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

@@ -13,3 +13,10 @@ DATABRICKS_BUNDLE_ENABLE_EXPERIMENTAL_YAML_SYNC = "true"
 
 [EnvMatrix]
 DATABRICKS_BUNDLE_ENGINE = ["direct", "terraform"]
+
+# The state serial counts state writes, and the terraform path writes more times
+# than the direct path for the same script, so the value differs per engine.
+# The serial is not what this test asserts — normalize it.
+[[Repls]]
+Old = '"state_serial": [0-9]+'
+New = '"state_serial": [SERIAL]'

acceptance/bundle/telemetry/config-remote-sync-error/output.txt

@@ -7,6 +7,8 @@ Exit code: 1
 >>> cat out.requests.txt
 {
   "engine": "terraform",
+  "state_source": "local",
+  "states_available_count": 0,
   "error_message": "state snapshot not available: resources state snapshot not found remotely at resources-config-sync-snapshot.json: state snapshot not found",
   "error_category": "STATE_NOT_FOUND"
 }

acceptance/bundle/telemetry/config-remote-sync-recreate/output.txt

@@ -26,5 +26,14 @@ Resource: resources.pipelines.foo
       "replace_count": 1
     }
   ],
-  "files_changed_count": 1
+  "files_changed_count": 1,
+  "state_serial": 1,
+  "state_lineage": "[UUID]",
+  "state_source": "local",
+  "states_available_count": 2,
+  "state_resource_ids": {
+    "resource_pipeline_ids": [
+      "[FOO_ID]"
+    ]
+  }
 }

acceptance/bundle/telemetry/config-remote-sync-save/output.txt

@@ -28,5 +28,14 @@ Resource: resources.jobs.foo
     }
   ],
   "files_changed_count": 1,
-  "files_written_count": 1
+  "files_written_count": 1,
+  "state_serial": 1,
+  "state_lineage": "[UUID]",
+  "state_source": "local",
+  "states_available_count": 2,
+  "state_resource_ids": {
+    "resource_job_ids": [
+      "[NUMID]"
+    ]
+  }
 }

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

@@ -25,5 +25,14 @@ Resource: resources.jobs.foo
       "replace_count": 1
     }
   ],
-  "files_changed_count": 1
+  "files_changed_count": 1,
+  "state_serial": 1,
+  "state_lineage": "[UUID]",
+  "state_source": "local",
+  "states_available_count": 2,
+  "state_resource_ids": {
+    "resource_job_ids": [
+      "[FOO_ID]"
+    ]
+  }
 }

bundle/configsync/select.go

@@ -10,6 +10,30 @@ import (
 	"github.com/databricks/cli/libs/log"
 )
 
+// IndexDeployedResources indexes the deployed resources by "<type>:<id>",
+// mapping to the plan key ("resources.<type>.<name>"). Indexing by the <type>
+// component means a selector can only ever match a resource of that exact type,
+// never an id that happens to collide across types.
+func IndexDeployedResources(state *dstate.DeploymentState) map[string]string {
+	byTypeID := make(map[string]string)
+	for key := range state.Data.State {
+		id := state.GetResourceID(key)
+		if id == "" {
+			continue
+		}
+		typeAndName, ok := strings.CutPrefix(key, "resources.")
+		if !ok {
+			continue
+		}
+		resourceType, _, ok := strings.Cut(typeAndName, ".")
+		if !ok {
+			continue
+		}
+		byTypeID[resourceType+":"+id] = key
+	}
+	return byTypeID
+}
+
 // ResolveResourceSelectors maps "<type>:<id>" selectors to their plan keys
 // ("resources.<type>.<name>") using the open deployment state (see
 // OpenDeploymentState).
@@ -35,26 +59,7 @@ import (
 // Duplicate selectors are deduplicated; the returned keys preserve the order in
 // which their selectors first appear.
 func ResolveResourceSelectors(ctx context.Context, state *dstate.DeploymentState, selectors []string) ([]string, error) {
-	// Index deployed resources by "<type>:<id>". State keys have the form
-	// "resources.<type>.<name>"; indexing by the <type> component means a
-	// selector can only ever match a resource of that exact type, never an id
-	// that happens to collide across types.
-	byTypeID := make(map[string]string)
-	for key := range state.Data.State {
-		id := state.GetResourceID(key)
-		if id == "" {
-			continue
-		}
-		typeAndName, ok := strings.CutPrefix(key, "resources.")
-		if !ok {
-			continue
-		}
-		resourceType, _, ok := strings.Cut(typeAndName, ".")
-		if !ok {
-			continue
-		}
-		byTypeID[resourceType+":"+id] = key
-	}
+	byTypeID := IndexDeployedResources(state)
 
 	keys := make([]string, 0, len(selectors))
 	var missing []string

bundle/configsync/telemetry.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/databricks/cli/bundle/config/engine"
 	"github.com/databricks/cli/bundle/direct/dresources"
+	"github.com/databricks/cli/bundle/statemgmt"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/telemetry"
@@ -51,10 +52,47 @@ type Stats struct {
 
 	Restore RestoreStats
 
+	// Identity of the resource state the run actually read. Without these, a
+	// "no deployed <type> resource with id" failure cannot be told apart from
+	// reading a stale local cache, another target's state, or no state at all.
+	StateSerial  int64
+	StateLineage string
+	StateSource  string
+	// Pointer so the informative zero (no state file found at all) survives
+	// omitempty on serialization.
+	StatesAvailableCount *int64
+
+	// Resource ids present in the state that was read, and the ids the run was
+	// asked to sync. Comparing the two classifies a selector miss.
+	StateResourceIDs    *protos.BundleConfigRemoteSyncResourceIds
+	SelectedResourceIDs *protos.BundleConfigRemoteSyncResourceIds
+
 	ErrorMessage  string
 	ErrorCategory protos.BundleConfigRemoteSyncErrorCategory
 }
 
+// CollectStateStats records which resource state the run read. Only the
+// state's identity is recorded (serial, system-generated lineage UUID,
+// local/remote origin, candidate count) — never a path or a target name.
+//
+// None of these can contain PII: the lineage is an opaque random UUID minted by
+// the state layer, and the source is one of two fixed literals.
+func (s *Stats) CollectStateStats(desc *statemgmt.StateDesc) {
+	if desc == nil {
+		return
+	}
+	s.StateSerial = int64(desc.Serial)
+	s.StateLineage = desc.Lineage
+	s.StateSource = "remote"
+	if desc.IsLocal {
+		s.StateSource = "local"
+	}
+	// AllStates is only populated when at least one state was found; a
+	// synthesized empty state descriptor leaves it nil.
+	n := int64(len(desc.AllStates))
+	s.StatesAvailableCount = &n
+}
+
 // RestoreStats counts the variable-reference restorations that can leak a
 // current-target-scoped reference into a shared file. The safe paths (kept /
 // compound-realigned) are not counted.
@@ -154,6 +192,70 @@ func resourceTypeFromKey(resourceKey string) string {
 	return parts[1]
 }
 
+// resourceIDLimit caps how many ids of one type are reported. Mirrors
+// phases.ResourceIdLimit: the telemetry upload has a short timeout, so a bundle
+// with very many resources must not produce an unbounded payload.
+const resourceIDLimit = 1000
+
+// collectResourceIDs groups "<type>:<id>" pairs into the per-type id lists.
+// Returns nil when nothing was collected so the field is omitted entirely.
+func collectResourceIDs(typeAndIDs []string) *protos.BundleConfigRemoteSyncResourceIds {
+	out := &protos.BundleConfigRemoteSyncResourceIds{}
+	any := false
+	for _, typeAndID := range typeAndIDs {
+		resourceType, id, ok := strings.Cut(typeAndID, ":")
+		if !ok || id == "" {
+			continue
+		}
+		// Sub-resources (permissions/grants) are indexed under their parent's
+		// type with a path-shaped object id; those are not selectable and the
+		// path would be redacted, so skip them.
+		if strings.Contains(id, "/") {
+			continue
+		}
+		switch resourceType {
+		case "jobs":
+			if len(out.ResourceJobIDs) < resourceIDLimit {
+				out.ResourceJobIDs = append(out.ResourceJobIDs, id)
+				any = true
+			}
+		case "pipelines":
+			if len(out.ResourcePipelineIDs) < resourceIDLimit {
+				out.ResourcePipelineIDs = append(out.ResourcePipelineIDs, id)
+				any = true
+			}
+		case "clusters":
+			if len(out.ResourceClusterIDs) < resourceIDLimit {
+				out.ResourceClusterIDs = append(out.ResourceClusterIDs, id)
+				any = true
+			}
+		case "dashboards":
+			if len(out.ResourceDashboardIDs) < resourceIDLimit {
+				out.ResourceDashboardIDs = append(out.ResourceDashboardIDs, id)
+				any = true
+			}
+		}
+	}
+	if !any {
+		return nil
+	}
+	slices.Sort(out.ResourceJobIDs)
+	slices.Sort(out.ResourcePipelineIDs)
+	slices.Sort(out.ResourceClusterIDs)
+	slices.Sort(out.ResourceDashboardIDs)
+	return out
+}
+
+// CollectSelectedIDs records the ids the run was asked to sync.
+func (s *Stats) CollectSelectedIDs(selectors []string) {
+	s.SelectedResourceIDs = collectResourceIDs(selectors)
+}
+
+// CollectStateIDs records the ids present in the state that was read.
+func (s *Stats) CollectStateIDs(typeAndIDs []string) {
+	s.StateResourceIDs = collectResourceIDs(typeAndIDs)
+}
+
 // LogTelemetry emits the BundleConfigRemoteSyncEvent for this run.
 func (s *Stats) LogTelemetry(ctx context.Context) {
 	defer recoverTelemetry(ctx)
@@ -180,6 +282,12 @@ func (s *Stats) LogTelemetry(ctx context.Context) {
 			FilesWrittenCount:      s.FilesWrittenCount,
 			RefsRetargeted:         s.Restore.Retargeted,
 			RefsFromSiblings:       s.Restore.FromSiblings,
+			StateSerial:            s.StateSerial,
+			StateLineage:           s.StateLineage,
+			StateSource:            s.StateSource,
+			StatesAvailableCount:   s.StatesAvailableCount,
+			StateResourceIDs:       s.StateResourceIDs,
+			SelectedResourceIDs:    s.SelectedResourceIDs,
 			ErrorMessage:           s.ErrorMessage,
 			ErrorCategory:          s.ErrorCategory,
 		},

cmd/bundle/config_remote_sync.go

@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"runtime"
+	"slices"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/configsync"
@@ -75,6 +77,7 @@ Examples:
 			},
 			PostStateFunc: func(ctx context.Context, b *bundle.Bundle, stateDesc *statemgmt.StateDesc) error {
 				stats.Engine = stateDesc.Engine
+				stats.CollectStateStats(stateDesc)
 
 				// Open the deployment state once and reuse it for both planning and
 				// selector resolution (avoids reading the terraform snapshot twice).
@@ -100,7 +103,12 @@ Examples:
 				}
 				stats.CollectChangeStats(ctx, changes)
 
+				// Record the ids present in state and the ids requested: on failure
+				// they are what classifies the miss.
+				stats.CollectStateIDs(slices.Collect(maps.Keys(configsync.IndexDeployedResources(&deployBundle.StateDB))))
+
 				if len(selectIDs) > 0 {
+					stats.CollectSelectedIDs(selectIDs)
 					// Filter after planning, never before: the plan must cover every
 					// resource so ${resources.*} references resolve; only the emitted
 					// changes are restricted to the selected resources.