From baf6fa2b9f3e2a319a5980de7a589a2c2f66ef9d Mon Sep 17 00:00:00 2001
From: Jon Langevin <codingsloth@pm.me>
Date: Sat, 3 Oct 2026 12:51:55 -0400
Subject: [PATCH] feat(iac): pass persisted state on update

Reread durable state at dispatch, including saved plans. Preserve legacy clients while state-dependent providers fail closed without prior state.
---
 cmd/wfctl/iac_typed_adapter.go               |  20 +-
 cmd/wfctl/iac_typed_adapter_test.go          |  73 +++++
 cmd/wfctl/infra_apply.go                     |  23 ++
 cmd/wfctl/infra_apply_test.go                | 131 +++++++++
 docs/WFCTL.md                                |  20 ++
 iac/providerclient/adapter.go                |  20 ++
 iac/providerclient/adapter_test.go           |  34 +++
 iac/wfctlhelpers/apply.go                    |  42 ++-
 iac/wfctlhelpers/apply_update_delete_test.go | 118 ++++++++
 interfaces/iac_canonical_types.go            |  13 +
 interfaces/iac_resource_driver.go            |  11 +
 plugin/external/proto/iac.pb.go              | 261 +++++++++--------
 plugin/external/proto/iac.proto              |   3 +
 plugin/external/sdk/iacserver.go             |  74 ++++-
 plugin/external/sdk/iacserver_test.go        | 287 ++++++++++++++++++-
 15 files changed, 992 insertions(+), 138 deletions(-)

diff --git a/cmd/wfctl/iac_typed_adapter.go b/cmd/wfctl/iac_typed_adapter.go
index a95e6ad4..eaaa7181 100644
--- a/cmd/wfctl/iac_typed_adapter.go
+++ b/cmd/wfctl/iac_typed_adapter.go
@@ -811,11 +811,29 @@ func (d *typedResourceDriver) Read(ctx context.Context, ref interfaces.ResourceR
 }
 
 func (d *typedResourceDriver) Update(ctx context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec) (*interfaces.ResourceOutput, error) {
+	return d.update(ctx, ref, spec, nil)
+}
+
+func (d *typedResourceDriver) UpdateWithState(ctx context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
+	if err := interfaces.ValidateUpdatePriorState(ref, spec, prior); err != nil {
+		return nil, err
+	}
+	return d.update(ctx, ref, spec, prior)
+}
+
+func (d *typedResourceDriver) update(ctx context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
 	pbSpec, err := specToPB(spec)
 	if err != nil {
 		return nil, fmt.Errorf("typed driver %s: encode Update spec: %w", d.resourceType, err)
 	}
-	resp, err := d.client.Update(ctx, &pb.ResourceUpdateRequest{ResourceType: d.resourceType, Ref: refToPB(ref), Spec: pbSpec})
+	var pbPrior *pb.ResourceState
+	if prior != nil {
+		pbPrior, err = stateToPB(prior)
+		if err != nil {
+			return nil, fmt.Errorf("typed driver: encode Update prior state: %w", err)
+		}
+	}
+	resp, err := d.client.Update(ctx, &pb.ResourceUpdateRequest{ResourceType: d.resourceType, Ref: refToPB(ref), Spec: pbSpec, PriorState: pbPrior})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/wfctl/iac_typed_adapter_test.go b/cmd/wfctl/iac_typed_adapter_test.go
index 3f070472..2defeffe 100644
--- a/cmd/wfctl/iac_typed_adapter_test.go
+++ b/cmd/wfctl/iac_typed_adapter_test.go
@@ -23,6 +23,7 @@ import (
 	"context"
 	"errors"
 	"net"
+	"reflect"
 	"strings"
 	"testing"
 
@@ -33,8 +34,80 @@ import (
 
 	"github.com/GoCodeAlone/workflow/interfaces"
 	pb "github.com/GoCodeAlone/workflow/plugin/external/proto"
+	"github.com/GoCodeAlone/workflow/plugin/external/sdk"
 )
 
+func TestTypedResourceDriver_UpdateWithState_OptionalContract(t *testing.T) {
+	driver := &typedResourceDriver{resourceType: "infra.fixture"}
+	if _, ok := any(driver).(interfaces.ResourceStateUpdater); !ok {
+		t.Fatal("wfctl driver omits the optional state-aware update contract")
+	}
+}
+
+type updateStateTypedProvider struct {
+	pb.UnimplementedIaCProviderRequiredServer
+	pb.UnimplementedResourceDriverServer
+	prior       *interfaces.ResourceState
+	stateCalls  int
+	legacyCalls int
+}
+
+func (p *updateStateTypedProvider) Update(_ context.Context, req *pb.ResourceUpdateRequest) (*pb.ResourceUpdateResponse, error) {
+	p.legacyCalls++
+	return &pb.ResourceUpdateResponse{Output: &pb.ResourceOutput{Name: req.GetRef().GetName(), Type: req.GetRef().GetType(), ProviderId: req.GetRef().GetProviderId()}}, nil
+}
+
+func (p *updateStateTypedProvider) UpdateWithState(_ context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
+	if prior == nil {
+		p.legacyCalls++
+		return &interfaces.ResourceOutput{Name: spec.Name, Type: spec.Type, ProviderID: ref.ProviderID}, nil
+	}
+	p.stateCalls++
+	p.prior = prior
+	return &interfaces.ResourceOutput{Name: spec.Name, Type: spec.Type, ProviderID: ref.ProviderID, Outputs: prior.Outputs}, nil
+}
+
+func TestTypedResourceDriver_UpdateWithState_SDKRoundTrip(t *testing.T) {
+	provider := &updateStateTypedProvider{}
+	srv := grpc.NewServer()
+	if err := sdk.RegisterAllIaCProviderServices(srv, provider); err != nil {
+		t.Fatal(err)
+	}
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	go func() { _ = srv.Serve(lis) }()
+	t.Cleanup(srv.Stop)
+	t.Cleanup(func() { _ = lis.Close() })
+	conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+	adapter := newTypedIaCAdapter(conn, map[string]bool{iacServiceResourceDriver: true})
+	driver, err := adapter.ResourceDriver("infra.fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	prior := &interfaces.ResourceState{ID: "fixture", Name: "fixture", Type: "infra.fixture", Provider: "fixture", ProviderRef: "provider-instance", ProviderID: "fixture-id", ConfigHash: "fixture-hash", AppliedConfigSource: "apply", Dependencies: []string{"dependency"}, AppliedConfig: map[string]any{"credential_ref": "secrets://fixture/credential"}, Outputs: map[string]any{"rotation_generation": float64(9)}}
+	ref := interfaces.ResourceRef{Name: prior.Name, Type: prior.Type, ProviderID: prior.ProviderID}
+	spec := interfaces.ResourceSpec{Name: prior.Name, Type: prior.Type}
+	updater := driver.(interfaces.ResourceStateUpdater)
+	out, err := updater.UpdateWithState(t.Context(), ref, spec, prior)
+	if err != nil || out == nil || !reflect.DeepEqual(provider.prior, prior) || !reflect.DeepEqual(out.Outputs, prior.Outputs) || provider.stateCalls != 1 || provider.legacyCalls != 0 {
+		t.Fatalf("wfctl/SDK state round trip failed: err=%v state_calls=%d legacy_calls=%d", err, provider.stateCalls, provider.legacyCalls)
+	}
+	wrong := *prior
+	wrong.ProviderID = "other-resource"
+	if _, err := updater.UpdateWithState(t.Context(), ref, spec, &wrong); !errors.Is(err, interfaces.ErrValidation) || provider.stateCalls != 1 {
+		t.Fatal("mismatched state reached the wfctl Update RPC")
+	}
+	if _, err := driver.Update(t.Context(), ref, spec); err != nil || provider.legacyCalls != 1 || provider.stateCalls != 1 {
+		t.Fatal("wfctl legacy Update did not preserve absent-state dispatch")
+	}
+}
+
 // TestTypedAdapter_SatisfiesIaCProvider asserts the adapter's Go
 // interface conformance at runtime so refactors that drop a method
 // fail tests rather than relying on the package-scope compile guard.
diff --git a/cmd/wfctl/infra_apply.go b/cmd/wfctl/infra_apply.go
index d72849e9..11a30e36 100644
--- a/cmd/wfctl/infra_apply.go
+++ b/cmd/wfctl/infra_apply.go
@@ -673,6 +673,29 @@ func statePersistenceHooks(
 	cleanupOptions ...cleanupProviderOptions,
 ) wfctlhelpers.ApplyPlanHooks {
 	hooks := wfctlhelpers.ApplyPlanHooks{
+		LookupCurrentState: func(ctx context.Context, ref interfaces.ResourceRef) (*interfaces.ResourceState, error) {
+			states, err := store.ListResources(ctx)
+			if err != nil {
+				return nil, err
+			}
+			var prior *interfaces.ResourceState
+			for i := range states {
+				if states[i].Name != ref.Name {
+					continue
+				}
+				if prior != nil {
+					return nil, fmt.Errorf("%w: update prior resource identity is ambiguous", interfaces.ErrValidation)
+				}
+				prior = &states[i]
+			}
+			if prior == nil {
+				return nil, fmt.Errorf("%w: update prior resource state unavailable", interfaces.ErrResourceNotFound)
+			}
+			if err := interfaces.ValidateUpdatePriorState(ref, interfaces.ResourceSpec{Name: ref.Name, Type: ref.Type}, prior); err != nil {
+				return nil, err
+			}
+			return prior, nil
+		},
 		ResolveSecret: func(ctx context.Context, key string) (string, error) {
 			if value, ok := hydratedOut[key]; ok {
 				return value, nil
diff --git a/cmd/wfctl/infra_apply_test.go b/cmd/wfctl/infra_apply_test.go
index f1d4b04d..ade51ce0 100644
--- a/cmd/wfctl/infra_apply_test.go
+++ b/cmd/wfctl/infra_apply_test.go
@@ -3,6 +3,7 @@ package main
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -18,6 +19,136 @@ import (
 	"github.com/GoCodeAlone/workflow/interfaces"
 )
 
+type updatePriorCLIDriver struct {
+	readDriver
+	prior *interfaces.ResourceState
+	calls int
+}
+
+func (d *updatePriorCLIDriver) Update(context.Context, interfaces.ResourceRef, interfaces.ResourceSpec) (*interfaces.ResourceOutput, error) {
+	return nil, errors.New("state-aware CLI fixture used legacy Update")
+}
+
+func (d *updatePriorCLIDriver) Diff(context.Context, interfaces.ResourceSpec, *interfaces.ResourceOutput) (*interfaces.DiffResult, error) {
+	return nil, errors.New("saved-plan fixture called Diff")
+}
+
+func (d *updatePriorCLIDriver) UpdateWithState(_ context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
+	d.calls++
+	d.prior = prior
+	return &interfaces.ResourceOutput{Name: spec.Name, Type: spec.Type, ProviderID: ref.ProviderID, Outputs: prior.Outputs}, nil
+}
+
+type updatePriorCLIProvider struct {
+	applyCapture
+	driver *updatePriorCLIDriver
+}
+
+func (p *updatePriorCLIProvider) ResourceDriver(string) (interfaces.ResourceDriver, error) {
+	return p.driver, nil
+}
+
+// The durable store advances immediately after the initial reference snapshot.
+// The update must reread it rather than reuse that captured slice.
+type advancingPriorStateStore struct {
+	infraStateStore
+	latest interfaces.ResourceState
+	lists  int
+}
+
+func (s *advancingPriorStateStore) ListResources(ctx context.Context) ([]interfaces.ResourceState, error) {
+	s.lists++
+	states, err := s.infraStateStore.ListResources(ctx)
+	if err == nil && s.lists == 1 {
+		err = s.SaveResource(ctx, s.latest)
+	}
+	return states, err
+}
+
+func TestApplyPrecomputedPlan_UpdateWithState_LatestPersistedState(t *testing.T) {
+	dir := t.TempDir()
+	configFile := filepath.Join(dir, "state.yaml")
+	stateDir := filepath.Join(dir, "state")
+	configData := fmt.Sprintf("modules:\n  - name: state\n    type: iac.state\n    config:\n      backend: filesystem\n      directory: %q\n", stateDir)
+	if err := os.WriteFile(configFile, []byte(configData), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	store, err := resolveStateStore(configFile, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	stale := interfaces.ResourceState{ID: "fixture", Name: "fixture", Type: "infra.fixture", Provider: "fixture", ProviderID: "fixture-id", Outputs: map[string]any{"rotation_generation": float64(1)}}
+	encoded, err := json.Marshal(interfaces.IaCPlan{Actions: []interfaces.PlanAction{{Action: "update", Resource: interfaces.ResourceSpec{Name: stale.Name, Type: stale.Type}, Current: &stale}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	current := stale
+	current.Outputs = map[string]any{"rotation_generation": float64(7)}
+	if err := store.SaveResource(t.Context(), current); err != nil {
+		t.Fatal(err)
+	}
+	restarted, err := resolveStateStore(configFile, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	latest := stale
+	latest.Outputs = map[string]any{"rotation_generation": float64(9)}
+	advancing := &advancingPriorStateStore{infraStateStore: restarted, latest: latest}
+	var plan interfaces.IaCPlan
+	if err := json.Unmarshal(encoded, &plan); err != nil {
+		t.Fatal(err)
+	}
+	driver := &updatePriorCLIDriver{}
+	provider := &updatePriorCLIProvider{driver: driver}
+	if err := applyPrecomputedPlanWithStore(t.Context(), plan, provider, "fixture", advancing, io.Discard, "", "", nil); err != nil {
+		t.Fatal(err)
+	}
+	if driver.calls != 1 || driver.prior == nil || driver.prior.Outputs["rotation_generation"] != float64(9) || advancing.lists < 2 {
+		t.Fatal("saved-plan CLI update reused a stale plan or captured reference-state slice")
+	}
+}
+
+func TestStatePersistenceHooks_UpdateWithState_RejectsMissingOrWrongIdentity(t *testing.T) {
+	for _, name := range []string{"missing", "wrong_name", "wrong_type", "wrong_id", "duplicate"} {
+		t.Run(name, func(t *testing.T) {
+			state := interfaces.ResourceState{Name: "fixture", Type: "infra.fixture", ProviderID: "fixture-id"}
+			store := &fakeStateStore{}
+			switch name {
+			case "missing":
+			case "wrong_name":
+				state.Name = "other"
+			case "wrong_type":
+				state.Type = "infra.other"
+			case "wrong_id":
+				state.ProviderID = "other-id"
+			case "duplicate":
+				// A named backend can return ambiguous records; reject them.
+			}
+			if name != "missing" {
+				store.saved = []interfaces.ResourceState{state}
+			}
+			var lookupStore infraStateStore = store
+			if name == "duplicate" {
+				lookupStore = &duplicatePriorStateStore{state: state}
+			}
+			hooks := statePersistenceHooks(lookupStore, nil, &applyCapture{}, "fixture", "fixture-plan", nil)
+			_, err := hooks.LookupCurrentState(t.Context(), interfaces.ResourceRef{Name: "fixture", Type: "infra.fixture", ProviderID: "fixture-id"})
+			if err == nil {
+				t.Fatal("missing or mismatched state was accepted")
+			}
+		})
+	}
+}
+
+type duplicatePriorStateStore struct {
+	noopStateStore
+	state interfaces.ResourceState
+}
+
+func (s duplicatePriorStateStore) ListResources(context.Context) ([]interfaces.ResourceState, error) {
+	return []interfaces.ResourceState{s.state, s.state}, nil
+}
+
 // ── fakes ──────────────────────────────────────────────────────────────────────
 
 // applyCapture is a minimal IaCProvider that records Plan and Apply calls.
diff --git a/docs/WFCTL.md b/docs/WFCTL.md
index edb0b94a..757f3e53 100644
--- a/docs/WFCTL.md
+++ b/docs/WFCTL.md
@@ -2124,6 +2124,26 @@ Provider plugins map ownership to their native mechanism, such as `managed-by:<o
 
 Reconcile cloud infrastructure to match the desired state declared in the config. Computes a diff plan via each `iac.provider` and dispatches creates/updates/replaces/deletes through the loaded provider plugin. State is persisted after every successful action so the next run sees the cloud-truth.
 
+State-aware resource updates reread the configured state backend at dispatch
+time, including when `--plan` skips `ComputePlan`. Neither the saved plan's
+`Current` record nor an earlier `Diff` supplies authoritative update state.
+Missing, ambiguous, or mismatched resource identity prevents the update before
+provider mutation. Persisted config and outputs remain reference-only; this
+does not persist or reconstruct credential values.
+
+Plugin authors can opt into `interfaces.ResourceStateUpdater.UpdateWithState`
+without changing the existing `ResourceDriver.Update` interface. The typed
+Update RPC carries additive `prior_state`; both wfctl and the reusable
+providerclient adapter support it. The SDK detects `ResourceStateUpdater` on
+the typed provider service and dispatches that method after identity validation.
+Its other RPC methods are unchanged. Legacy providers retain their original
+Update dispatch. A mixed provider's native updater may receive nil state from
+a legacy RPC and delegate non-state-dependent updates. It must reject updates
+that lack required prior state rather than using a private `Diff` cache.
+Existing public adapter callers without `LookupCurrentState` retain the legacy
+Update path; a state-dependent provider must reject that call before mutation.
+Supplying the hook enables strict prior-state lookup and identity validation.
+
 Provider-declared sensitive input paths must contain references, not literal
 credentials. Plans, hashes, persistence hooks, and applied config retain those
 references. Only the driver receives a deep runtime-resolved copy, including
diff --git a/iac/providerclient/adapter.go b/iac/providerclient/adapter.go
index b985e206..46ac44a1 100644
--- a/iac/providerclient/adapter.go
+++ b/iac/providerclient/adapter.go
@@ -295,14 +295,34 @@ func (r *resourceDriverAdapter) Read(ctx context.Context, ref interfaces.Resourc
 
 // Update calls ResourceDriver.Update with the resource ref and desired spec.
 func (r *resourceDriverAdapter) Update(ctx context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec) (*interfaces.ResourceOutput, error) {
+	return r.update(ctx, ref, spec, nil)
+}
+
+// UpdateWithState sends dispatch-time persisted state on the existing Update RPC.
+func (r *resourceDriverAdapter) UpdateWithState(ctx context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
+	if err := interfaces.ValidateUpdatePriorState(ref, spec, prior); err != nil {
+		return nil, err
+	}
+	return r.update(ctx, ref, spec, prior)
+}
+
+func (r *resourceDriverAdapter) update(ctx context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
 	pbSpec, err := specToPB(spec)
 	if err != nil {
 		return nil, fmt.Errorf("providerclient: encode Update spec: %w", err)
 	}
+	var pbPrior *pb.ResourceState
+	if prior != nil {
+		pbPrior, err = stateToPB(prior)
+		if err != nil {
+			return nil, fmt.Errorf("providerclient: encode Update prior state: %w", err)
+		}
+	}
 	resp, err := r.client.Update(ctx, &pb.ResourceUpdateRequest{
 		ResourceType: r.resourceType,
 		Ref:          refToPB(ref),
 		Spec:         pbSpec,
+		PriorState:   pbPrior,
 	})
 	if err != nil {
 		return nil, mapResourceDriverGRPCError(err, "Update")
diff --git a/iac/providerclient/adapter_test.go b/iac/providerclient/adapter_test.go
index f2a70158..f05a2c3e 100644
--- a/iac/providerclient/adapter_test.go
+++ b/iac/providerclient/adapter_test.go
@@ -20,6 +20,40 @@ import (
 // Compile-time guard: *Adapter must satisfy interfaces.IaCProvider.
 var _ interfaces.IaCProvider = (*providerclient.Adapter)(nil)
 
+func TestAdapter_UpdateWithState_OptionalContract(t *testing.T) {
+	srv := grpc.NewServer()
+	server := &fakeResourceDriverServer{}
+	pb.RegisterResourceDriverServer(srv, server)
+	adapter := providerclient.New(startFakeServer(t, srv), map[string]bool{providerclient.IaCServiceResourceDriver: true})
+	driver, err := adapter.ResourceDriver("infra.fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	updater, ok := driver.(interfaces.ResourceStateUpdater)
+	if !ok {
+		t.Fatal("providerclient driver omits the optional state-aware update contract")
+	}
+	prior := &interfaces.ResourceState{Name: "fixture", Type: "infra.fixture", ProviderID: "fixture-id", Outputs: map[string]any{"rotation_generation": float64(7)}}
+	_, err = updater.UpdateWithState(t.Context(), interfaces.ResourceRef{Name: prior.Name, Type: prior.Type, ProviderID: prior.ProviderID}, interfaces.ResourceSpec{Name: prior.Name, Type: prior.Type}, prior)
+	if err != nil || server.lastUpdateReq == nil {
+		t.Fatalf("state-aware Update RPC failed: %v", err)
+	}
+	state := server.lastUpdateReq.GetPriorState()
+	var outputs map[string]any
+	if state == nil || json.Unmarshal(state.GetOutputsJson(), &outputs) != nil || outputs["rotation_generation"] != float64(7) || state.GetProviderId() != prior.ProviderID || state.GetName() != prior.Name || state.GetType() != prior.Type {
+		t.Fatal("Update RPC discarded prior state")
+	}
+	server.lastUpdateReq = nil
+	wrong := *prior
+	wrong.ProviderID = "other-resource"
+	if _, err := updater.UpdateWithState(t.Context(), interfaces.ResourceRef{Name: prior.Name, Type: prior.Type, ProviderID: prior.ProviderID}, interfaces.ResourceSpec{Name: prior.Name, Type: prior.Type}, &wrong); !errors.Is(err, interfaces.ErrValidation) || server.lastUpdateReq != nil {
+		t.Fatal("mismatched state reached the providerclient Update RPC")
+	}
+	if _, err := driver.Update(t.Context(), interfaces.ResourceRef{Name: prior.Name, Type: prior.Type, ProviderID: prior.ProviderID}, interfaces.ResourceSpec{Name: prior.Name, Type: prior.Type}); err != nil || server.lastUpdateReq == nil || server.lastUpdateReq.GetPriorState() != nil {
+		t.Fatal("providerclient legacy Update did not preserve absent-state RPC")
+	}
+}
+
 // *Adapter must NOT unconditionally satisfy interfaces.IaCProviderRegionLister —
 // that interface is gated behind the RegionListerProvider accessor. The
 // negative assertion is enforced by TestAdapter_RegionLister_Unadvertised below
diff --git a/iac/wfctlhelpers/apply.go b/iac/wfctlhelpers/apply.go
index fb237cd8..931e7e24 100644
--- a/iac/wfctlhelpers/apply.go
+++ b/iac/wfctlhelpers/apply.go
@@ -94,6 +94,9 @@ type ApplyPlanHooks struct {
 	// CurrentState seeds references to unchanged siblings that have no action.
 	// It is an in-memory dispatch source, never embedded in the plan.
 	CurrentState []interfaces.ResourceState
+	// LookupCurrentState rereads durable state at each state-aware update.
+	// CurrentState and PlanAction.Current are not authoritative update inputs.
+	LookupCurrentState func(context.Context, interfaces.ResourceRef) (*interfaces.ResourceState, error)
 	// ResolveSecret retrieves a secret key only at the dispatch boundary.
 	ResolveSecret func(context.Context, string) (string, error)
 	// ResourceDeletionComplete reads durable cloud-deletion evidence. A true
@@ -719,9 +722,9 @@ func snapshotKeys(m map[string]string) []string {
 func dispatchAction(ctx context.Context, d interfaces.ResourceDriver, action interfaces.PlanAction, result *interfaces.ApplyResult, hooks ApplyPlanHooks, deleteHookActive bool) (bool, error) {
 	switch action.Action {
 	case "create":
-		return doCreate(ctx, d, action, result)
+		return doCreate(ctx, d, action, result, hooks)
 	case "update":
-		return false, doUpdate(ctx, d, action, result)
+		return false, doUpdate(ctx, d, action, result, hooks)
 	case "replace":
 		err := doReplace(ctx, d, action, result, hooks, deleteHookActive)
 		return err == nil, err
@@ -766,7 +769,7 @@ func dispatchAction(ctx context.Context, d interfaces.ResourceDriver, action int
 //     prefix instead. This boundary is deliberate: ActionError carries
 //     the per-resource action context fields the wrap chain otherwise
 //     duplicates.
-func doCreate(ctx context.Context, d interfaces.ResourceDriver, action interfaces.PlanAction, result *interfaces.ApplyResult) (bool, error) {
+func doCreate(ctx context.Context, d interfaces.ResourceDriver, action interfaces.PlanAction, result *interfaces.ApplyResult, hooks ApplyPlanHooks) (bool, error) {
 	out, err := d.Create(ctx, action.Resource)
 	created := err == nil
 	if errors.Is(err, interfaces.ErrResourceAlreadyExists) {
@@ -783,7 +786,7 @@ func doCreate(ctx context.Context, d interfaces.ResourceDriver, action interface
 			return false, fmt.Errorf("upsert: resource %q found by name but ProviderID is empty: %w", ref.Name, err)
 		}
 		ref.ProviderID = existing.ProviderID
-		out, err = d.Update(ctx, ref, action.Resource)
+		out, err = updateWithCurrentState(ctx, d, ref, action.Resource, hooks)
 	}
 	if err == nil && out != nil {
 		result.Resources = append(result.Resources, *out)
@@ -798,14 +801,11 @@ func doCreate(ctx context.Context, d interfaces.ResourceDriver, action interface
 // (the helper loop body) records them with the canonical action +
 // resource fields.
 //
-// Defensive contract: doUpdate does NOT synthesize a precondition error
-// when action.Current is nil — the driver is the authority on what an
-// empty ProviderID means. ComputePlan upstream is responsible for never
-// emitting an Update without action.Current; if it does, the driver's
-// own typed validation surfaces the bug.
-func doUpdate(ctx context.Context, d interfaces.ResourceDriver, action interfaces.PlanAction, result *interfaces.ApplyResult) error {
+// Legacy drivers remain the authority on an empty ProviderID. State-aware
+// drivers instead require matching persisted identity before dispatch.
+func doUpdate(ctx context.Context, d interfaces.ResourceDriver, action interfaces.PlanAction, result *interfaces.ApplyResult, hooks ApplyPlanHooks) error {
 	ref := refFromAction(action)
-	out, err := d.Update(ctx, ref, action.Resource)
+	out, err := updateWithCurrentState(ctx, d, ref, action.Resource, hooks)
 	if err != nil {
 		return err
 	}
@@ -815,6 +815,26 @@ func doUpdate(ctx context.Context, d interfaces.ResourceDriver, action interface
 	return nil
 }
 
+func updateWithCurrentState(ctx context.Context, d interfaces.ResourceDriver, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, hooks ApplyPlanHooks) (*interfaces.ResourceOutput, error) {
+	updater, ok := d.(interfaces.ResourceStateUpdater)
+	if !ok {
+		return d.Update(ctx, ref, spec)
+	}
+	if hooks.LookupCurrentState == nil {
+		// Existing adapter callers need no new hook. State-dependent providers
+		// must reject legacy Update before mutation when prior state is absent.
+		return d.Update(ctx, ref, spec)
+	}
+	prior, err := hooks.LookupCurrentState(ctx, ref)
+	if err != nil {
+		return nil, fmt.Errorf("update prior state lookup: %w", err)
+	}
+	if err := interfaces.ValidateUpdatePriorState(ref, spec, prior); err != nil {
+		return nil, err
+	}
+	return updater.UpdateWithState(ctx, ref, spec, prior)
+}
+
 // DefaultReplace is the engine's default Replace dispatcher: Delete the
 // resource at action.Current, then Create from action.Resource. Public
 // so drivers that opt into ResourceReplacer can delegate a particular
diff --git a/iac/wfctlhelpers/apply_update_delete_test.go b/iac/wfctlhelpers/apply_update_delete_test.go
index f2b5386c..647d0f41 100644
--- a/iac/wfctlhelpers/apply_update_delete_test.go
+++ b/iac/wfctlhelpers/apply_update_delete_test.go
@@ -2,12 +2,130 @@ package wfctlhelpers
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"reflect"
 	"testing"
 
 	"github.com/GoCodeAlone/workflow/interfaces"
 )
 
+type priorStateUpdateDriver struct {
+	*fakeDriver
+	prior      *interfaces.ResourceState
+	stateCalls int
+	diffCalls  int
+}
+
+func (d *priorStateUpdateDriver) Update(context.Context, interfaces.ResourceRef, interfaces.ResourceSpec) (*interfaces.ResourceOutput, error) {
+	d.updateCount++
+	return nil, errors.New("state-aware update used legacy dispatch")
+}
+
+func (d *priorStateUpdateDriver) UpdateWithState(_ context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
+	d.stateCalls++
+	d.prior = prior
+	if prior == nil {
+		return nil, interfaces.ErrValidation
+	}
+	return &interfaces.ResourceOutput{Name: spec.Name, Type: spec.Type, ProviderID: ref.ProviderID}, nil
+}
+
+func (d *priorStateUpdateDriver) Diff(context.Context, interfaces.ResourceSpec, *interfaces.ResourceOutput) (*interfaces.DiffResult, error) {
+	d.diffCalls++
+	return nil, errors.New("saved-plan update must not call Diff")
+}
+
+type priorStateUpdateProvider struct {
+	*fakeProvider
+	driver *priorStateUpdateDriver
+}
+
+func (p *priorStateUpdateProvider) ResourceDriver(string) (interfaces.ResourceDriver, error) {
+	return p.driver, nil
+}
+
+func TestApplyPlan_UpdateWithState_SavedPlanAfterStateRestart(t *testing.T) {
+	store := &FSStateStore{dir: t.TempDir()}
+	stale := interfaces.ResourceState{ID: "fixture", Name: "fixture", Type: "infra.fixture", ProviderID: "fixture-id", Outputs: map[string]any{"rotation_generation": float64(1)}}
+	if err := store.SaveResource(t.Context(), stale); err != nil {
+		t.Fatal(err)
+	}
+	encoded, err := json.Marshal(interfaces.IaCPlan{Actions: []interfaces.PlanAction{{Action: "update", Resource: spec(stale.Name, stale.Type), Current: &stale}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	latest := stale
+	latest.Outputs = map[string]any{"rotation_generation": float64(2), "credential_ref": "secrets://fixture/credential"}
+	if err := store.SaveResource(t.Context(), latest); err != nil {
+		t.Fatal(err)
+	}
+	restarted := &FSStateStore{dir: store.dir}
+	current, err := restarted.ListResources(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	var saved interfaces.IaCPlan
+	if err := json.Unmarshal(encoded, &saved); err != nil {
+		t.Fatal(err)
+	}
+	driver := &priorStateUpdateDriver{fakeDriver: &fakeDriver{}}
+	provider := &priorStateUpdateProvider{fakeProvider: newFakeProvider(), driver: driver}
+	latest.Outputs["rotation_generation"] = float64(3)
+	result, err := ApplyPlanWithHooks(t.Context(), provider, &saved, ApplyPlanHooks{
+		CurrentState: current,
+		OnBeforeAction: func(context.Context, interfaces.PlanAction) error {
+			return restarted.SaveResource(t.Context(), latest)
+		},
+		LookupCurrentState: func(ctx context.Context, ref interfaces.ResourceRef) (*interfaces.ResourceState, error) {
+			return restarted.GetResource(ctx, ref.Name)
+		},
+	})
+	if err != nil || len(result.Errors) != 0 || driver.stateCalls != 1 || driver.updateCount != 0 || driver.diffCalls != 0 {
+		t.Fatalf("saved-plan state-aware dispatch failed: err=%v errors=%v state_calls=%d legacy_calls=%d diff_calls=%d", err, result.Errors, driver.stateCalls, driver.updateCount, driver.diffCalls)
+	}
+	if driver.prior == nil || !reflect.DeepEqual(driver.prior.Outputs, latest.Outputs) {
+		t.Fatal("update did not receive latest persisted outputs after state-store restart")
+	}
+}
+
+func TestApplyPlan_UpdateWithState_InvalidPriorNeverMutates(t *testing.T) {
+	valid := interfaces.ResourceState{Name: "fixture", Type: "infra.fixture", ProviderID: "fixture-id"}
+	for _, name := range []string{"missing_lookup", "missing_state", "wrong_name", "wrong_type", "wrong_provider_id", "lookup_failure"} {
+		t.Run(name, func(t *testing.T) {
+			prior := valid
+			hooks := ApplyPlanHooks{LookupCurrentState: func(context.Context, interfaces.ResourceRef) (*interfaces.ResourceState, error) { return &prior, nil }}
+			switch name {
+			case "missing_lookup":
+				hooks.LookupCurrentState = nil
+			case "missing_state":
+				hooks.LookupCurrentState = func(context.Context, interfaces.ResourceRef) (*interfaces.ResourceState, error) { return nil, nil }
+			case "wrong_name":
+				prior.Name = "other"
+			case "wrong_type":
+				prior.Type = "infra.other"
+			case "wrong_provider_id":
+				prior.ProviderID = "other-id"
+			case "lookup_failure":
+				hooks.LookupCurrentState = func(context.Context, interfaces.ResourceRef) (*interfaces.ResourceState, error) {
+					return nil, errors.New("fixture lookup failed")
+				}
+			}
+			driver := &priorStateUpdateDriver{fakeDriver: &fakeDriver{}}
+			provider := &priorStateUpdateProvider{fakeProvider: newFakeProvider(), driver: driver}
+			plan := &interfaces.IaCPlan{Actions: []interfaces.PlanAction{{Action: "update", Resource: spec(valid.Name, valid.Type), Current: &valid}}}
+			result, err := ApplyPlanWithHooks(t.Context(), provider, plan, hooks)
+			legacyCalls := 0
+			if name == "missing_lookup" {
+				legacyCalls = 1 // The state-dependent fixture rejects legacy Update.
+			}
+			if err != nil || len(result.Errors) != 1 || driver.stateCalls != 0 || driver.updateCount != legacyCalls || driver.diffCalls != 0 {
+				t.Fatal("invalid prior state reached mutation or Diff")
+			}
+		})
+	}
+}
+
 func TestReplaceCleanup_CloudDeletionResumeUsesDurableHook(t *testing.T) {
 	for _, kind := range []string{"delete", "replace"} {
 		for _, completed := range []bool{false, true} {
diff --git a/interfaces/iac_canonical_types.go b/interfaces/iac_canonical_types.go
index 023c4514..ae809075 100644
--- a/interfaces/iac_canonical_types.go
+++ b/interfaces/iac_canonical_types.go
@@ -5,6 +5,19 @@ import (
 	"strings"
 )
 
+// ValidateUpdatePriorState binds an update to the same resource incarnation.
+// Diagnostics intentionally omit config, outputs, and provider identifiers.
+func ValidateUpdatePriorState(ref ResourceRef, spec ResourceSpec, prior *ResourceState) error {
+	if prior == nil {
+		return fmt.Errorf("%w: update requires persisted prior state", ErrValidation)
+	}
+	if strings.TrimSpace(ref.Name) == "" || strings.TrimSpace(ref.Type) == "" || strings.TrimSpace(ref.ProviderID) == "" ||
+		ref.Name != spec.Name || ref.Type != spec.Type || prior.Name != ref.Name || prior.Type != ref.Type || prior.ProviderID != ref.ProviderID {
+		return fmt.Errorf("%w: update prior state does not match resource identity", ErrValidation)
+	}
+	return nil
+}
+
 const (
 	JobKindPreDeploy    = "PRE_DEPLOY"
 	JobKindPostDeploy   = "POST_DEPLOY"
diff --git a/interfaces/iac_resource_driver.go b/interfaces/iac_resource_driver.go
index debd9b39..a31f172c 100644
--- a/interfaces/iac_resource_driver.go
+++ b/interfaces/iac_resource_driver.go
@@ -90,6 +90,17 @@ type ResourceDriver interface {
 	SensitiveKeys() []string
 }
 
+// ResourceStateUpdater optionally receives authoritative persisted state on the
+// update call. Callers must load it at dispatch time, not from a saved plan or
+// an earlier Diff. Drivers requiring it must reject legacy Update calls before
+// mutation; ResourceDriver remains unchanged for existing implementations.
+// A typed provider bridge may receive nil prior state from a legacy RPC. It
+// must reject state-dependent updates before mutation, but may delegate updates
+// that do not require state to their existing implementation.
+type ResourceStateUpdater interface {
+	UpdateWithState(ctx context.Context, ref ResourceRef, spec ResourceSpec, prior *ResourceState) (*ResourceOutput, error)
+}
+
 // ResourceAdoptionLocator is an optional interface ResourceDriver
 // implementations may provide when a desired ResourceSpec can be resolved to a
 // live provider resource before local state exists. wfctl infra apply uses this
diff --git a/plugin/external/proto/iac.pb.go b/plugin/external/proto/iac.pb.go
index d9d97cd4..6e8c0ad2 100644
--- a/plugin/external/proto/iac.pb.go
+++ b/plugin/external/proto/iac.pb.go
@@ -6088,10 +6088,13 @@ func (x *ResourceReadResponse) GetOutput() *ResourceOutput {
 }
 
 type ResourceUpdateRequest struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	ResourceType  string                 `protobuf:"bytes,1,opt,name=resource_type,json=resourceType,proto3" json:"resource_type,omitempty"`
-	Ref           *ResourceRef           `protobuf:"bytes,2,opt,name=ref,proto3" json:"ref,omitempty"`
-	Spec          *ResourceSpec          `protobuf:"bytes,3,opt,name=spec,proto3" json:"spec,omitempty"`
+	state        protoimpl.MessageState `protogen:"open.v1"`
+	ResourceType string                 `protobuf:"bytes,1,opt,name=resource_type,json=resourceType,proto3" json:"resource_type,omitempty"`
+	Ref          *ResourceRef           `protobuf:"bytes,2,opt,name=ref,proto3" json:"ref,omitempty"`
+	Spec         *ResourceSpec          `protobuf:"bytes,3,opt,name=spec,proto3" json:"spec,omitempty"`
+	// Authoritative persisted state loaded at update dispatch, never a saved
+	// plan's Current or a prior Diff cache. Absent for legacy Update callers.
+	PriorState    *ResourceState `protobuf:"bytes,4,opt,name=prior_state,json=priorState,proto3" json:"prior_state,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -6147,6 +6150,13 @@ func (x *ResourceUpdateRequest) GetSpec() *ResourceSpec {
 	return nil
 }
 
+func (x *ResourceUpdateRequest) GetPriorState() *ResourceState {
+	if x != nil {
+		return x.PriorState
+	}
+	return nil
+}
+
 type ResourceUpdateResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Output        *ResourceOutput        `protobuf:"bytes,1,opt,name=output,proto3" json:"output,omitempty"`
@@ -8107,11 +8117,13 @@ const file_iac_proto_rawDesc = "" +
 	"\rresource_type\x18\x01 \x01(\tR\fresourceType\x12;\n" +
 	"\x03ref\x18\x02 \x01(\v2).workflow.plugin.external.iac.ResourceRefR\x03ref\"\\\n" +
 	"\x14ResourceReadResponse\x12D\n" +
-	"\x06output\x18\x01 \x01(\v2,.workflow.plugin.external.iac.ResourceOutputR\x06output\"\xb9\x01\n" +
+	"\x06output\x18\x01 \x01(\v2,.workflow.plugin.external.iac.ResourceOutputR\x06output\"\x87\x02\n" +
 	"\x15ResourceUpdateRequest\x12#\n" +
 	"\rresource_type\x18\x01 \x01(\tR\fresourceType\x12;\n" +
 	"\x03ref\x18\x02 \x01(\v2).workflow.plugin.external.iac.ResourceRefR\x03ref\x12>\n" +
-	"\x04spec\x18\x03 \x01(\v2*.workflow.plugin.external.iac.ResourceSpecR\x04spec\"^\n" +
+	"\x04spec\x18\x03 \x01(\v2*.workflow.plugin.external.iac.ResourceSpecR\x04spec\x12L\n" +
+	"\vprior_state\x18\x04 \x01(\v2+.workflow.plugin.external.iac.ResourceStateR\n" +
+	"priorState\"^\n" +
 	"\x16ResourceUpdateResponse\x12D\n" +
 	"\x06output\x18\x01 \x01(\v2,.workflow.plugin.external.iac.ResourceOutputR\x06output\"y\n" +
 	"\x15ResourceDeleteRequest\x12#\n" +
@@ -8588,124 +8600,125 @@ var file_iac_proto_depIdxs = []int32{
 	16,  // 78: workflow.plugin.external.iac.ResourceReadResponse.output:type_name -> workflow.plugin.external.iac.ResourceOutput
 	11,  // 79: workflow.plugin.external.iac.ResourceUpdateRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
 	10,  // 80: workflow.plugin.external.iac.ResourceUpdateRequest.spec:type_name -> workflow.plugin.external.iac.ResourceSpec
-	16,  // 81: workflow.plugin.external.iac.ResourceUpdateResponse.output:type_name -> workflow.plugin.external.iac.ResourceOutput
-	11,  // 82: workflow.plugin.external.iac.ResourceDeleteRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
-	10,  // 83: workflow.plugin.external.iac.ResourceDiffRequest.desired:type_name -> workflow.plugin.external.iac.ResourceSpec
-	16,  // 84: workflow.plugin.external.iac.ResourceDiffRequest.current:type_name -> workflow.plugin.external.iac.ResourceOutput
-	19,  // 85: workflow.plugin.external.iac.ResourceDiffResponse.result:type_name -> workflow.plugin.external.iac.DiffResult
-	11,  // 86: workflow.plugin.external.iac.ResourceScaleRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
-	16,  // 87: workflow.plugin.external.iac.ResourceScaleResponse.output:type_name -> workflow.plugin.external.iac.ResourceOutput
-	11,  // 88: workflow.plugin.external.iac.ResourceHealthCheckRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
-	32,  // 89: workflow.plugin.external.iac.ResourceHealthCheckResponse.result:type_name -> workflow.plugin.external.iac.HealthResult
-	11,  // 90: workflow.plugin.external.iac.TroubleshootRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
-	33,  // 91: workflow.plugin.external.iac.TroubleshootResponse.diagnostics:type_name -> workflow.plugin.external.iac.Diagnostic
-	116, // 92: workflow.plugin.external.iac.GetStateResponse.state:type_name -> workflow.plugin.external.iac.IaCState
-	116, // 93: workflow.plugin.external.iac.SaveStateRequest.state:type_name -> workflow.plugin.external.iac.IaCState
-	145, // 94: workflow.plugin.external.iac.ListStatesRequest.filter:type_name -> workflow.plugin.external.iac.ListStatesRequest.FilterEntry
-	116, // 95: workflow.plugin.external.iac.ListStatesResponse.states:type_name -> workflow.plugin.external.iac.IaCState
-	10,  // 96: workflow.plugin.external.iac.DetectDriftWithSpecsRequest.SpecsEntry.value:type_name -> workflow.plugin.external.iac.ResourceSpec
-	10,  // 97: workflow.plugin.external.iac.DetectDriftConfigRequest.SpecsEntry.value:type_name -> workflow.plugin.external.iac.ResourceSpec
-	42,  // 98: workflow.plugin.external.iac.IaCProviderRequired.Initialize:input_type -> workflow.plugin.external.iac.InitializeRequest
-	44,  // 99: workflow.plugin.external.iac.IaCProviderRequired.Name:input_type -> workflow.plugin.external.iac.NameRequest
-	46,  // 100: workflow.plugin.external.iac.IaCProviderRequired.Version:input_type -> workflow.plugin.external.iac.VersionRequest
-	48,  // 101: workflow.plugin.external.iac.IaCProviderRequired.Capabilities:input_type -> workflow.plugin.external.iac.CapabilitiesRequest
-	50,  // 102: workflow.plugin.external.iac.IaCProviderRequired.Plan:input_type -> workflow.plugin.external.iac.PlanRequest
-	52,  // 103: workflow.plugin.external.iac.IaCProviderRequired.Destroy:input_type -> workflow.plugin.external.iac.DestroyRequest
-	54,  // 104: workflow.plugin.external.iac.IaCProviderRequired.Status:input_type -> workflow.plugin.external.iac.StatusRequest
-	56,  // 105: workflow.plugin.external.iac.IaCProviderRequired.Import:input_type -> workflow.plugin.external.iac.ImportRequest
-	58,  // 106: workflow.plugin.external.iac.IaCProviderRequired.ResolveSizing:input_type -> workflow.plugin.external.iac.ResolveSizingRequest
-	60,  // 107: workflow.plugin.external.iac.IaCProviderRequired.BootstrapStateBackend:input_type -> workflow.plugin.external.iac.BootstrapStateBackendRequest
-	62,  // 108: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateAll:input_type -> workflow.plugin.external.iac.EnumerateAllRequest
-	64,  // 109: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateByTag:input_type -> workflow.plugin.external.iac.EnumerateByTagRequest
-	66,  // 110: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDrift:input_type -> workflow.plugin.external.iac.DetectDriftRequest
-	68,  // 111: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDriftWithSpecs:input_type -> workflow.plugin.external.iac.DetectDriftWithSpecsRequest
-	70,  // 112: workflow.plugin.external.iac.IaCProviderCredentialRevoker.RevokeProviderCredential:input_type -> workflow.plugin.external.iac.RevokeProviderCredentialRequest
-	72,  // 113: workflow.plugin.external.iac.IaCProviderRegionLister.ListRegions:input_type -> workflow.plugin.external.iac.ListRegionsRequest
-	75,  // 114: workflow.plugin.external.iac.IaCProviderOwnership.GetOwner:input_type -> workflow.plugin.external.iac.GetOwnerRequest
-	77,  // 115: workflow.plugin.external.iac.IaCProviderOwnership.SetOwner:input_type -> workflow.plugin.external.iac.SetOwnerRequest
-	79,  // 116: workflow.plugin.external.iac.IaCProviderOwnership.ListOwners:input_type -> workflow.plugin.external.iac.ListOwnersRequest
-	82,  // 117: workflow.plugin.external.iac.IaCProviderFinalizer.FinalizeApply:input_type -> workflow.plugin.external.iac.FinalizeApplyRequest
-	84,  // 118: workflow.plugin.external.iac.IaCProviderMigrationRepairer.RepairDirtyMigration:input_type -> workflow.plugin.external.iac.RepairDirtyMigrationRequest
-	86,  // 119: workflow.plugin.external.iac.IaCProviderValidator.ValidatePlan:input_type -> workflow.plugin.external.iac.ValidatePlanRequest
-	88,  // 120: workflow.plugin.external.iac.IaCProviderDriftConfigDetector.DetectDriftConfig:input_type -> workflow.plugin.external.iac.DetectDriftConfigRequest
-	90,  // 121: workflow.plugin.external.iac.IaCProviderLogCapture.CaptureLogs:input_type -> workflow.plugin.external.iac.CaptureLogsRequest
-	92,  // 122: workflow.plugin.external.iac.IaCProviderRunner.RunJob:input_type -> workflow.plugin.external.iac.JobSpec
-	96,  // 123: workflow.plugin.external.iac.IaCProviderRunner.JobStatus:input_type -> workflow.plugin.external.iac.JobHandle
-	96,  // 124: workflow.plugin.external.iac.IaCProviderRunner.JobLogs:input_type -> workflow.plugin.external.iac.JobHandle
-	96,  // 125: workflow.plugin.external.iac.IaCProviderJobCanceler.CancelJob:input_type -> workflow.plugin.external.iac.JobHandle
-	21,  // 126: workflow.plugin.external.iac.IaCRequirementDiscovery.DiscoverRequirements:input_type -> workflow.plugin.external.iac.DiscoverRequirementsRequest
-	25,  // 127: workflow.plugin.external.iac.IaCProviderRequirementMapper.MapRequirements:input_type -> workflow.plugin.external.iac.MapRequirementsRequest
-	98,  // 128: workflow.plugin.external.iac.ResourceDriver.Create:input_type -> workflow.plugin.external.iac.ResourceCreateRequest
-	100, // 129: workflow.plugin.external.iac.ResourceDriver.Read:input_type -> workflow.plugin.external.iac.ResourceReadRequest
-	102, // 130: workflow.plugin.external.iac.ResourceDriver.Update:input_type -> workflow.plugin.external.iac.ResourceUpdateRequest
-	104, // 131: workflow.plugin.external.iac.ResourceDriver.Delete:input_type -> workflow.plugin.external.iac.ResourceDeleteRequest
-	106, // 132: workflow.plugin.external.iac.ResourceDriver.Diff:input_type -> workflow.plugin.external.iac.ResourceDiffRequest
-	108, // 133: workflow.plugin.external.iac.ResourceDriver.Scale:input_type -> workflow.plugin.external.iac.ResourceScaleRequest
-	110, // 134: workflow.plugin.external.iac.ResourceDriver.HealthCheck:input_type -> workflow.plugin.external.iac.ResourceHealthCheckRequest
-	112, // 135: workflow.plugin.external.iac.ResourceDriver.SensitiveKeys:input_type -> workflow.plugin.external.iac.SensitiveKeysRequest
-	114, // 136: workflow.plugin.external.iac.ResourceDriver.Troubleshoot:input_type -> workflow.plugin.external.iac.TroubleshootRequest
-	117, // 137: workflow.plugin.external.iac.IaCStateBackend.Configure:input_type -> workflow.plugin.external.iac.ConfigureRequest
-	119, // 138: workflow.plugin.external.iac.IaCStateBackend.GetState:input_type -> workflow.plugin.external.iac.GetStateRequest
-	121, // 139: workflow.plugin.external.iac.IaCStateBackend.SaveState:input_type -> workflow.plugin.external.iac.SaveStateRequest
-	123, // 140: workflow.plugin.external.iac.IaCStateBackend.ListStates:input_type -> workflow.plugin.external.iac.ListStatesRequest
-	125, // 141: workflow.plugin.external.iac.IaCStateBackend.DeleteState:input_type -> workflow.plugin.external.iac.DeleteStateRequest
-	127, // 142: workflow.plugin.external.iac.IaCStateBackend.Lock:input_type -> workflow.plugin.external.iac.LockRequest
-	129, // 143: workflow.plugin.external.iac.IaCStateBackend.Unlock:input_type -> workflow.plugin.external.iac.UnlockRequest
-	131, // 144: workflow.plugin.external.iac.IaCStateBackend.ListBackendNames:input_type -> workflow.plugin.external.iac.ListBackendNamesRequest
-	133, // 145: workflow.plugin.external.iac.ResourceSensitiveInputDeclarer.SensitiveInputPaths:input_type -> workflow.plugin.external.iac.ResourceSensitiveInputPathsRequest
-	43,  // 146: workflow.plugin.external.iac.IaCProviderRequired.Initialize:output_type -> workflow.plugin.external.iac.InitializeResponse
-	45,  // 147: workflow.plugin.external.iac.IaCProviderRequired.Name:output_type -> workflow.plugin.external.iac.NameResponse
-	47,  // 148: workflow.plugin.external.iac.IaCProviderRequired.Version:output_type -> workflow.plugin.external.iac.VersionResponse
-	49,  // 149: workflow.plugin.external.iac.IaCProviderRequired.Capabilities:output_type -> workflow.plugin.external.iac.CapabilitiesResponse
-	51,  // 150: workflow.plugin.external.iac.IaCProviderRequired.Plan:output_type -> workflow.plugin.external.iac.PlanResponse
-	53,  // 151: workflow.plugin.external.iac.IaCProviderRequired.Destroy:output_type -> workflow.plugin.external.iac.DestroyResponse
-	55,  // 152: workflow.plugin.external.iac.IaCProviderRequired.Status:output_type -> workflow.plugin.external.iac.StatusResponse
-	57,  // 153: workflow.plugin.external.iac.IaCProviderRequired.Import:output_type -> workflow.plugin.external.iac.ImportResponse
-	59,  // 154: workflow.plugin.external.iac.IaCProviderRequired.ResolveSizing:output_type -> workflow.plugin.external.iac.ResolveSizingResponse
-	61,  // 155: workflow.plugin.external.iac.IaCProviderRequired.BootstrapStateBackend:output_type -> workflow.plugin.external.iac.BootstrapStateBackendResponse
-	63,  // 156: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateAll:output_type -> workflow.plugin.external.iac.EnumerateAllResponse
-	65,  // 157: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateByTag:output_type -> workflow.plugin.external.iac.EnumerateByTagResponse
-	67,  // 158: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDrift:output_type -> workflow.plugin.external.iac.DetectDriftResponse
-	69,  // 159: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDriftWithSpecs:output_type -> workflow.plugin.external.iac.DetectDriftWithSpecsResponse
-	71,  // 160: workflow.plugin.external.iac.IaCProviderCredentialRevoker.RevokeProviderCredential:output_type -> workflow.plugin.external.iac.RevokeProviderCredentialResponse
-	74,  // 161: workflow.plugin.external.iac.IaCProviderRegionLister.ListRegions:output_type -> workflow.plugin.external.iac.ListRegionsResponse
-	76,  // 162: workflow.plugin.external.iac.IaCProviderOwnership.GetOwner:output_type -> workflow.plugin.external.iac.GetOwnerResponse
-	78,  // 163: workflow.plugin.external.iac.IaCProviderOwnership.SetOwner:output_type -> workflow.plugin.external.iac.SetOwnerResponse
-	81,  // 164: workflow.plugin.external.iac.IaCProviderOwnership.ListOwners:output_type -> workflow.plugin.external.iac.ListOwnersResponse
-	83,  // 165: workflow.plugin.external.iac.IaCProviderFinalizer.FinalizeApply:output_type -> workflow.plugin.external.iac.FinalizeApplyResponse
-	85,  // 166: workflow.plugin.external.iac.IaCProviderMigrationRepairer.RepairDirtyMigration:output_type -> workflow.plugin.external.iac.RepairDirtyMigrationResponse
-	87,  // 167: workflow.plugin.external.iac.IaCProviderValidator.ValidatePlan:output_type -> workflow.plugin.external.iac.ValidatePlanResponse
-	89,  // 168: workflow.plugin.external.iac.IaCProviderDriftConfigDetector.DetectDriftConfig:output_type -> workflow.plugin.external.iac.DetectDriftConfigResponse
-	91,  // 169: workflow.plugin.external.iac.IaCProviderLogCapture.CaptureLogs:output_type -> workflow.plugin.external.iac.LogChunk
-	96,  // 170: workflow.plugin.external.iac.IaCProviderRunner.RunJob:output_type -> workflow.plugin.external.iac.JobHandle
-	97,  // 171: workflow.plugin.external.iac.IaCProviderRunner.JobStatus:output_type -> workflow.plugin.external.iac.JobStatusReply
-	91,  // 172: workflow.plugin.external.iac.IaCProviderRunner.JobLogs:output_type -> workflow.plugin.external.iac.LogChunk
-	147, // 173: workflow.plugin.external.iac.IaCProviderJobCanceler.CancelJob:output_type -> google.protobuf.Empty
-	24,  // 174: workflow.plugin.external.iac.IaCRequirementDiscovery.DiscoverRequirements:output_type -> workflow.plugin.external.iac.DiscoverRequirementsResponse
-	29,  // 175: workflow.plugin.external.iac.IaCProviderRequirementMapper.MapRequirements:output_type -> workflow.plugin.external.iac.MapRequirementsResponse
-	99,  // 176: workflow.plugin.external.iac.ResourceDriver.Create:output_type -> workflow.plugin.external.iac.ResourceCreateResponse
-	101, // 177: workflow.plugin.external.iac.ResourceDriver.Read:output_type -> workflow.plugin.external.iac.ResourceReadResponse
-	103, // 178: workflow.plugin.external.iac.ResourceDriver.Update:output_type -> workflow.plugin.external.iac.ResourceUpdateResponse
-	105, // 179: workflow.plugin.external.iac.ResourceDriver.Delete:output_type -> workflow.plugin.external.iac.ResourceDeleteResponse
-	107, // 180: workflow.plugin.external.iac.ResourceDriver.Diff:output_type -> workflow.plugin.external.iac.ResourceDiffResponse
-	109, // 181: workflow.plugin.external.iac.ResourceDriver.Scale:output_type -> workflow.plugin.external.iac.ResourceScaleResponse
-	111, // 182: workflow.plugin.external.iac.ResourceDriver.HealthCheck:output_type -> workflow.plugin.external.iac.ResourceHealthCheckResponse
-	113, // 183: workflow.plugin.external.iac.ResourceDriver.SensitiveKeys:output_type -> workflow.plugin.external.iac.SensitiveKeysResponse
-	115, // 184: workflow.plugin.external.iac.ResourceDriver.Troubleshoot:output_type -> workflow.plugin.external.iac.TroubleshootResponse
-	118, // 185: workflow.plugin.external.iac.IaCStateBackend.Configure:output_type -> workflow.plugin.external.iac.ConfigureResponse
-	120, // 186: workflow.plugin.external.iac.IaCStateBackend.GetState:output_type -> workflow.plugin.external.iac.GetStateResponse
-	122, // 187: workflow.plugin.external.iac.IaCStateBackend.SaveState:output_type -> workflow.plugin.external.iac.SaveStateResponse
-	124, // 188: workflow.plugin.external.iac.IaCStateBackend.ListStates:output_type -> workflow.plugin.external.iac.ListStatesResponse
-	126, // 189: workflow.plugin.external.iac.IaCStateBackend.DeleteState:output_type -> workflow.plugin.external.iac.DeleteStateResponse
-	128, // 190: workflow.plugin.external.iac.IaCStateBackend.Lock:output_type -> workflow.plugin.external.iac.LockResponse
-	130, // 191: workflow.plugin.external.iac.IaCStateBackend.Unlock:output_type -> workflow.plugin.external.iac.UnlockResponse
-	132, // 192: workflow.plugin.external.iac.IaCStateBackend.ListBackendNames:output_type -> workflow.plugin.external.iac.ListBackendNamesResponse
-	134, // 193: workflow.plugin.external.iac.ResourceSensitiveInputDeclarer.SensitiveInputPaths:output_type -> workflow.plugin.external.iac.ResourceSensitiveInputPathsResponse
-	146, // [146:194] is the sub-list for method output_type
-	98,  // [98:146] is the sub-list for method input_type
-	98,  // [98:98] is the sub-list for extension type_name
-	98,  // [98:98] is the sub-list for extension extendee
-	0,   // [0:98] is the sub-list for field type_name
+	15,  // 81: workflow.plugin.external.iac.ResourceUpdateRequest.prior_state:type_name -> workflow.plugin.external.iac.ResourceState
+	16,  // 82: workflow.plugin.external.iac.ResourceUpdateResponse.output:type_name -> workflow.plugin.external.iac.ResourceOutput
+	11,  // 83: workflow.plugin.external.iac.ResourceDeleteRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
+	10,  // 84: workflow.plugin.external.iac.ResourceDiffRequest.desired:type_name -> workflow.plugin.external.iac.ResourceSpec
+	16,  // 85: workflow.plugin.external.iac.ResourceDiffRequest.current:type_name -> workflow.plugin.external.iac.ResourceOutput
+	19,  // 86: workflow.plugin.external.iac.ResourceDiffResponse.result:type_name -> workflow.plugin.external.iac.DiffResult
+	11,  // 87: workflow.plugin.external.iac.ResourceScaleRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
+	16,  // 88: workflow.plugin.external.iac.ResourceScaleResponse.output:type_name -> workflow.plugin.external.iac.ResourceOutput
+	11,  // 89: workflow.plugin.external.iac.ResourceHealthCheckRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
+	32,  // 90: workflow.plugin.external.iac.ResourceHealthCheckResponse.result:type_name -> workflow.plugin.external.iac.HealthResult
+	11,  // 91: workflow.plugin.external.iac.TroubleshootRequest.ref:type_name -> workflow.plugin.external.iac.ResourceRef
+	33,  // 92: workflow.plugin.external.iac.TroubleshootResponse.diagnostics:type_name -> workflow.plugin.external.iac.Diagnostic
+	116, // 93: workflow.plugin.external.iac.GetStateResponse.state:type_name -> workflow.plugin.external.iac.IaCState
+	116, // 94: workflow.plugin.external.iac.SaveStateRequest.state:type_name -> workflow.plugin.external.iac.IaCState
+	145, // 95: workflow.plugin.external.iac.ListStatesRequest.filter:type_name -> workflow.plugin.external.iac.ListStatesRequest.FilterEntry
+	116, // 96: workflow.plugin.external.iac.ListStatesResponse.states:type_name -> workflow.plugin.external.iac.IaCState
+	10,  // 97: workflow.plugin.external.iac.DetectDriftWithSpecsRequest.SpecsEntry.value:type_name -> workflow.plugin.external.iac.ResourceSpec
+	10,  // 98: workflow.plugin.external.iac.DetectDriftConfigRequest.SpecsEntry.value:type_name -> workflow.plugin.external.iac.ResourceSpec
+	42,  // 99: workflow.plugin.external.iac.IaCProviderRequired.Initialize:input_type -> workflow.plugin.external.iac.InitializeRequest
+	44,  // 100: workflow.plugin.external.iac.IaCProviderRequired.Name:input_type -> workflow.plugin.external.iac.NameRequest
+	46,  // 101: workflow.plugin.external.iac.IaCProviderRequired.Version:input_type -> workflow.plugin.external.iac.VersionRequest
+	48,  // 102: workflow.plugin.external.iac.IaCProviderRequired.Capabilities:input_type -> workflow.plugin.external.iac.CapabilitiesRequest
+	50,  // 103: workflow.plugin.external.iac.IaCProviderRequired.Plan:input_type -> workflow.plugin.external.iac.PlanRequest
+	52,  // 104: workflow.plugin.external.iac.IaCProviderRequired.Destroy:input_type -> workflow.plugin.external.iac.DestroyRequest
+	54,  // 105: workflow.plugin.external.iac.IaCProviderRequired.Status:input_type -> workflow.plugin.external.iac.StatusRequest
+	56,  // 106: workflow.plugin.external.iac.IaCProviderRequired.Import:input_type -> workflow.plugin.external.iac.ImportRequest
+	58,  // 107: workflow.plugin.external.iac.IaCProviderRequired.ResolveSizing:input_type -> workflow.plugin.external.iac.ResolveSizingRequest
+	60,  // 108: workflow.plugin.external.iac.IaCProviderRequired.BootstrapStateBackend:input_type -> workflow.plugin.external.iac.BootstrapStateBackendRequest
+	62,  // 109: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateAll:input_type -> workflow.plugin.external.iac.EnumerateAllRequest
+	64,  // 110: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateByTag:input_type -> workflow.plugin.external.iac.EnumerateByTagRequest
+	66,  // 111: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDrift:input_type -> workflow.plugin.external.iac.DetectDriftRequest
+	68,  // 112: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDriftWithSpecs:input_type -> workflow.plugin.external.iac.DetectDriftWithSpecsRequest
+	70,  // 113: workflow.plugin.external.iac.IaCProviderCredentialRevoker.RevokeProviderCredential:input_type -> workflow.plugin.external.iac.RevokeProviderCredentialRequest
+	72,  // 114: workflow.plugin.external.iac.IaCProviderRegionLister.ListRegions:input_type -> workflow.plugin.external.iac.ListRegionsRequest
+	75,  // 115: workflow.plugin.external.iac.IaCProviderOwnership.GetOwner:input_type -> workflow.plugin.external.iac.GetOwnerRequest
+	77,  // 116: workflow.plugin.external.iac.IaCProviderOwnership.SetOwner:input_type -> workflow.plugin.external.iac.SetOwnerRequest
+	79,  // 117: workflow.plugin.external.iac.IaCProviderOwnership.ListOwners:input_type -> workflow.plugin.external.iac.ListOwnersRequest
+	82,  // 118: workflow.plugin.external.iac.IaCProviderFinalizer.FinalizeApply:input_type -> workflow.plugin.external.iac.FinalizeApplyRequest
+	84,  // 119: workflow.plugin.external.iac.IaCProviderMigrationRepairer.RepairDirtyMigration:input_type -> workflow.plugin.external.iac.RepairDirtyMigrationRequest
+	86,  // 120: workflow.plugin.external.iac.IaCProviderValidator.ValidatePlan:input_type -> workflow.plugin.external.iac.ValidatePlanRequest
+	88,  // 121: workflow.plugin.external.iac.IaCProviderDriftConfigDetector.DetectDriftConfig:input_type -> workflow.plugin.external.iac.DetectDriftConfigRequest
+	90,  // 122: workflow.plugin.external.iac.IaCProviderLogCapture.CaptureLogs:input_type -> workflow.plugin.external.iac.CaptureLogsRequest
+	92,  // 123: workflow.plugin.external.iac.IaCProviderRunner.RunJob:input_type -> workflow.plugin.external.iac.JobSpec
+	96,  // 124: workflow.plugin.external.iac.IaCProviderRunner.JobStatus:input_type -> workflow.plugin.external.iac.JobHandle
+	96,  // 125: workflow.plugin.external.iac.IaCProviderRunner.JobLogs:input_type -> workflow.plugin.external.iac.JobHandle
+	96,  // 126: workflow.plugin.external.iac.IaCProviderJobCanceler.CancelJob:input_type -> workflow.plugin.external.iac.JobHandle
+	21,  // 127: workflow.plugin.external.iac.IaCRequirementDiscovery.DiscoverRequirements:input_type -> workflow.plugin.external.iac.DiscoverRequirementsRequest
+	25,  // 128: workflow.plugin.external.iac.IaCProviderRequirementMapper.MapRequirements:input_type -> workflow.plugin.external.iac.MapRequirementsRequest
+	98,  // 129: workflow.plugin.external.iac.ResourceDriver.Create:input_type -> workflow.plugin.external.iac.ResourceCreateRequest
+	100, // 130: workflow.plugin.external.iac.ResourceDriver.Read:input_type -> workflow.plugin.external.iac.ResourceReadRequest
+	102, // 131: workflow.plugin.external.iac.ResourceDriver.Update:input_type -> workflow.plugin.external.iac.ResourceUpdateRequest
+	104, // 132: workflow.plugin.external.iac.ResourceDriver.Delete:input_type -> workflow.plugin.external.iac.ResourceDeleteRequest
+	106, // 133: workflow.plugin.external.iac.ResourceDriver.Diff:input_type -> workflow.plugin.external.iac.ResourceDiffRequest
+	108, // 134: workflow.plugin.external.iac.ResourceDriver.Scale:input_type -> workflow.plugin.external.iac.ResourceScaleRequest
+	110, // 135: workflow.plugin.external.iac.ResourceDriver.HealthCheck:input_type -> workflow.plugin.external.iac.ResourceHealthCheckRequest
+	112, // 136: workflow.plugin.external.iac.ResourceDriver.SensitiveKeys:input_type -> workflow.plugin.external.iac.SensitiveKeysRequest
+	114, // 137: workflow.plugin.external.iac.ResourceDriver.Troubleshoot:input_type -> workflow.plugin.external.iac.TroubleshootRequest
+	117, // 138: workflow.plugin.external.iac.IaCStateBackend.Configure:input_type -> workflow.plugin.external.iac.ConfigureRequest
+	119, // 139: workflow.plugin.external.iac.IaCStateBackend.GetState:input_type -> workflow.plugin.external.iac.GetStateRequest
+	121, // 140: workflow.plugin.external.iac.IaCStateBackend.SaveState:input_type -> workflow.plugin.external.iac.SaveStateRequest
+	123, // 141: workflow.plugin.external.iac.IaCStateBackend.ListStates:input_type -> workflow.plugin.external.iac.ListStatesRequest
+	125, // 142: workflow.plugin.external.iac.IaCStateBackend.DeleteState:input_type -> workflow.plugin.external.iac.DeleteStateRequest
+	127, // 143: workflow.plugin.external.iac.IaCStateBackend.Lock:input_type -> workflow.plugin.external.iac.LockRequest
+	129, // 144: workflow.plugin.external.iac.IaCStateBackend.Unlock:input_type -> workflow.plugin.external.iac.UnlockRequest
+	131, // 145: workflow.plugin.external.iac.IaCStateBackend.ListBackendNames:input_type -> workflow.plugin.external.iac.ListBackendNamesRequest
+	133, // 146: workflow.plugin.external.iac.ResourceSensitiveInputDeclarer.SensitiveInputPaths:input_type -> workflow.plugin.external.iac.ResourceSensitiveInputPathsRequest
+	43,  // 147: workflow.plugin.external.iac.IaCProviderRequired.Initialize:output_type -> workflow.plugin.external.iac.InitializeResponse
+	45,  // 148: workflow.plugin.external.iac.IaCProviderRequired.Name:output_type -> workflow.plugin.external.iac.NameResponse
+	47,  // 149: workflow.plugin.external.iac.IaCProviderRequired.Version:output_type -> workflow.plugin.external.iac.VersionResponse
+	49,  // 150: workflow.plugin.external.iac.IaCProviderRequired.Capabilities:output_type -> workflow.plugin.external.iac.CapabilitiesResponse
+	51,  // 151: workflow.plugin.external.iac.IaCProviderRequired.Plan:output_type -> workflow.plugin.external.iac.PlanResponse
+	53,  // 152: workflow.plugin.external.iac.IaCProviderRequired.Destroy:output_type -> workflow.plugin.external.iac.DestroyResponse
+	55,  // 153: workflow.plugin.external.iac.IaCProviderRequired.Status:output_type -> workflow.plugin.external.iac.StatusResponse
+	57,  // 154: workflow.plugin.external.iac.IaCProviderRequired.Import:output_type -> workflow.plugin.external.iac.ImportResponse
+	59,  // 155: workflow.plugin.external.iac.IaCProviderRequired.ResolveSizing:output_type -> workflow.plugin.external.iac.ResolveSizingResponse
+	61,  // 156: workflow.plugin.external.iac.IaCProviderRequired.BootstrapStateBackend:output_type -> workflow.plugin.external.iac.BootstrapStateBackendResponse
+	63,  // 157: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateAll:output_type -> workflow.plugin.external.iac.EnumerateAllResponse
+	65,  // 158: workflow.plugin.external.iac.IaCProviderEnumerator.EnumerateByTag:output_type -> workflow.plugin.external.iac.EnumerateByTagResponse
+	67,  // 159: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDrift:output_type -> workflow.plugin.external.iac.DetectDriftResponse
+	69,  // 160: workflow.plugin.external.iac.IaCProviderDriftDetector.DetectDriftWithSpecs:output_type -> workflow.plugin.external.iac.DetectDriftWithSpecsResponse
+	71,  // 161: workflow.plugin.external.iac.IaCProviderCredentialRevoker.RevokeProviderCredential:output_type -> workflow.plugin.external.iac.RevokeProviderCredentialResponse
+	74,  // 162: workflow.plugin.external.iac.IaCProviderRegionLister.ListRegions:output_type -> workflow.plugin.external.iac.ListRegionsResponse
+	76,  // 163: workflow.plugin.external.iac.IaCProviderOwnership.GetOwner:output_type -> workflow.plugin.external.iac.GetOwnerResponse
+	78,  // 164: workflow.plugin.external.iac.IaCProviderOwnership.SetOwner:output_type -> workflow.plugin.external.iac.SetOwnerResponse
+	81,  // 165: workflow.plugin.external.iac.IaCProviderOwnership.ListOwners:output_type -> workflow.plugin.external.iac.ListOwnersResponse
+	83,  // 166: workflow.plugin.external.iac.IaCProviderFinalizer.FinalizeApply:output_type -> workflow.plugin.external.iac.FinalizeApplyResponse
+	85,  // 167: workflow.plugin.external.iac.IaCProviderMigrationRepairer.RepairDirtyMigration:output_type -> workflow.plugin.external.iac.RepairDirtyMigrationResponse
+	87,  // 168: workflow.plugin.external.iac.IaCProviderValidator.ValidatePlan:output_type -> workflow.plugin.external.iac.ValidatePlanResponse
+	89,  // 169: workflow.plugin.external.iac.IaCProviderDriftConfigDetector.DetectDriftConfig:output_type -> workflow.plugin.external.iac.DetectDriftConfigResponse
+	91,  // 170: workflow.plugin.external.iac.IaCProviderLogCapture.CaptureLogs:output_type -> workflow.plugin.external.iac.LogChunk
+	96,  // 171: workflow.plugin.external.iac.IaCProviderRunner.RunJob:output_type -> workflow.plugin.external.iac.JobHandle
+	97,  // 172: workflow.plugin.external.iac.IaCProviderRunner.JobStatus:output_type -> workflow.plugin.external.iac.JobStatusReply
+	91,  // 173: workflow.plugin.external.iac.IaCProviderRunner.JobLogs:output_type -> workflow.plugin.external.iac.LogChunk
+	147, // 174: workflow.plugin.external.iac.IaCProviderJobCanceler.CancelJob:output_type -> google.protobuf.Empty
+	24,  // 175: workflow.plugin.external.iac.IaCRequirementDiscovery.DiscoverRequirements:output_type -> workflow.plugin.external.iac.DiscoverRequirementsResponse
+	29,  // 176: workflow.plugin.external.iac.IaCProviderRequirementMapper.MapRequirements:output_type -> workflow.plugin.external.iac.MapRequirementsResponse
+	99,  // 177: workflow.plugin.external.iac.ResourceDriver.Create:output_type -> workflow.plugin.external.iac.ResourceCreateResponse
+	101, // 178: workflow.plugin.external.iac.ResourceDriver.Read:output_type -> workflow.plugin.external.iac.ResourceReadResponse
+	103, // 179: workflow.plugin.external.iac.ResourceDriver.Update:output_type -> workflow.plugin.external.iac.ResourceUpdateResponse
+	105, // 180: workflow.plugin.external.iac.ResourceDriver.Delete:output_type -> workflow.plugin.external.iac.ResourceDeleteResponse
+	107, // 181: workflow.plugin.external.iac.ResourceDriver.Diff:output_type -> workflow.plugin.external.iac.ResourceDiffResponse
+	109, // 182: workflow.plugin.external.iac.ResourceDriver.Scale:output_type -> workflow.plugin.external.iac.ResourceScaleResponse
+	111, // 183: workflow.plugin.external.iac.ResourceDriver.HealthCheck:output_type -> workflow.plugin.external.iac.ResourceHealthCheckResponse
+	113, // 184: workflow.plugin.external.iac.ResourceDriver.SensitiveKeys:output_type -> workflow.plugin.external.iac.SensitiveKeysResponse
+	115, // 185: workflow.plugin.external.iac.ResourceDriver.Troubleshoot:output_type -> workflow.plugin.external.iac.TroubleshootResponse
+	118, // 186: workflow.plugin.external.iac.IaCStateBackend.Configure:output_type -> workflow.plugin.external.iac.ConfigureResponse
+	120, // 187: workflow.plugin.external.iac.IaCStateBackend.GetState:output_type -> workflow.plugin.external.iac.GetStateResponse
+	122, // 188: workflow.plugin.external.iac.IaCStateBackend.SaveState:output_type -> workflow.plugin.external.iac.SaveStateResponse
+	124, // 189: workflow.plugin.external.iac.IaCStateBackend.ListStates:output_type -> workflow.plugin.external.iac.ListStatesResponse
+	126, // 190: workflow.plugin.external.iac.IaCStateBackend.DeleteState:output_type -> workflow.plugin.external.iac.DeleteStateResponse
+	128, // 191: workflow.plugin.external.iac.IaCStateBackend.Lock:output_type -> workflow.plugin.external.iac.LockResponse
+	130, // 192: workflow.plugin.external.iac.IaCStateBackend.Unlock:output_type -> workflow.plugin.external.iac.UnlockResponse
+	132, // 193: workflow.plugin.external.iac.IaCStateBackend.ListBackendNames:output_type -> workflow.plugin.external.iac.ListBackendNamesResponse
+	134, // 194: workflow.plugin.external.iac.ResourceSensitiveInputDeclarer.SensitiveInputPaths:output_type -> workflow.plugin.external.iac.ResourceSensitiveInputPathsResponse
+	147, // [147:195] is the sub-list for method output_type
+	99,  // [99:147] is the sub-list for method input_type
+	99,  // [99:99] is the sub-list for extension type_name
+	99,  // [99:99] is the sub-list for extension extendee
+	0,   // [0:99] is the sub-list for field type_name
 }
 
 func init() { file_iac_proto_init() }
diff --git a/plugin/external/proto/iac.proto b/plugin/external/proto/iac.proto
index c8cc831e..804aadae 100644
--- a/plugin/external/proto/iac.proto
+++ b/plugin/external/proto/iac.proto
@@ -920,6 +920,9 @@ message ResourceUpdateRequest {
   string resource_type = 1;
   ResourceRef ref = 2;
   ResourceSpec spec = 3;
+  // Authoritative persisted state loaded at update dispatch, never a saved
+  // plan's Current or a prior Diff cache. Absent for legacy Update callers.
+  ResourceState prior_state = 4;
 }
 message ResourceUpdateResponse {
   ResourceOutput output = 1;
diff --git a/plugin/external/sdk/iacserver.go b/plugin/external/sdk/iacserver.go
index 9b7d432a..6afa51c0 100644
--- a/plugin/external/sdk/iacserver.go
+++ b/plugin/external/sdk/iacserver.go
@@ -2,10 +2,12 @@ package sdk
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"reflect"
 	"sort"
 	"strings"
+	"time"
 
 	goplugin "github.com/GoCodeAlone/go-plugin"
 	"github.com/GoCodeAlone/workflow/interfaces"
@@ -14,6 +16,7 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	pluginpkg "github.com/GoCodeAlone/workflow/plugin"
 	"github.com/GoCodeAlone/workflow/plugin/external/contract"
@@ -215,7 +218,11 @@ func registerIaCServicesOnly(s *grpc.Server, provider any) error {
 		pb.RegisterIaCProviderFinalizerServer(s, v)
 	}
 	if v, ok := provider.(pb.ResourceDriverServer); ok {
-		pb.RegisterResourceDriverServer(s, v)
+		if updater, ok := provider.(interfaces.ResourceStateUpdater); ok {
+			pb.RegisterResourceDriverServer(s, &resourceStateUpdateServer{ResourceDriverServer: v, updater: updater})
+		} else {
+			pb.RegisterResourceDriverServer(s, v)
+		}
 	}
 	if v, ok := provider.(pb.ResourceSensitiveInputDeclarerServer); ok {
 		pb.RegisterResourceSensitiveInputDeclarerServer(s, v)
@@ -232,6 +239,71 @@ func registerIaCServicesOnly(s *grpc.Server, provider any) error {
 	return nil
 }
 
+// resourceStateUpdateServer changes only Update; the embedded typed server
+// continues to own all other RPCs. Legacy providers are registered unchanged.
+type resourceStateUpdateServer struct {
+	pb.ResourceDriverServer
+	updater interfaces.ResourceStateUpdater
+}
+
+func (s *resourceStateUpdateServer) Update(ctx context.Context, req *pb.ResourceUpdateRequest) (*pb.ResourceUpdateResponse, error) {
+	ref := interfaces.ResourceRef{Name: req.GetRef().GetName(), Type: req.GetRef().GetType(), ProviderID: req.GetRef().GetProviderId()}
+	spec := interfaces.ResourceSpec{Name: req.GetSpec().GetName(), Type: req.GetSpec().GetType(), Size: interfaces.Size(req.GetSpec().GetSize()), DependsOn: append([]string(nil), req.GetSpec().GetDependsOn()...)}
+	if req.GetResourceType() != ref.Type {
+		return nil, status.Error(codes.InvalidArgument, "update resource type does not match identity")
+	}
+	if hint := req.GetSpec().GetHints(); hint != nil {
+		spec.Hints = &interfaces.ResourceHints{CPU: hint.GetCpu(), Memory: hint.GetMemory(), Storage: hint.GetStorage()}
+	}
+	decodeMap := func(data []byte, target *map[string]any) error {
+		if len(data) == 0 {
+			return nil
+		}
+		return json.Unmarshal(data, target)
+	}
+	if err := decodeMap(req.GetSpec().GetConfigJson(), &spec.Config); err != nil {
+		return nil, status.Error(codes.InvalidArgument, "update config JSON is invalid")
+	}
+	var prior *interfaces.ResourceState
+	if state := req.GetPriorState(); state != nil {
+		prior = &interfaces.ResourceState{
+			ID: state.GetId(), Name: state.GetName(), Type: state.GetType(), Provider: state.GetProvider(), ProviderRef: state.GetProviderRef(), ProviderID: state.GetProviderId(),
+			ConfigHash: state.GetConfigHash(), AppliedConfigSource: state.GetAppliedConfigSource(), Dependencies: append([]string(nil), state.GetDependencies()...),
+			CreatedAt: resourceUpdateTime(state.GetCreatedAt()), UpdatedAt: resourceUpdateTime(state.GetUpdatedAt()), LastDriftCheck: resourceUpdateTime(state.GetLastDriftCheck()),
+		}
+		if err := decodeMap(state.GetAppliedConfigJson(), &prior.AppliedConfig); err != nil {
+			return nil, status.Error(codes.InvalidArgument, "update prior config JSON is invalid")
+		}
+		if err := decodeMap(state.GetOutputsJson(), &prior.Outputs); err != nil {
+			return nil, status.Error(codes.InvalidArgument, "update prior outputs JSON is invalid")
+		}
+	}
+	if prior != nil {
+		if err := interfaces.ValidateUpdatePriorState(ref, spec, prior); err != nil {
+			return nil, nativeJobRPCError(err)
+		}
+	}
+	out, err := s.updater.UpdateWithState(ctx, ref, spec, prior)
+	if err != nil {
+		return nil, nativeJobRPCError(err)
+	}
+	if out == nil {
+		return &pb.ResourceUpdateResponse{}, nil
+	}
+	outputs, err := json.Marshal(out.Outputs)
+	if err != nil {
+		return nil, status.Error(codes.Internal, "update output JSON is invalid")
+	}
+	return &pb.ResourceUpdateResponse{Output: &pb.ResourceOutput{Name: out.Name, Type: out.Type, ProviderId: out.ProviderID, OutputsJson: outputs, Sensitive: out.Sensitive, Status: out.Status}}, nil
+}
+
+func resourceUpdateTime(value *timestamppb.Timestamp) time.Time {
+	if value == nil {
+		return time.Time{}
+	}
+	return value.AsTime()
+}
+
 // iacPluginServiceBridge is a minimal pb.PluginServiceServer registered on
 // the gRPC server by RegisterAllIaCProviderServices. It implements
 // GetContractRegistry (always) plus GetManifest (when diskManifest is wired
diff --git a/plugin/external/sdk/iacserver_test.go b/plugin/external/sdk/iacserver_test.go
index f01dea2d..86cab829 100644
--- a/plugin/external/sdk/iacserver_test.go
+++ b/plugin/external/sdk/iacserver_test.go
@@ -2,18 +2,303 @@ package sdk_test
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/json"
+	"fmt"
+	"maps"
 	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"reflect"
 	"strings"
 	"testing"
-
+	"time"
+
+	goplugin "github.com/GoCodeAlone/go-plugin"
+	"github.com/GoCodeAlone/workflow/iac/providerclient"
+	"github.com/GoCodeAlone/workflow/iac/wfctlhelpers"
+	"github.com/GoCodeAlone/workflow/interfaces"
+	"github.com/GoCodeAlone/workflow/plugin/external/contract"
+	"github.com/hashicorp/go-hclog"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	pb "github.com/GoCodeAlone/workflow/plugin/external/proto"
 	"github.com/GoCodeAlone/workflow/plugin/external/sdk"
 )
 
+type updateLegacyProvider struct {
+	fullProviderStub
+	pb.UnimplementedResourceDriverServer
+	legacyCalls int
+}
+
+func (p *updateLegacyProvider) Update(_ context.Context, req *pb.ResourceUpdateRequest) (*pb.ResourceUpdateResponse, error) {
+	p.legacyCalls++
+	return &pb.ResourceUpdateResponse{Output: &pb.ResourceOutput{Name: req.GetRef().GetName(), Type: req.GetRef().GetType(), ProviderId: req.GetRef().GetProviderId()}}, nil
+}
+
+type updateStateProvider struct {
+	updateLegacyProvider
+	stateCalls int
+	prior      *interfaces.ResourceState
+	process    bool
+}
+
+func (p *updateStateProvider) UpdateWithState(ctx context.Context, ref interfaces.ResourceRef, spec interfaces.ResourceSpec, prior *interfaces.ResourceState) (*interfaces.ResourceOutput, error) {
+	if spec.Type == "infra.legacy" && prior == nil {
+		response, err := p.Update(ctx, &pb.ResourceUpdateRequest{Ref: &pb.ResourceRef{Name: ref.Name, Type: ref.Type, ProviderId: ref.ProviderID}})
+		if err != nil {
+			return nil, err
+		}
+		return &interfaces.ResourceOutput{Name: response.Output.Name, Type: response.Output.Type, ProviderID: response.Output.ProviderId}, nil
+	}
+	if err := interfaces.ValidateUpdatePriorState(ref, spec, prior); err != nil {
+		return nil, err
+	}
+	p.stateCalls++
+	p.prior = prior
+	outputs := prior.Outputs
+	if p.process {
+		outputs = maps.Clone(prior.Outputs)
+		outputs["fixture_server_pid"] = os.Getpid()
+		outputs["fixture_state_calls"] = p.stateCalls
+		outputs["fixture_legacy_calls"] = p.legacyCalls
+	}
+	return &interfaces.ResourceOutput{Name: spec.Name, Type: spec.Type, ProviderID: ref.ProviderID, Outputs: outputs}, nil
+}
+
+func startUpdateProvider(t *testing.T, provider any) *grpc.ClientConn {
+	t.Helper()
+	srv := grpc.NewServer()
+	if err := sdk.RegisterAllIaCProviderServices(srv, provider); err != nil {
+		t.Fatal(err)
+	}
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	go func() { _ = srv.Serve(lis) }()
+	t.Cleanup(srv.Stop)
+	t.Cleanup(func() { _ = lis.Close() })
+	conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+	return conn
+}
+
+func TestSDK_UpdateWithState_RoundTripAndLegacy(t *testing.T) {
+	provider := &updateStateProvider{}
+	conn := startUpdateProvider(t, provider)
+	adapter := providerclient.New(conn, map[string]bool{providerclient.IaCServiceResourceDriver: true})
+	driver, err := adapter.ResourceDriver("infra.fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	prior := &interfaces.ResourceState{ID: "fixture", Name: "fixture", Type: "infra.fixture", Provider: "fixture-provider", ProviderRef: "fixture-instance", ProviderID: "fixture-id", ConfigHash: "fixture-hash", AppliedConfigSource: "apply", AppliedConfig: map[string]any{"credential_ref": "secrets://fixture/credential"}, Outputs: map[string]any{"rotation_generation": float64(9)}}
+	ref := interfaces.ResourceRef{Name: prior.Name, Type: prior.Type, ProviderID: prior.ProviderID}
+	spec := interfaces.ResourceSpec{Name: prior.Name, Type: prior.Type}
+	updater, ok := driver.(interfaces.ResourceStateUpdater)
+	if !ok {
+		t.Fatal("client omitted state-aware update")
+	}
+	out, err := updater.UpdateWithState(t.Context(), ref, spec, prior)
+	if err != nil || out == nil || provider.stateCalls != 1 || provider.legacyCalls != 0 || !reflect.DeepEqual(provider.prior, prior) {
+		t.Fatalf("SDK failed state-aware update round trip: err=%v state_calls=%d legacy_calls=%d", err, provider.stateCalls, provider.legacyCalls)
+	}
+	legacy := &updateLegacyProvider{}
+	legacyClient := pb.NewResourceDriverClient(startUpdateProvider(t, legacy))
+	_, err = legacyClient.Update(t.Context(), &pb.ResourceUpdateRequest{ResourceType: ref.Type, Ref: &pb.ResourceRef{Name: ref.Name, Type: ref.Type, ProviderId: ref.ProviderID}, Spec: &pb.ResourceSpec{Name: spec.Name, Type: spec.Type}})
+	if err != nil || legacy.legacyCalls != 1 {
+		t.Fatal("legacy Update RPC lost backward compatibility")
+	}
+	_, err = pb.NewResourceDriverClient(conn).Update(t.Context(), &pb.ResourceUpdateRequest{ResourceType: "infra.legacy", Ref: &pb.ResourceRef{Name: "legacy", Type: "infra.legacy", ProviderId: "legacy-id"}, Spec: &pb.ResourceSpec{Name: "legacy", Type: "infra.legacy"}})
+	if err != nil || provider.legacyCalls != 1 || provider.stateCalls != 1 {
+		t.Fatal("non-state-dependent legacy Update failed on a mixed state-aware provider")
+	}
+}
+
+func TestSDK_UpdateWithState_ApplyWithoutLookupPreservesLegacy(t *testing.T) {
+	for _, name := range []string{"legacy provider", "mixed legacy resource", "state-dependent resource"} {
+		t.Run(name, func(t *testing.T) {
+			legacy := &updateLegacyProvider{}
+			mixed := &updateStateProvider{}
+			var service any = legacy
+			resourceType := "infra.legacy"
+			if name != "legacy provider" {
+				service = mixed
+			}
+			if name == "state-dependent resource" {
+				resourceType = "infra.fixture"
+			}
+			provider := providerclient.New(startUpdateProvider(t, service), map[string]bool{
+				providerclient.IaCServiceResourceDriver: true,
+			})
+			state := interfaces.ResourceState{Name: "fixture", Type: resourceType, ProviderID: "fixture-id"}
+			plan := &interfaces.IaCPlan{Actions: []interfaces.PlanAction{{
+				Action:   "update",
+				Resource: interfaces.ResourceSpec{Name: state.Name, Type: state.Type},
+				Current:  &state,
+			}}}
+			result, err := wfctlhelpers.ApplyPlanWithHooks(t.Context(), provider, plan, wfctlhelpers.ApplyPlanHooks{})
+			if err != nil || result == nil {
+				t.Fatalf("apply without new hook: err=%v result=%v", err, result)
+			}
+			if name == "state-dependent resource" {
+				if len(result.Errors) != 1 || mixed.stateCalls != 0 || mixed.legacyCalls != 0 {
+					t.Fatal("state-dependent update without persisted state reached mutation")
+				}
+				return
+			}
+			calls := legacy.legacyCalls + mixed.legacyCalls
+			if len(result.Errors) != 0 || calls != 1 || mixed.stateCalls != 0 {
+				t.Fatalf("legacy update requires new lookup: errors=%v legacy_calls=%d native_calls=%d", result.Errors, calls, mixed.stateCalls)
+			}
+		})
+	}
+}
+
+func TestSDK_UpdateWithState_InvalidPriorNeverMutates(t *testing.T) {
+	outputs, _ := json.Marshal(map[string]any{"rotation_generation": 9})
+	valid := &pb.ResourceUpdateRequest{ResourceType: "infra.fixture", Ref: &pb.ResourceRef{Name: "fixture", Type: "infra.fixture", ProviderId: "fixture-id"}, Spec: &pb.ResourceSpec{Name: "fixture", Type: "infra.fixture"}, PriorState: &pb.ResourceState{Name: "fixture", Type: "infra.fixture", ProviderId: "fixture-id", OutputsJson: outputs}}
+	for _, name := range []string{"missing", "wrong_name", "wrong_type", "wrong_provider_id", "wrong_route_type", "invalid_outputs", "invalid_config"} {
+		t.Run(name, func(t *testing.T) {
+			provider := &updateStateProvider{}
+			client := pb.NewResourceDriverClient(startUpdateProvider(t, provider))
+			req := proto.Clone(valid).(*pb.ResourceUpdateRequest)
+			switch name {
+			case "missing":
+				req.PriorState = nil
+			case "wrong_name":
+				req.PriorState.Name = "other"
+			case "wrong_type":
+				req.PriorState.Type = "infra.other"
+			case "wrong_provider_id":
+				req.PriorState.ProviderId = "other-id"
+			case "wrong_route_type":
+				req.ResourceType = "infra.other"
+			case "invalid_outputs":
+				req.PriorState.OutputsJson = []byte(`{"secret":"sentinel-credential" broken}`)
+			case "invalid_config":
+				req.PriorState.AppliedConfigJson = []byte(`{"secret":"sentinel-credential" broken}`)
+			}
+			_, err := client.Update(t.Context(), req)
+			if status.Code(err) != codes.InvalidArgument || provider.stateCalls != 0 || provider.legacyCalls != 0 || strings.Contains(err.Error(), "sentinel-credential") {
+				t.Fatalf("invalid prior crossed mutation boundary: code=%v state_calls=%d legacy_calls=%d", status.Code(err), provider.stateCalls, provider.legacyCalls)
+			}
+		})
+	}
+}
+
+type updateConsumerPlugin struct{}
+
+func (*updateConsumerPlugin) GRPCServer(*goplugin.GRPCBroker, *grpc.Server) error {
+	return fmt.Errorf("consumer fixture is host-only")
+}
+
+func (*updateConsumerPlugin) GRPCClient(_ context.Context, _ *goplugin.GRPCBroker, conn *grpc.ClientConn) (any, error) {
+	return providerclient.New(conn, map[string]bool{providerclient.IaCServiceResourceDriver: true}), nil
+}
+
+// The child serves the real SDK; the parent uses only the public providerclient
+// and shared apply APIs against reopened filesystem state and a stale plan.
+func TestSDK_UpdateWithState_ExternalConsumerRestart(t *testing.T) {
+	if os.Getenv("WORKFLOW_IAC_UPDATE_TEST_PROCESS") == "1" {
+		sdk.ServeIaCPlugin(&updateStateProvider{process: true}, sdk.IaCServeOptions{})
+		return
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
+	defer cancel()
+	binary, err := os.Executable()
+	if err != nil {
+		t.Fatal(err)
+	}
+	data, err := os.ReadFile(binary)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Logf("SDK consumer executable sha256=%x", sha256.Sum256(data))
+	dir := t.TempDir()
+	configFile := filepath.Join(dir, "state.yaml")
+	stateDir := filepath.Join(dir, "state")
+	config := fmt.Sprintf("modules:\n  - name: state\n    type: iac.state\n    config:\n      backend: filesystem\n      directory: %q\n", stateDir)
+	if err := os.WriteFile(configFile, []byte(config), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	stale := interfaces.ResourceState{ID: "fixture", Name: "fixture", Type: "infra.fixture", Provider: "fixture-provider", ProviderID: "fixture-id", Outputs: map[string]any{"rotation_generation": float64(1), "credential_ref": "secrets://fixture/credential"}}
+	savedPlan, err := json.Marshal(interfaces.IaCPlan{Actions: []interfaces.PlanAction{{Action: "update", Resource: interfaces.ResourceSpec{Name: stale.Name, Type: stale.Type}, Current: &stale}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, generation := range []float64{7, 9} {
+		store, err := wfctlhelpers.ResolveStateStore(configFile, "", "")
+		if err != nil {
+			t.Fatal(err)
+		}
+		latest := stale
+		latest.Outputs = maps.Clone(stale.Outputs)
+		latest.Outputs["rotation_generation"] = generation
+		if err := store.SaveResource(ctx, latest); err != nil {
+			t.Fatal(err)
+		}
+		if err := store.Close(); err != nil {
+			t.Fatal(err)
+		}
+		restarted, err := wfctlhelpers.ResolveStateStore(configFile, "", "")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = restarted.Close() })
+		cmd := exec.CommandContext(ctx, binary, "-test.run=^TestSDK_UpdateWithState_ExternalConsumerRestart$")
+		cmd.Dir = dir
+		cmd.Env = []string{"PATH=" + os.Getenv("PATH"), "HOME=" + dir, "TMPDIR=" + dir, "WORKFLOW_IAC_UPDATE_TEST_PROCESS=1"}
+		client := goplugin.NewClient(&goplugin.ClientConfig{HandshakeConfig: contract.Handshake, Plugins: goplugin.PluginSet{"iac": &updateConsumerPlugin{}}, Cmd: cmd, StartTimeout: 10 * time.Second, Logger: hclog.NewNullLogger()})
+		t.Cleanup(client.Kill)
+		protocol, err := client.Client()
+		if err != nil {
+			t.Fatal(err)
+		}
+		consumer, err := protocol.Dispense("iac")
+		if err != nil {
+			t.Fatal(err)
+		}
+		provider, ok := consumer.(*providerclient.Adapter)
+		if !ok {
+			t.Fatal("SDK plugin did not expose the public providerclient adapter")
+		}
+		var plan interfaces.IaCPlan
+		if err := json.Unmarshal(savedPlan, &plan); err != nil {
+			t.Fatal(err)
+		}
+		result, err := wfctlhelpers.ApplyPlanWithHooks(ctx, provider, &plan, wfctlhelpers.ApplyPlanHooks{LookupCurrentState: func(ctx context.Context, ref interfaces.ResourceRef) (*interfaces.ResourceState, error) {
+			return restarted.GetResource(ctx, ref.Name)
+		}})
+		if err != nil || result == nil || len(result.Errors) != 0 || len(result.Resources) != 1 {
+			t.Fatalf("separate SDK process failed saved-plan dispatch: err=%v result=%v", err, result)
+		}
+		outputs := result.Resources[0].Outputs
+		if outputs["rotation_generation"] != generation || outputs["credential_ref"] != "secrets://fixture/credential" || outputs["fixture_server_pid"] != float64(cmd.Process.Pid) || outputs["fixture_state_calls"] != float64(1) || outputs["fixture_legacy_calls"] != float64(0) {
+			t.Fatal("separate SDK process did not receive latest persisted state on its exact update call")
+		}
+		client.Kill()
+		if !client.Exited() {
+			t.Fatal("owned SDK plugin process did not terminate")
+		}
+		t.Logf("SDK child pid=%d received generation=%.0f native_updates=1 legacy_updates=0; terminated=true", cmd.Process.Pid, generation)
+	}
+	state, err := os.ReadFile(filepath.Join(stateDir, "fixture.json"))
+	if err != nil || !strings.Contains(string(state), "secrets://fixture/credential") || strings.Contains(string(state), "sentinel-credential") {
+		t.Fatal("fixture state lost credential-reference-only persistence")
+	}
+}
+
 // TestRegisterAllIaCProviderServices_RequiredSatisfied_RegistersRequired
 // asserts that a provider satisfying IaCProviderRequiredServer succeeds and
 // the gRPC server actually advertises the typed service.