diff --git a/cmd/wfctl/deploy_providers.go b/cmd/wfctl/deploy_providers.go
index 79c7d8a7..fb588701 100644
--- a/cmd/wfctl/deploy_providers.go
+++ b/cmd/wfctl/deploy_providers.go
@@ -487,6 +487,61 @@ func (r *remoteIaCProvider) DetectDrift(_ context.Context, refs []interfaces.Res
 	return drifts, nil
 }
 
+// DetectDriftWithApplied dispatches to the v2 RPC if the remote plugin
+// supports it, falling back to legacy DetectDrift when the remote returns
+// "method not found". This is the wire-format counterpart of the type-
+// assertion fallback in callers; both layers need the fallback because:
+//   - Caller-side: provider may not implement the interface (type-
+//     assertion fails → legacy path).
+//   - Wire-side: provider implements the interface in newer code but the
+//     LOADED plugin binary on disk is older (manifest accepted but RPC
+//     dispatch returns method-not-found).
+func (r *remoteIaCProvider) DetectDriftWithApplied(ctx context.Context, refs []interfaces.ResourceRef, applied map[string]map[string]any) ([]interfaces.DriftResult, error) {
+	refsAny, err := jsonToAny(refs)
+	if err != nil {
+		return nil, fmt.Errorf("IaCProvider.DetectDriftWithApplied: marshal refs: %w", err)
+	}
+	appliedAny, err := jsonToAny(applied)
+	if err != nil {
+		return nil, fmt.Errorf("IaCProvider.DetectDriftWithApplied: marshal applied: %w", err)
+	}
+	res, err := r.invoker.InvokeService("IaCProvider.DetectDriftWithApplied", map[string]any{
+		"refs": refsAny, "applied": appliedAny,
+	})
+	if err != nil {
+		// Method-not-found from the plugin: fall back to legacy DetectDrift.
+		// Match on substring to tolerate framework-specific error wrappers
+		// (HashiCorp go-plugin emits "method not found"; future schemes may
+		// wrap the literal differently). DO NOT fall back on other errors —
+		// a transient cloud failure must propagate so callers don't silently
+		// lose drift signal.
+		if isMethodNotFound(err) {
+			return r.DetectDrift(ctx, refs)
+		}
+		return nil, err
+	}
+	raw, ok := res["drifts"]
+	if !ok {
+		return nil, nil
+	}
+	var drifts []interfaces.DriftResult
+	if err := anyToStruct(raw, &drifts); err != nil {
+		return nil, fmt.Errorf("IaCProvider.DetectDriftWithApplied: decode result: %w", err)
+	}
+	return drifts, nil
+}
+
+// isMethodNotFound reports whether err is the canonical signal a remote
+// plugin emits when it has no case for the requested RPC method. Tolerant
+// substring match across known wrapper formats.
+func isMethodNotFound(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := err.Error()
+	return strings.Contains(msg, "method not found") || strings.Contains(msg, "Method not found")
+}
+
 func (r *remoteIaCProvider) Import(_ context.Context, cloudID string, resourceType string) (*interfaces.ResourceState, error) {
 	res, err := r.invoker.InvokeService("IaCProvider.Import", map[string]any{
 		"provider_id":   cloudID,
diff --git a/cmd/wfctl/deploy_providers_remote_iac_compat_test.go b/cmd/wfctl/deploy_providers_remote_iac_compat_test.go
new file mode 100644
index 00000000..79a4ca53
--- /dev/null
+++ b/cmd/wfctl/deploy_providers_remote_iac_compat_test.go
@@ -0,0 +1,77 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/GoCodeAlone/workflow/interfaces"
+)
+
+// TestRemoteIaC_OptionalDriftConfigDetector_FallsBackOnLegacyPlugin pins the
+// pipeline's compat story: when a remote plugin does NOT support
+// IaCProvider.DetectDriftWithApplied (legacy plugin: returns "method not
+// found"), the wfctl caller falls through to legacy IaCProvider.DetectDrift.
+//
+// This is the wire-format counterpart of the type-assertion fallback in
+// runInfraApplyRefreshPhase / runInfraStatusDrift; without this gate, a v0.22
+// wfctl talking to a v0.10.3 DO plugin (or an out-of-tree plugin) could trip
+// a hard error instead of falling back.
+func TestRemoteIaC_OptionalDriftConfigDetector_FallsBackOnLegacyPlugin(t *testing.T) {
+	si := &multiMethodStubInvoker{
+		// Method not found = canonical signal a legacy plugin emits when
+		// it doesn't have a case for this RPC name.
+		errByMethod: map[string]error{
+			"IaCProvider.DetectDriftWithApplied": errors.New("method not found: IaCProvider.DetectDriftWithApplied"),
+		},
+		// Legacy DetectDrift returns success.
+		respByMethod: map[string]map[string]any{
+			"IaCProvider.DetectDrift": {
+				"drifts": []any{map[string]any{"name": "x", "type": "infra.test", "drifted": false, "class": "in-sync"}},
+			},
+		},
+	}
+	p := &remoteIaCProvider{invoker: si}
+
+	// Caller-side type-assertion: remoteIaCProvider implements
+	// DetectDriftWithApplied (it always does, since the wfctl-side wrapper
+	// ALWAYS exposes the method). The fallback happens INSIDE the wrapper:
+	// if the remote returns method-not-found, the wrapper retries with
+	// legacy DetectDrift.
+	refs := []interfaces.ResourceRef{{Name: "x", Type: "infra.test"}}
+	drifts, err := p.DetectDriftWithApplied(context.Background(), refs, nil)
+	if err != nil {
+		t.Fatalf("DetectDriftWithApplied should NOT propagate method-not-found; should fall back. err=%v", err)
+	}
+	if len(drifts) != 1 || drifts[0].Class != interfaces.DriftClassInSync {
+		t.Errorf("expected fallback InSync drift; got %+v", drifts)
+	}
+
+	// Verify the second call to invoker WAS the legacy method.
+	if !si.calledMethods["IaCProvider.DetectDrift"] {
+		t.Errorf("expected fallback to call IaCProvider.DetectDrift; called methods: %v", si.calledMethods)
+	}
+}
+
+// multiMethodStubInvoker is a test double for remoteServiceInvoker that supports
+// per-method response and error configuration (unlike the basic stubInvoker
+// which records only a single method/resp/err).
+type multiMethodStubInvoker struct {
+	calledMethods map[string]bool
+	respByMethod  map[string]map[string]any
+	errByMethod   map[string]error
+}
+
+func (s *multiMethodStubInvoker) InvokeService(method string, args map[string]any) (map[string]any, error) {
+	if s.calledMethods == nil {
+		s.calledMethods = map[string]bool{}
+	}
+	s.calledMethods[method] = true
+	if err, ok := s.errByMethod[method]; ok && err != nil {
+		return nil, err
+	}
+	if resp, ok := s.respByMethod[method]; ok {
+		return resp, nil
+	}
+	return nil, nil
+}
diff --git a/cmd/wfctl/deploy_providers_remote_iac_test.go b/cmd/wfctl/deploy_providers_remote_iac_test.go
index dd90800e..36cbdba6 100644
--- a/cmd/wfctl/deploy_providers_remote_iac_test.go
+++ b/cmd/wfctl/deploy_providers_remote_iac_test.go
@@ -473,6 +473,33 @@ func TestRemoteIaC_DetectDrift_Empty(t *testing.T) {
 	}
 }
 
+// ── DetectDriftWithApplied ─────────────────────────────────────────────────────
+
+func TestRemoteIaC_DetectDriftWithApplied_HappyPath(t *testing.T) {
+	si := &stubInvoker{resp: map[string]any{
+		"drifts": []any{map[string]any{
+			"name":    "x",
+			"type":    "infra.test",
+			"drifted": true,
+			"class":   "config",
+			"fields":  []any{"region"},
+		}},
+	}}
+	p := newIaCProvider(si)
+	refs := []interfaces.ResourceRef{{Name: "x", Type: "infra.test"}}
+	applied := map[string]map[string]any{"x": {"region": "nyc1"}}
+	drifts, err := p.DetectDriftWithApplied(context.Background(), refs, applied)
+	if err != nil {
+		t.Fatalf("DetectDriftWithApplied: %v", err)
+	}
+	if si.method != "IaCProvider.DetectDriftWithApplied" {
+		t.Errorf("method: got %q, want IaCProvider.DetectDriftWithApplied", si.method)
+	}
+	if len(drifts) != 1 || drifts[0].Class != interfaces.DriftClassConfig {
+		t.Errorf("drifts: %+v", drifts)
+	}
+}
+
 // ── Import ────────────────────────────────────────────────────────────────────
 
 func TestRemoteIaC_Import(t *testing.T) {
diff --git a/cmd/wfctl/infra_apply.go b/cmd/wfctl/infra_apply.go
index 9fbd9db6..d5b1a3c7 100644
--- a/cmd/wfctl/infra_apply.go
+++ b/cmd/wfctl/infra_apply.go
@@ -490,18 +490,19 @@ func applyWithProviderAndStore(ctx context.Context, provider interfaces.IaCProvi
 
 			now := time.Now().UTC()
 			rs := interfaces.ResourceState{
-				ID:            r.Name,
-				Name:          r.Name,
-				Type:          r.Type,
-				Provider:      providerType,
-				ProviderRef:   providerRef,
-				ProviderID:    r.ProviderID,
-				ConfigHash:    configHashMap(appliedCfg),
-				AppliedConfig: appliedCfg,
-				Outputs:       r.Outputs,
-				Dependencies:  dependencies,
-				CreatedAt:     now,
-				UpdatedAt:     now,
+				ID:                  r.Name,
+				Name:                r.Name,
+				Type:                r.Type,
+				Provider:            providerType,
+				ProviderRef:         providerRef,
+				ProviderID:          r.ProviderID,
+				ConfigHash:          configHashMap(appliedCfg),
+				AppliedConfig:       appliedCfg,
+				AppliedConfigSource: "apply",
+				Outputs:             r.Outputs,
+				Dependencies:        dependencies,
+				CreatedAt:           now,
+				UpdatedAt:           now,
 			}
 			if saveErr := store.SaveResource(ctx, rs); saveErr != nil {
 				return fmt.Errorf("%s/%s: persist state after apply: %w", r.Type, r.Name, saveErr)
@@ -639,18 +640,19 @@ func resourceStateFromLiveOutput(spec interfaces.ResourceSpec, providerType stri
 	appliedConfig := liveConfigFromOutputs(live.Outputs)
 	now := time.Now().UTC()
 	return interfaces.ResourceState{
-		ID:            spec.Name,
-		Name:          spec.Name,
-		Type:          spec.Type,
-		Provider:      providerType,
-		ProviderRef:   resourceSpecProviderRef(spec),
-		ProviderID:    live.ProviderID,
-		ConfigHash:    configHashMap(appliedConfig),
-		AppliedConfig: appliedConfig,
-		Outputs:       cloneMap(live.Outputs),
-		Dependencies:  append([]string(nil), spec.DependsOn...),
-		CreatedAt:     now,
-		UpdatedAt:     now,
+		ID:                  spec.Name,
+		Name:                spec.Name,
+		Type:                spec.Type,
+		Provider:            providerType,
+		ProviderRef:         resourceSpecProviderRef(spec),
+		ProviderID:          live.ProviderID,
+		ConfigHash:          configHashMap(appliedConfig),
+		AppliedConfig:       appliedConfig,
+		AppliedConfigSource: "adoption",
+		Outputs:             cloneMap(live.Outputs),
+		Dependencies:        append([]string(nil), spec.DependsOn...),
+		CreatedAt:           now,
+		UpdatedAt:           now,
 	}, nil
 }
 
@@ -959,18 +961,19 @@ func applyPrecomputedPlanWithStore(ctx context.Context, plan interfaces.IaCPlan,
 
 			now := time.Now().UTC()
 			rs := interfaces.ResourceState{
-				ID:            r.Name,
-				Name:          r.Name,
-				Type:          r.Type,
-				Provider:      providerType,
-				ProviderRef:   providerRef,
-				ProviderID:    r.ProviderID,
-				ConfigHash:    configHashMap(appliedCfg),
-				AppliedConfig: appliedCfg,
-				Outputs:       r.Outputs,
-				Dependencies:  dependencies,
-				CreatedAt:     now,
-				UpdatedAt:     now,
+				ID:                  r.Name,
+				Name:                r.Name,
+				Type:                r.Type,
+				Provider:            providerType,
+				ProviderRef:         providerRef,
+				ProviderID:          r.ProviderID,
+				ConfigHash:          configHashMap(appliedCfg),
+				AppliedConfig:       appliedCfg,
+				AppliedConfigSource: "apply",
+				Outputs:             r.Outputs,
+				Dependencies:        dependencies,
+				CreatedAt:           now,
+				UpdatedAt:           now,
 			}
 			if saveErr := store.SaveResource(ctx, rs); saveErr != nil {
 				return fmt.Errorf("%s/%s: persist state after apply: %w", r.Type, r.Name, saveErr)
diff --git a/cmd/wfctl/infra_apply_refresh.go b/cmd/wfctl/infra_apply_refresh.go
index 4015e366..3b4a2d40 100644
--- a/cmd/wfctl/infra_apply_refresh.go
+++ b/cmd/wfctl/infra_apply_refresh.go
@@ -46,7 +46,21 @@ func runInfraApplyRefreshPhase(
 		return nil
 	}
 
-	results, err := provider.DetectDrift(ctx, refs)
+	// Use DriftConfigDetector when the provider supports it (optional interface).
+	// Short-circuits to legacy DetectDrift when appliedMap is nil (no "apply"-
+	// provenance entries available) to avoid unnecessary RPC round-trips.
+	var results []interfaces.DriftResult
+	var err error
+	if d, ok := provider.(interfaces.DriftConfigDetector); ok {
+		appliedMap := buildAppliedSpecMap(states, refs)
+		if appliedMap != nil {
+			results, err = d.DetectDriftWithApplied(ctx, refs, appliedMap)
+		} else {
+			results, err = provider.DetectDrift(ctx, refs)
+		}
+	} else {
+		results, err = provider.DetectDrift(ctx, refs)
+	}
 	if err != nil {
 		// Transient or auth error — propagate; do NOT prune anything.
 		return fmt.Errorf("detect drift: %w", err)
diff --git a/cmd/wfctl/infra_apply_test.go b/cmd/wfctl/infra_apply_test.go
index 6beb8016..d5156c17 100644
--- a/cmd/wfctl/infra_apply_test.go
+++ b/cmd/wfctl/infra_apply_test.go
@@ -1592,3 +1592,79 @@ modules:
 		t.Errorf("stderr = %q, want it to contain 'warning'", stderrOutput)
 	}
 }
+
+// ── TestApply_StateRecordsAppliedConfigSource* ─────────────────────────────────
+
+// TestApply_StateRecordsAppliedConfigSourceApply asserts that applyWithProviderAndStore
+// writes AppliedConfigSource="apply" on successful resource creation.
+func TestApply_StateRecordsAppliedConfigSourceApply(t *testing.T) {
+	spec := interfaces.ResourceSpec{
+		Name:   "res-A",
+		Type:   "infra.test",
+		Config: map[string]any{"k": "v"},
+	}
+	fake := &stateReturningProvider{
+		applyResult: &interfaces.ApplyResult{
+			Resources: []interfaces.ResourceOutput{
+				{Name: "res-A", Type: "infra.test", ProviderID: "id-A", Outputs: map[string]any{"k": "v"}},
+			},
+		},
+	}
+	store := &fakeStateStore{}
+
+	if err := applyWithProviderAndStore(t.Context(), fake, "test", []interfaces.ResourceSpec{spec}, nil, store, io.Discard, ""); err != nil {
+		t.Fatalf("apply: %v", err)
+	}
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if len(store.saved) != 1 {
+		t.Fatalf("expected 1 saved state; got %d", len(store.saved))
+	}
+	saved := store.saved[0]
+	if saved.AppliedConfigSource != "apply" {
+		t.Errorf("AppliedConfigSource: got %q, want %q", saved.AppliedConfigSource, "apply")
+	}
+}
+
+// TestAdoption_StateRecordsAppliedConfigSourceAdoption asserts that state saved
+// via the adoption path (adoptExistingResources) records AppliedConfigSource="adoption".
+func TestAdoption_StateRecordsAppliedConfigSourceAdoption(t *testing.T) {
+	spec := interfaces.ResourceSpec{
+		Name:   "site-dns",
+		Type:   "infra.dns",
+		Config: map[string]any{"provider": "do-provider", "domain": "example.com"},
+	}
+	driver := &readDriver{
+		expectedProviderID: "example.com",
+		readOut: &interfaces.ResourceOutput{
+			Name:       "site-dns",
+			Type:       "infra.dns",
+			ProviderID: "do-domain-123",
+			Outputs:    map[string]any{"domain": "example.com"},
+		},
+	}
+	provider := &readBackedProvider{driver: driver}
+	store := &fakeStateStore{}
+
+	if err := applyWithProviderAndStore(t.Context(), provider, "digitalocean", []interfaces.ResourceSpec{spec}, nil, store, io.Discard, ""); err != nil {
+		t.Fatalf("adopt: %v", err)
+	}
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	// The first saved state is from adoption; subsequent saves are from apply.
+	var adoptedState *interfaces.ResourceState
+	for i := range store.saved {
+		if store.saved[i].Name == "site-dns" && store.saved[i].AppliedConfigSource == "adoption" {
+			adoptedState = &store.saved[i]
+			break
+		}
+	}
+	if adoptedState == nil {
+		t.Fatalf("expected adopted state with AppliedConfigSource=adoption; saved=%+v", store.saved)
+	}
+	if adoptedState.AppliedConfigSource != "adoption" {
+		t.Errorf("AppliedConfigSource: got %q, want %q", adoptedState.AppliedConfigSource, "adoption")
+	}
+}
diff --git a/cmd/wfctl/infra_drift_applied.go b/cmd/wfctl/infra_drift_applied.go
new file mode 100644
index 00000000..de7755f2
--- /dev/null
+++ b/cmd/wfctl/infra_drift_applied.go
@@ -0,0 +1,51 @@
+package main
+
+import "github.com/GoCodeAlone/workflow/interfaces"
+
+// buildAppliedSpecMap walks states + refs and returns the per-ref applied-
+// config map for DriftConfigDetector.DetectDriftWithApplied. Entries whose
+// AppliedConfigSource is anything other than "apply" are OMITTED from the
+// returned map — providers cannot meaningfully compute config-drift against
+// adoption-shaped Outputs (would yield false-positives) and legacy state
+// (no provenance recorded) defaults to adoption treatment per ADR 0010.
+//
+// Empty AppliedConfig (nil map or zero-length map) is also omitted: there
+// is no spec to compare against.
+//
+// Returns nil when no safe entries exist, so callers can short-circuit the
+// type-assertion entirely and fall back to legacy DetectDrift.
+func buildAppliedSpecMap(states []interfaces.ResourceState, refs []interfaces.ResourceRef) map[string]map[string]any {
+	if len(states) == 0 || len(refs) == 0 {
+		return nil
+	}
+	byName := make(map[string]*interfaces.ResourceState, len(states))
+	for i := range states {
+		byName[states[i].Name] = &states[i]
+	}
+	out := make(map[string]map[string]any, len(refs))
+	for _, ref := range refs {
+		st, ok := byName[ref.Name]
+		if !ok {
+			continue
+		}
+		// Only "apply"-provenance entries carry true user-supplied config.
+		// "adoption", "" (legacy), or any other value defaults to adoption
+		// treatment: omit to avoid false-positive config-drift.
+		if st.AppliedConfigSource != "apply" {
+			continue
+		}
+		if len(st.AppliedConfig) == 0 {
+			continue
+		}
+		// Shallow copy so providers cannot mutate the caller's state map.
+		cfg := make(map[string]any, len(st.AppliedConfig))
+		for k, v := range st.AppliedConfig {
+			cfg[k] = v
+		}
+		out[ref.Name] = cfg
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
diff --git a/cmd/wfctl/infra_drift_applied_test.go b/cmd/wfctl/infra_drift_applied_test.go
new file mode 100644
index 00000000..027848a6
--- /dev/null
+++ b/cmd/wfctl/infra_drift_applied_test.go
@@ -0,0 +1,83 @@
+package main
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/GoCodeAlone/workflow/interfaces"
+)
+
+func TestBuildAppliedSpecMap_OmitsAdoptionAndLegacy(t *testing.T) {
+	states := []interfaces.ResourceState{
+		{Name: "apply-resource", AppliedConfig: map[string]any{"k": "v"}, AppliedConfigSource: "apply"},
+		{Name: "adoption-resource", AppliedConfig: map[string]any{"k": "v"}, AppliedConfigSource: "adoption"},
+		{Name: "legacy-resource", AppliedConfig: map[string]any{"k": "v"}, AppliedConfigSource: ""},
+		{Name: "nil-config-resource", AppliedConfig: nil, AppliedConfigSource: "apply"},
+		{Name: "empty-map-config-resource", AppliedConfig: map[string]any{}, AppliedConfigSource: "apply"},
+	}
+	refs := []interfaces.ResourceRef{
+		{Name: "apply-resource"},
+		{Name: "adoption-resource"},
+		{Name: "legacy-resource"},
+		{Name: "nil-config-resource"},
+		{Name: "empty-map-config-resource"},
+		{Name: "missing-from-state"},
+	}
+
+	got := buildAppliedSpecMap(states, refs)
+	want := map[string]map[string]any{
+		"apply-resource": {"k": "v"},
+		// adoption-resource: omitted (refuse false-positive on adoption)
+		// legacy-resource: omitted (legacy default-to-adoption)
+		// nil-config-resource: omitted (nil AppliedConfig)
+		// empty-map-config-resource: omitted (len 0 — same branch as nil)
+		// missing-from-state: omitted (no state)
+	}
+
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("got %v, want %v", got, want)
+	}
+}
+
+func TestBuildAppliedSpecMap_NilStatesReturnsNil(t *testing.T) {
+	refs := []interfaces.ResourceRef{{Name: "x"}}
+	got := buildAppliedSpecMap(nil, refs)
+	if got != nil {
+		t.Errorf("expected nil for empty states, got %v", got)
+	}
+}
+
+func TestBuildAppliedSpecMap_NilRefsReturnsNil(t *testing.T) {
+	states := []interfaces.ResourceState{{Name: "x", AppliedConfig: map[string]any{"k": "v"}, AppliedConfigSource: "apply"}}
+	got := buildAppliedSpecMap(states, nil)
+	if got != nil {
+		t.Errorf("expected nil for nil refs, got %v", got)
+	}
+}
+
+func TestBuildAppliedSpecMap_NoSafeEntriesReturnsNil(t *testing.T) {
+	// When all entries are adoption/legacy, result is nil (not empty map).
+	// Callers may use nil check to short-circuit the type-assertion.
+	states := []interfaces.ResourceState{
+		{Name: "x", AppliedConfig: map[string]any{"k": "v"}, AppliedConfigSource: "adoption"},
+	}
+	refs := []interfaces.ResourceRef{{Name: "x"}}
+	got := buildAppliedSpecMap(states, refs)
+	if got != nil {
+		t.Errorf("expected nil when no safe entries; got %v", got)
+	}
+}
+
+func TestBuildAppliedSpecMap_ShallowCopyPreventsCallerMutation(t *testing.T) {
+	states := []interfaces.ResourceState{
+		{Name: "x", AppliedConfig: map[string]any{"k": "v"}, AppliedConfigSource: "apply"},
+	}
+	refs := []interfaces.ResourceRef{{Name: "x"}}
+
+	got := buildAppliedSpecMap(states, refs)
+	// Mutate the returned map; verify the source state is not affected.
+	got["x"]["k"] = "mutated"
+	if states[0].AppliedConfig["k"] == "mutated" {
+		t.Errorf("buildAppliedSpecMap must return a shallow copy; source state was mutated")
+	}
+}
diff --git a/cmd/wfctl/infra_status_drift.go b/cmd/wfctl/infra_status_drift.go
index df55f3d8..7e51fdb2 100644
--- a/cmd/wfctl/infra_status_drift.go
+++ b/cmd/wfctl/infra_status_drift.go
@@ -100,7 +100,20 @@ func driftInfraModules(ctx context.Context, cfgFile, envName string) error {
 			}()
 		}
 
-		results, err := provider.DetectDrift(ctx, g.refs)
+		// Use DriftConfigDetector when the provider supports it (optional interface).
+		// Short-circuits to legacy DetectDrift when appliedMap is nil (no "apply"-
+		// provenance entries available) to avoid unnecessary RPC round-trips.
+		var results []interfaces.DriftResult
+		if d, ok := provider.(interfaces.DriftConfigDetector); ok {
+			appliedMap := buildAppliedSpecMap(states, g.refs)
+			if appliedMap != nil {
+				results, err = d.DetectDriftWithApplied(ctx, g.refs, appliedMap)
+			} else {
+				results, err = provider.DetectDrift(ctx, g.refs)
+			}
+		} else {
+			results, err = provider.DetectDrift(ctx, g.refs)
+		}
 		if err != nil {
 			fmt.Printf("WARNING: drift detection for provider %q: %v\n", moduleRef, err)
 			return false
diff --git a/decisions/0007-iac-driftconfigdetector-optional-interface.md b/decisions/0007-iac-driftconfigdetector-optional-interface.md
new file mode 100644
index 00000000..55977d0a
--- /dev/null
+++ b/decisions/0007-iac-driftconfigdetector-optional-interface.md
@@ -0,0 +1,63 @@
+# ADR 0007 — IaC DriftConfigDetector Optional Interface
+
+- **Status:** Accepted
+- **Date:** 2026-05-06
+- **Deciders:** Claude (autonomous design pipeline), Jon Langevin (mandate)
+- **Refs:** workflow-plugin-digitalocean#47, [IaC state-truth design (GoCodeAlone/workflow-plugin-digitalocean)](https://github.com/GoCodeAlone/workflow-plugin-digitalocean/blob/main/docs/plans/2026-05-06-iac-state-truth-and-tc2-closeout-design.md)
+
+## Context
+
+`IaCProvider.DetectDrift(ctx, refs)` is existence-only: it classifies resources as `Ghost`, `InSync`, or `Unknown`, but never `DriftClassConfig`. Drivers like `VPCDriver` and `AppPlatformDriver` cannot compute meaningful config-drift without the spec that was applied — they receive only cloud-live state, not the operator's intended config. Issue workflow-plugin-digitalocean#47 tracked two possible remediation paths:
+
+1. **Required-signature change** — extend `IaCProvider.DetectDrift` to accept a third argument: `applied map[string]map[string]any`. Every plugin implementing the interface would need a signature update, including aws/gcp/azure and any out-of-tree plugin.
+
+2. **Optional capability interface** — add a new interface `DriftConfigDetector` that plugins MAY implement. Callers type-assert; plugins that don't implement it fall back to the existing existence-only `DetectDrift`. This is the pattern already established in the repo by `ComputePlanVersionDeclarer`, `ProviderValidator`, `Enumerator`, and `UpsertSupporter`.
+
+## Decision
+
+Add `DriftConfigDetector` as an **OPTIONAL interface** (Path 2). `IaCProvider.DetectDrift` is unchanged.
+
+```go
+type DriftConfigDetector interface {
+    DetectDriftWithApplied(ctx context.Context, resources []ResourceRef, applied map[string]map[string]any) ([]DriftResult, error)
+}
+```
+
+Callers type-assert and fall back:
+
+```go
+if d, ok := provider.(DriftConfigDetector); ok {
+    results, err = d.DetectDriftWithApplied(ctx, refs, appliedMap)
+} else {
+    results, err = provider.DetectDrift(ctx, refs)
+}
+```
+
+The `applied` map is keyed by `ref.Name` and sourced from `ResourceState.AppliedConfig`. The sentinel field `ResourceState.AppliedConfigSource` (see ADR 0010) discriminates "apply" (true user-supplied config) from "adoption" (Outputs reflow). Providers MUST refuse to compute config-drift on adoption-shaped entries to avoid false-positives.
+
+## Consequences
+
+**Positive:**
+- Sibling plugins (aws/gcp/azure) require **zero code changes** — they continue to satisfy `IaCProvider` and fall through to existence-only detection.
+- Out-of-tree plugins remain binary-compatible and compile-compatible.
+- The optional-declarer pattern is already established in this repo; a new developer encountering `DriftConfigDetector` has existing precedents to follow.
+- Detection capability can be added incrementally, one plugin at a time, without a coordinated multi-repo release.
+
+**Negative:**
+- Callers accumulate type-assertion branches. Mitigated by the established pattern and the small number of call sites (two: `runInfraApplyRefreshPhase`, `runInfraStatusDrift`).
+- Plugin authors must discover the optional interface via docs or code search. Tracked as a follow-up: consolidate the opt-in capability list in `docs/IAC_PLUGIN_AUTHORING.md`.
+
+## Rejected Alternative: Required-Signature Change
+
+Adversarial design review #1 (Critical finding) rejected changing `DetectDrift(ctx, refs)` to `DetectDrift(ctx, refs, applied)` because:
+- Every plugin implementing `IaCProvider` must update its signature simultaneously — a coordinated multi-repo change.
+- Breaks binary compatibility for any out-of-tree plugin (missing method → plugin fails to load).
+- Conflicts with the repo's established optional-declarer precedent.
+
+## References
+
+- `interfaces/iac_provider.go` — `DriftConfigDetector` declaration
+- `interfaces/iac_state.go` — `ResourceState.AppliedConfigSource` (ADR 0010)
+- `cmd/wfctl/infra_apply_refresh.go` — caller type-assertion
+- `cmd/wfctl/infra_status_drift.go` — caller type-assertion
+- `cmd/wfctl/infra_drift_applied.go` — `buildAppliedSpecMap` helper
diff --git a/decisions/0010-applied-config-source-on-resourcestate.md b/decisions/0010-applied-config-source-on-resourcestate.md
new file mode 100644
index 00000000..2e180d33
--- /dev/null
+++ b/decisions/0010-applied-config-source-on-resourcestate.md
@@ -0,0 +1,59 @@
+# ADR 0010 — AppliedConfigSource Field on ResourceState
+
+- **Status:** Accepted
+- **Date:** 2026-05-06
+- **Deciders:** Claude (autonomous design pipeline), Jon Langevin (mandate)
+- **Refs:** ADR 0007, [IaC state-truth design (GoCodeAlone/workflow-plugin-digitalocean)](https://github.com/GoCodeAlone/workflow-plugin-digitalocean/blob/main/docs/plans/2026-05-06-iac-state-truth-and-tc2-closeout-design.md)
+
+## Context
+
+`DriftConfigDetector.DetectDriftWithApplied` (ADR 0007) receives the `applied` map from `ResourceState.AppliedConfig`. However, `AppliedConfig` can contain two distinct kinds of data:
+
+1. **True user config** — set by `applyWithProviderAndStore` and `applyPrecomputedPlanWithStore` from `ResourceSpec.Config`. Comparing this against a fresh cloud Read is safe and meaningful.
+
+2. **Adoption-shaped Outputs reflow** — set by `adoptExistingResources` via `liveConfigFromOutputs(live.Outputs)`. When no embedded "config" key exists in Outputs, `liveConfigFromOutputs` falls back to `cloneMap(live.Outputs)`. Comparing such an entry against a fresh Read yields false-positive config-drift (Outputs contain fields like `id`, `created_at`, `endpoint` that are not user-controllable via spec).
+
+Without discrimination, `DetectDriftWithApplied` cannot tell safe from unsafe entries. Two placement options were considered:
+
+1. **Magic key in AppliedConfig** — store the sentinel as `applied_config["_source"]`. Reads directly during `DetectDriftWithApplied` without a separate lookup.
+2. **Dedicated field on ResourceState** — `ResourceState.AppliedConfigSource string`. Clean separation of concerns; sentinel is not part of the user-config payload; JSON encoding is `omitempty` so legacy state files remain unaffected.
+
+## Decision
+
+Add `AppliedConfigSource string` as a **dedicated field on `ResourceState`** (Option 2), JSON-tagged `applied_config_source,omitempty`.
+
+Valid values:
+- `"apply"` — `AppliedConfig` came from a user-supplied `ResourceSpec.Config`. Config-drift detection is safe and meaningful for this entry.
+- `"adoption"` — `AppliedConfig` was reflowed from live cloud Outputs. Config-drift detection MUST be skipped to avoid false-positives.
+- `""` (empty / absent from JSON) — legacy state written before this field existed. Consumers MUST treat as `"adoption"` (conservative default).
+
+Write sites:
+- `applyWithProviderAndStore` → writes `"apply"`
+- `applyPrecomputedPlanWithStore` → writes `"apply"`
+- `resourceStateFromLiveOutput` (called by `adoptExistingResources`) → writes `"adoption"`
+
+## Consequences
+
+**Positive:**
+- The sentinel is structurally separate from the user config payload — a provider parsing `AppliedConfig` cannot accidentally consume or mutate it.
+- `omitempty` tag ensures pre-existing state-store files (no `applied_config_source` key) continue to decode cleanly; empty string is the conservative "adoption" default.
+- Backward-compat is bidirectional: old code reading new state JSON silently drops the unknown field (Go `encoding/json` default); new code reading old state JSON sees empty string → adoption default.
+- The field is self-documenting in the state store for debugging.
+
+**Negative:**
+- State store entries grow by ~30 bytes per resource (JSON key + value). Acceptable at typical resource counts (10–100 per tenant).
+- A third write site may be added in the future (e.g. `wfctl infra import` if it adopts the pattern). Each new write site must be audited for correct source labeling.
+
+## Rejected Alternative: Magic Key in AppliedConfig
+
+Adversarial design review #1 cycle 2 (Important finding "sentinel placement") rejected storing the sentinel as `applied_config["_source"]` because:
+- Pollutes the user-config payload; providers iterating `AppliedConfig` for their own purposes must filter out the sentinel key.
+- A provider that passes `AppliedConfig` unchanged to a downstream system (e.g. a Terraform variable map) would transmit the sentinel as a Terraform variable, causing unexpected behavior.
+- Harder to discover and document; JSON state files show the field mixed in with user config rather than at the struct level.
+
+## References
+
+- `interfaces/iac_state.go` — `ResourceState.AppliedConfigSource` field declaration
+- `cmd/wfctl/infra_apply.go` — write sites (`applyWithProviderAndStore`, `applyPrecomputedPlanWithStore`, `resourceStateFromLiveOutput`)
+- `cmd/wfctl/infra_drift_applied.go` — `buildAppliedSpecMap` (reads the field to filter safe entries)
+- ADR 0007 — `DriftConfigDetector` optional interface (consumer of this field)
diff --git a/interfaces/iac_provider.go b/interfaces/iac_provider.go
index 95b51a6b..a217e52e 100644
--- a/interfaces/iac_provider.go
+++ b/interfaces/iac_provider.go
@@ -84,6 +84,27 @@ type Enumerator interface {
 	EnumerateByTag(ctx context.Context, tag string) ([]ResourceRef, error)
 }
 
+// DriftConfigDetector is an OPTIONAL interface a provider MAY implement to
+// surface config-drift in addition to the existence-only Ghost / InSync /
+// Unknown classifications produced by DetectDrift.
+//
+// applied is the per-ref applied_config map recorded in state. Callers
+// pass it from ResourceState.AppliedConfig; missing or empty entries
+// instruct the provider to fall back to existence-only behavior for
+// that ref. The map key is ref.Name (matches ResourceState.Name).
+//
+// Callers MUST type-assert against this interface and fall back to
+// IaCProvider.DetectDrift(refs) on the negative case. Providers that do
+// not implement DriftConfigDetector continue to work unchanged.
+//
+// Providers SHOULD only return DriftClassConfig when they have high
+// confidence the applied entry represents user-supplied config (not
+// adoption-shaped Outputs reflow); see ResourceState.AppliedConfigSource
+// (iac_state.go) for the canonical discriminator.
+type DriftConfigDetector interface {
+	DetectDriftWithApplied(ctx context.Context, resources []ResourceRef, applied map[string]map[string]any) ([]DriftResult, error)
+}
+
 // BootstrapResult contains metadata returned by a successful BootstrapStateBackend call.
 type BootstrapResult struct {
 	// Bucket is the name of the created or confirmed state bucket/container.
diff --git a/interfaces/iac_provider_test.go b/interfaces/iac_provider_test.go
index e7364613..40e52d3c 100644
--- a/interfaces/iac_provider_test.go
+++ b/interfaces/iac_provider_test.go
@@ -222,3 +222,26 @@ func TestProviderValidator_TypeAssertion_NonImplementor(t *testing.T) {
 		t.Errorf("nonValidatingProvider must NOT satisfy ProviderValidator (interface must remain optional)")
 	}
 }
+
+// TestDriftConfigDetector_OptionalInterface verifies the optional-declarer
+// pattern works as intended: an IaCProvider impl that does NOT implement
+// DriftConfigDetector type-asserts to false (caller falls back to existence-
+// only DetectDrift); an impl that DOES implement it type-asserts to true.
+func TestDriftConfigDetector_OptionalInterface(t *testing.T) {
+	var minimal interfaces.IaCProvider = nonValidatingProvider{}
+	if _, ok := minimal.(interfaces.DriftConfigDetector); ok {
+		t.Errorf("nonValidatingProvider should NOT satisfy DriftConfigDetector")
+	}
+
+	var capable interfaces.IaCProvider = &capableIaCProvider{}
+	if _, ok := capable.(interfaces.DriftConfigDetector); !ok {
+		t.Errorf("capableIaCProvider MUST satisfy DriftConfigDetector")
+	}
+}
+
+// capableIaCProvider extends nonValidatingProvider with DriftConfigDetector.
+type capableIaCProvider struct{ nonValidatingProvider }
+
+func (*capableIaCProvider) DetectDriftWithApplied(context.Context, []interfaces.ResourceRef, map[string]map[string]any) ([]interfaces.DriftResult, error) {
+	return nil, nil
+}
diff --git a/interfaces/iac_state.go b/interfaces/iac_state.go
index c0dd3caf..bf9fa382 100644
--- a/interfaces/iac_state.go
+++ b/interfaces/iac_state.go
@@ -27,19 +27,29 @@ type IaCStateStore interface {
 
 // ResourceState is the persisted state record for a single managed resource.
 type ResourceState struct {
-	ID             string         `json:"id"`
-	Name           string         `json:"name"`
-	Type           string         `json:"type"`
-	Provider       string         `json:"provider"`
-	ProviderRef    string         `json:"provider_ref,omitempty"`
-	ProviderID     string         `json:"provider_id"`
-	ConfigHash     string         `json:"config_hash"`
-	AppliedConfig  map[string]any `json:"applied_config"`
-	Outputs        map[string]any `json:"outputs"`
-	Dependencies   []string       `json:"dependencies"`
-	CreatedAt      time.Time      `json:"created_at"`
-	UpdatedAt      time.Time      `json:"updated_at"`
-	LastDriftCheck time.Time      `json:"last_drift_check,omitempty"`
+	ID            string         `json:"id"`
+	Name          string         `json:"name"`
+	Type          string         `json:"type"`
+	Provider      string         `json:"provider"`
+	ProviderRef   string         `json:"provider_ref,omitempty"`
+	ProviderID    string         `json:"provider_id"`
+	ConfigHash    string         `json:"config_hash"`
+	AppliedConfig map[string]any `json:"applied_config"`
+	// AppliedConfigSource records the provenance of AppliedConfig:
+	//   - "apply": AppliedConfig came from a user-supplied ResourceSpec.Config
+	//     (set by applyWithProviderAndStore / applyPrecomputedPlanWithStore).
+	//   - "adoption": AppliedConfig was reflowed from live cloud Outputs
+	//     (set by adoptExistingResources). Comparing such entries against
+	//     a fresh Read produces false-positive config-drift; consumers MUST
+	//     refuse to compute config-drift on adoption-shaped entries.
+	//   - "" (empty): legacy state written before this field existed.
+	//     Consumers MUST treat as "adoption" (conservative default).
+	AppliedConfigSource string         `json:"applied_config_source,omitempty"`
+	Outputs             map[string]any `json:"outputs"`
+	Dependencies        []string       `json:"dependencies"`
+	CreatedAt           time.Time      `json:"created_at"`
+	UpdatedAt           time.Time      `json:"updated_at"`
+	LastDriftCheck      time.Time      `json:"last_drift_check,omitempty"`
 }
 
 // IaCPlan is the complete execution plan for a set of infrastructure changes.
diff --git a/interfaces/iac_state_test.go b/interfaces/iac_state_test.go
index 90f4e034..d8ea843d 100644
--- a/interfaces/iac_state_test.go
+++ b/interfaces/iac_state_test.go
@@ -1,6 +1,7 @@
 package interfaces
 
 import (
+	"bytes"
 	"encoding/json"
 	"testing"
 )
@@ -153,3 +154,85 @@ func containsString(s, substr string) bool {
 	}
 	return false
 }
+
+func TestResourceState_AppliedConfigSourceJSONRoundtrip(t *testing.T) {
+	cases := []struct {
+		name string
+		src  string
+	}{
+		{"apply provenance", "apply"},
+		{"adoption provenance", "adoption"},
+		{"empty (legacy state)", ""},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			in := ResourceState{Name: "x", Type: "infra.foo", AppliedConfigSource: tc.src}
+			data, err := json.Marshal(in)
+			if err != nil {
+				t.Fatalf("marshal: %v", err)
+			}
+			var out ResourceState
+			if err := json.Unmarshal(data, &out); err != nil {
+				t.Fatalf("unmarshal: %v", err)
+			}
+			if out.AppliedConfigSource != tc.src {
+				t.Errorf("source: got %q, want %q", out.AppliedConfigSource, tc.src)
+			}
+		})
+	}
+}
+
+func TestResourceState_AppliedConfigSourceOmitemptyWhenLegacy(t *testing.T) {
+	// Legacy state: AppliedConfigSource not set. JSON output must omit
+	// the field so old state-store readers don't trip on unknown keys.
+	in := ResourceState{Name: "x", Type: "infra.foo"}
+	data, err := json.Marshal(in)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	if bytes.Contains(data, []byte("applied_config_source")) {
+		t.Errorf("legacy state should omit applied_config_source; got %s", data)
+	}
+}
+
+// TestResourceState_OldReaderTolerates_NewWriter pins state-store roundtrip
+// compat: state JSON written by NEW code (with applied_config_source field)
+// MUST still decode without error in OLD code (which has no AppliedConfigSource
+// field). Go's encoding/json silently ignores unknown fields by default; this
+// test pins that contract.
+func TestResourceState_OldReaderTolerates_NewWriter(t *testing.T) {
+	in := ResourceState{Name: "x", Type: "infra.foo", AppliedConfigSource: "apply"}
+	data, err := json.Marshal(in)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+
+	// "Old" reader = a struct with NO AppliedConfigSource field.
+	type oldResourceState struct {
+		Name          string         `json:"name"`
+		Type          string         `json:"type"`
+		AppliedConfig map[string]any `json:"applied_config"`
+	}
+	var old oldResourceState
+	if err := json.Unmarshal(data, &old); err != nil {
+		t.Fatalf("old reader rejected new state: %v", err)
+	}
+	if old.Name != "x" {
+		t.Errorf("old reader: Name corrupted; got %q", old.Name)
+	}
+}
+
+// TestResourceState_NewReaderTolerates_OldWriter: legacy state with NO
+// applied_config_source key must decode cleanly in new code with the field
+// defaulted to empty string (which downstream code treats as "adoption" per
+// ADR 0010, conservative default).
+func TestResourceState_NewReaderTolerates_OldWriter(t *testing.T) {
+	legacyJSON := []byte(`{"name":"x","type":"infra.foo","applied_config":{"k":"v"}}`)
+	var out ResourceState
+	if err := json.Unmarshal(legacyJSON, &out); err != nil {
+		t.Fatalf("new reader rejected legacy state: %v", err)
+	}
+	if out.AppliedConfigSource != "" {
+		t.Errorf("AppliedConfigSource on legacy state: got %q, want empty", out.AppliedConfigSource)
+	}
+}