feat: add bicep snapshot as primary RG classification source

Implements @vhvb1989's proposal to use bicep snapshot predictedResources
as the primary classification mechanism for resource group ownership.

Snapshot path (when available):
- RG in predictedResources → owned (template creates it)
- RG NOT in predictedResources → external (uses existing keyword)
- Tier 4 (locks/foreign resources) runs as defense-in-depth

Graceful fallback to existing Tier 1-4 pipeline when snapshot is
unavailable (older Bicep CLI, non-bicepparam mode, errors).

Changes:
- Add SnapshotPredictedRGs field to ClassifyOptions
- Add classifyFromSnapshot() with Tier 4 defense-in-depth
- Add getSnapshotPredictedRGs() to BicepProvider for snapshot extraction
- Add 8 unit tests for snapshot classification
- Update architecture.md with snapshot-primary data flow

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: jongio

cli/azd/pkg/azapi/resource_group_classifier.go

@@ -42,9 +42,25 @@ type ManagementLock struct {
 
 // ClassifyOptions configures the classification pipeline.
 type ClassifyOptions struct {
+	// SnapshotPredictedRGs is the set of resource group names (lowercased) that the
+	// Bicep template declares as created resources (not 'existing' references).
+	// Populated from `bicep snapshot` → predictedResources filtered by RG type.
+	//
+	// When non-nil, snapshot-based classification replaces Tiers 1-3:
+	//   - RG in set → owned (template creates it)
+	//   - RG not in set → external (template references it as existing)
+	//   - Tier 4 still runs on all owned candidates (defense-in-depth)
+	//
+	// When nil, the full Tier 1-4 pipeline runs as fallback (older Bicep CLI,
+	// non-bicepparam mode, or snapshot failure).
+	SnapshotPredictedRGs map[string]bool
+
 	// ForceMode runs only Tier 1 (zero API calls). External RGs identified by
 	// deployment operations are still protected; unknown RGs are treated as owned.
 	// Tier 2/3/4 callbacks are not invoked.
+	//
+	// When combined with SnapshotPredictedRGs, snapshot classification is used
+	// (deterministic, zero API calls) and Tier 4 is skipped.
 	ForceMode bool
 	// Interactive enables per-RG prompts for unknown and foreign-resource RGs.
 	// When false, unknown/unverified RGs are always skipped without deletion.
@@ -120,6 +136,12 @@ func ClassifyResourceGroups(
 
 	result := &ClassifyResult{}
 
+	// --- Snapshot path: when predictedResources are available, use them as primary signal ---
+	// This replaces Tiers 1-3 with a deterministic, offline classification from bicep snapshot.
+	if opts.SnapshotPredictedRGs != nil {
+		return classifyFromSnapshot(ctx, rgNames, opts, result)
+	}
+
 	// --- Tier 1: classify all RGs from deployment operations (zero extra API calls) ---
 	owned, unknown := classifyTier1(operations, rgNames, result)
 
@@ -276,6 +298,126 @@ func ClassifyResourceGroups(
 	return result, nil
 }
 
+// classifyFromSnapshot uses the Bicep snapshot predictedResources to classify RGs.
+// RGs whose names appear in the predicted set are owned (the template creates them).
+// RGs not in the predicted set are external (referenced via the `existing` keyword).
+//
+// Tier 4 (lock + foreign-resource veto) still runs on owned candidates unless ForceMode
+// is active, providing defense-in-depth even when snapshot says "owned."
+func classifyFromSnapshot(
+	ctx context.Context,
+	rgNames []string,
+	opts ClassifyOptions,
+	result *ClassifyResult,
+) (*ClassifyResult, error) {
+	var owned []string
+	for _, rg := range rgNames {
+		if opts.SnapshotPredictedRGs[strings.ToLower(rg)] {
+			owned = append(owned, rg)
+		} else {
+			result.Skipped = append(result.Skipped, ClassifiedSkip{
+				Name:   rg,
+				Reason: "external (snapshot: not in predictedResources)",
+			})
+		}
+	}
+
+	// ForceMode + snapshot: deterministic classification, zero API calls, no Tier 4.
+	if opts.ForceMode {
+		result.Owned = owned
+		return result, nil
+	}
+
+	// --- Tier 4: veto checks on all snapshot-owned candidates (defense-in-depth) ---
+	// Same logic as the tier pipeline path. Even if the snapshot says "owned," a
+	// management lock or foreign resources should still prevent deletion.
+	type veto struct {
+		rg     string
+		reason string
+	}
+	type pendingPrompt struct {
+		rg     string
+		reason string
+	}
+	vetoCh := make(chan veto, len(owned))
+	promptCh := make(chan pendingPrompt, len(owned))
+	sem := make(chan struct{}, cTier4Parallelism)
+	var wg sync.WaitGroup
+	for _, rg := range owned {
+		select {
+		case sem <- struct{}{}:
+			if ctx.Err() != nil {
+				<-sem
+				vetoCh <- veto{
+					rg:     rg,
+					reason: "error during safety check: " + ctx.Err().Error(),
+				}
+				continue
+			}
+		case <-ctx.Done():
+			vetoCh <- veto{
+				rg:     rg,
+				reason: "error during safety check: " + ctx.Err().Error(),
+			}
+			continue
+		}
+		wg.Go(func() {
+			defer func() { <-sem }()
+			reason, vetoed, needsPrompt, err := classifyTier4(ctx, rg, opts)
+			if err != nil {
+				log.Printf(
+					"ERROR: classify rg=%s tier=4: safety check failed: %v (treating as veto)",
+					rg, err,
+				)
+				vetoCh <- veto{
+					rg:     rg,
+					reason: fmt.Sprintf("error during safety check: %s", err.Error()),
+				}
+				return
+			}
+			if needsPrompt {
+				promptCh <- pendingPrompt{rg: rg, reason: reason}
+				return
+			}
+			if vetoed {
+				vetoCh <- veto{rg: rg, reason: reason}
+			}
+		})
+	}
+	wg.Wait()
+	close(vetoCh)
+	close(promptCh)
+
+	vetoedSet := make(map[string]string, len(owned))
+	for v := range vetoCh {
+		vetoedSet[v.rg] = v.reason
+	}
+
+	for p := range promptCh {
+		if opts.Interactive && opts.Prompter != nil {
+			accept, err := opts.Prompter(p.rg, p.reason)
+			if err != nil {
+				return nil, fmt.Errorf("classify rg=%s tier=4 prompt: %w", p.rg, err)
+			}
+			if !accept {
+				vetoedSet[p.rg] = p.reason
+			}
+		} else {
+			vetoedSet[p.rg] = p.reason
+		}
+	}
+
+	for _, rg := range owned {
+		if reason, vetoed := vetoedSet[rg]; vetoed {
+			result.Skipped = append(result.Skipped, ClassifiedSkip{Name: rg, Reason: reason})
+		} else {
+			result.Owned = append(result.Owned, rg)
+		}
+	}
+
+	return result, nil
+}
+
 // classifyTier1 uses deployment operations to classify RGs with zero extra API calls.
 // Returns (owned, unknown) slices. External RGs are appended directly to result.Skipped.
 func classifyTier1(

cli/azd/pkg/azapi/resource_group_classifier_test.go

@@ -1265,4 +1265,178 @@ func TestClassifyResourceGroups_ForceMode(t *testing.T) {
 		require.Len(t, res.Skipped, 1)
 		assert.Equal(t, rgExternal, res.Skipped[0].Name)
 	})
+
+}
+
+func TestClassifyResourceGroups_Snapshot(t *testing.T) {
+	t.Parallel()
+
+	const (
+		rgA     = "rg-alpha"
+		rgB     = "rg-beta"
+		rgC     = "rg-gamma"
+		envName = "myenv"
+	)
+
+	rgOp := "Microsoft.Resources/resourceGroups"
+
+	t.Run("owned and external", func(t *testing.T) {
+		t.Parallel()
+		predicted := map[string]bool{
+			"rg-alpha": true,
+			"rg-beta":  true,
+		}
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			SnapshotPredictedRGs: predicted,
+		}
+		// rgC is NOT in the predicted set → external
+		res, err := ClassifyResourceGroups(t.Context(), nil, []string{rgA, rgB, rgC}, opts)
+		require.NoError(t, err)
+		assert.ElementsMatch(t, []string{rgA, rgB}, res.Owned)
+		require.Len(t, res.Skipped, 1)
+		assert.Equal(t, rgC, res.Skipped[0].Name)
+		assert.Contains(t, res.Skipped[0].Reason, "snapshot")
+	})
+
+	t.Run("case insensitive matching", func(t *testing.T) {
+		t.Parallel()
+		predicted := map[string]bool{
+			"rg-alpha": true, // lowercased in the map
+		}
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			SnapshotPredictedRGs: predicted,
+		}
+		// "RG-Alpha" should match "rg-alpha" via ToLower
+		res, err := ClassifyResourceGroups(t.Context(), nil, []string{"RG-Alpha"}, opts)
+		require.NoError(t, err)
+		assert.Equal(t, []string{"RG-Alpha"}, res.Owned)
+		assert.Empty(t, res.Skipped)
+	})
+
+	t.Run("all external", func(t *testing.T) {
+		t.Parallel()
+		predicted := map[string]bool{
+			"rg-unrelated": true, // no overlap with test RGs
+		}
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			SnapshotPredictedRGs: predicted,
+		}
+		res, err := ClassifyResourceGroups(t.Context(), nil, []string{rgA, rgB}, opts)
+		require.NoError(t, err)
+		assert.Empty(t, res.Owned)
+		assert.Len(t, res.Skipped, 2)
+	})
+
+	t.Run("ForceMode skips Tier4", func(t *testing.T) {
+		t.Parallel()
+		predicted := map[string]bool{
+			"rg-alpha": true,
+		}
+		var tier4Called bool
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			ForceMode:            true,
+			SnapshotPredictedRGs: predicted,
+			ListResourceGroupLocks: func(_ context.Context, _ string) ([]*ManagementLock, error) {
+				tier4Called = true
+				return nil, nil
+			},
+		}
+		res, err := ClassifyResourceGroups(t.Context(), nil, []string{rgA, rgB}, opts)
+		require.NoError(t, err)
+		assert.False(t, tier4Called, "Tier 4 should not run when ForceMode + snapshot")
+		assert.Equal(t, []string{rgA}, res.Owned)
+		require.Len(t, res.Skipped, 1)
+		assert.Equal(t, rgB, res.Skipped[0].Name)
+	})
+
+	t.Run("Tier4 lock veto", func(t *testing.T) {
+		t.Parallel()
+		predicted := map[string]bool{
+			"rg-alpha": true,
+			"rg-beta":  true,
+		}
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			SnapshotPredictedRGs: predicted,
+			ListResourceGroupLocks: func(_ context.Context, rgName string) ([]*ManagementLock, error) {
+				if rgName == rgA {
+					return []*ManagementLock{{Name: "mylock", LockType: "CanNotDelete"}}, nil
+				}
+				return nil, nil
+			},
+		}
+		res, err := ClassifyResourceGroups(t.Context(), nil, []string{rgA, rgB}, opts)
+		require.NoError(t, err)
+		// rgA is snapshot-owned but vetoed by lock
+		assert.Equal(t, []string{rgB}, res.Owned)
+		require.Len(t, res.Skipped, 1)
+		assert.Equal(t, rgA, res.Skipped[0].Name)
+		assert.Contains(t, res.Skipped[0].Reason, "lock")
+	})
+
+	t.Run("Tier4 foreign resource veto", func(t *testing.T) {
+		t.Parallel()
+		predicted := map[string]bool{
+			"rg-alpha": true,
+		}
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			Interactive:          false,
+			SnapshotPredictedRGs: predicted,
+			ListResourceGroupResources: func(_ context.Context, _ string) ([]*ResourceWithTags, error) {
+				return []*ResourceWithTags{
+					{Name: "foreign-vm", Type: "Microsoft.Compute/virtualMachines", Tags: map[string]*string{
+						"azd-env-name": strPtr("otherenv"),
+					}},
+				}, nil
+			},
+		}
+		res, err := ClassifyResourceGroups(t.Context(), nil, []string{rgA}, opts)
+		require.NoError(t, err)
+		assert.Empty(t, res.Owned)
+		require.Len(t, res.Skipped, 1)
+		assert.Contains(t, res.Skipped[0].Reason, "foreign")
+	})
+
+	t.Run("nil falls back to tier pipeline", func(t *testing.T) {
+		t.Parallel()
+		// SnapshotPredictedRGs is nil → should use Tier 1 pipeline
+		ops := []*armresources.DeploymentOperation{
+			makeOperation("Create", rgOp, rgA),
+		}
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			SnapshotPredictedRGs: nil, // explicitly nil
+		}
+		res, err := ClassifyResourceGroups(t.Context(), ops, []string{rgA, rgB}, opts)
+		require.NoError(t, err)
+		// rgA is owned via Tier 1 Create, rgB is unknown → skipped (no Tier 2/3 callbacks)
+		assert.Equal(t, []string{rgA}, res.Owned)
+		require.Len(t, res.Skipped, 1)
+		assert.Equal(t, rgB, res.Skipped[0].Name)
+	})
+
+	t.Run("overrides deployment operations", func(t *testing.T) {
+		t.Parallel()
+		// Even though operations say rgA is "Read" (external), snapshot says it's owned.
+		// Snapshot should take precedence when available.
+		ops := []*armresources.DeploymentOperation{
+			makeOperation("Read", rgOp, rgA),
+		}
+		predicted := map[string]bool{
+			"rg-alpha": true,
+		}
+		opts := ClassifyOptions{
+			EnvName:              envName,
+			SnapshotPredictedRGs: predicted,
+		}
+		res, err := ClassifyResourceGroups(t.Context(), ops, []string{rgA}, opts)
+		require.NoError(t, err)
+		assert.Equal(t, []string{rgA}, res.Owned)
+		assert.Empty(t, res.Skipped)
+	})
 }