diff --git a/cmd/wfctl/infra_state_store.go b/cmd/wfctl/infra_state_store.go
index 7f1a68b6..e7dc9c91 100644
--- a/cmd/wfctl/infra_state_store.go
+++ b/cmd/wfctl/infra_state_store.go
@@ -12,6 +12,8 @@ import (
 	"github.com/GoCodeAlone/workflow/config"
 	"github.com/GoCodeAlone/workflow/interfaces"
 	"github.com/GoCodeAlone/workflow/module"
+	"github.com/GoCodeAlone/workflow/plugin/external"
+	pb "github.com/GoCodeAlone/workflow/plugin/external/proto"
 )
 
 // infraStateStore is the minimal state persistence interface used by wfctl
@@ -86,16 +88,13 @@ func resolveStateStore(cfgFile, envName string) (infraStateStore, error) {
 		return resolvePostgresStateStore(cfg)
 
 	case "spaces":
-		return nil, fmt.Errorf("iac.state backend %q is now plugin-served by workflow-plugin-digitalocean v1.1.0; "+
-			"install and load the plugin to use the Spaces backend (wfctl direct-path commands no longer support in-tree spaces)", backend)
+		return resolvePluginStateStore(context.Background(), backend, cfg)
 
 	case "s3":
-		return nil, fmt.Errorf("iac.state backend %q is now plugin-served by workflow-plugin-aws v1.1.0; "+
-			"install and load the plugin to use the S3 backend (wfctl direct-path commands no longer support in-tree s3)", backend)
+		return resolvePluginStateStore(context.Background(), backend, cfg)
 
 	case "gcs":
-		return nil, fmt.Errorf("iac.state backend %q is now plugin-served by workflow-plugin-gcp v1.1.0; "+
-			"install and load the plugin to use the GCS backend (wfctl direct-path commands no longer support in-tree gcs)", backend)
+		return resolvePluginStateStore(context.Background(), backend, cfg)
 
 	case "azure":
 		return nil, fmt.Errorf("azure state store backend not yet supported by wfctl direct-path commands; " +
@@ -107,6 +106,131 @@ func resolveStateStore(cfgFile, envName string) (infraStateStore, error) {
 	}
 }
 
+type pluginWfctlStateStore struct {
+	inner module.IaCStateStore
+	mgr   *external.ExternalPluginManager
+}
+
+func (s *pluginWfctlStateStore) ListResources(ctx context.Context) ([]interfaces.ResourceState, error) {
+	states, err := s.inner.ListStates(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]interfaces.ResourceState, 0, len(states))
+	for _, state := range states {
+		out = append(out, iacStateToResourceState(state))
+	}
+	return out, nil
+}
+
+func (s *pluginWfctlStateStore) SaveResource(ctx context.Context, state interfaces.ResourceState) error {
+	return s.inner.SaveState(ctx, resourceStateToIaCState(state))
+}
+
+func (s *pluginWfctlStateStore) DeleteResource(ctx context.Context, name string) error {
+	return s.inner.DeleteState(ctx, name)
+}
+
+func (s *pluginWfctlStateStore) Close() error {
+	if s.mgr == nil {
+		return nil
+	}
+	s.mgr.Shutdown()
+	return nil
+}
+
+func resolvePluginStateStore(ctx context.Context, backend string, cfg map[string]any) (infraStateStore, error) {
+	pluginDir := currentInfraPluginDir
+	if pluginDir == "" {
+		pluginDir = os.Getenv("WFCTL_PLUGIN_DIR")
+	}
+	if pluginDir == "" {
+		pluginDir = "./data/plugins"
+	}
+
+	entries, err := os.ReadDir(pluginDir)
+	if err != nil {
+		return nil, fmt.Errorf("iac.state backend %q is plugin-served but plugin directory %q is unavailable: %w", backend, pluginDir, err)
+	}
+
+	mgr := external.NewExternalPluginManager(pluginDir, nil)
+	for _, pluginName := range stateBackendPluginCandidates(backend, entries) {
+		clients, clientsErr := loadPluginStateBackendClients(mgr, pluginName, backend)
+		if clientsErr != nil {
+			mgr.Shutdown()
+			return nil, clientsErr
+		}
+		client, ok := clients[backend]
+		if !ok {
+			continue
+		}
+		store := module.NewGRPCIaCStateStore(client)
+		if err := store.Configure(ctx, backend, cfg); err != nil {
+			mgr.Shutdown()
+			return nil, fmt.Errorf("configure plugin-served iac.state backend %q via plugin %q: %w", backend, pluginName, err)
+		}
+		return &pluginWfctlStateStore{inner: store, mgr: mgr}, nil
+	}
+
+	mgr.Shutdown()
+	return nil, fmt.Errorf("iac.state backend %q is plugin-served but no installed plugin in %s advertises it", backend, pluginDir)
+}
+
+var loadPluginStateBackendClients = func(mgr *external.ExternalPluginManager, pluginName, backend string) (map[string]pb.IaCStateBackendClient, error) {
+	adapter, loadErr := mgr.LoadPlugin(pluginName)
+	if loadErr != nil {
+		return nil, fmt.Errorf("load plugin %q for iac.state backend %q: %w", pluginName, backend, loadErr)
+	}
+	clients, clientsErr := adapter.IaCStateBackendClients()
+	if clientsErr != nil {
+		return nil, fmt.Errorf("plugin %q iac.state backends: %w", pluginName, clientsErr)
+	}
+	return clients, nil
+}
+
+func stateBackendPluginCandidates(backend string, entries []os.DirEntry) []string {
+	seen := map[string]struct{}{}
+	var candidates []string
+	hasDir := func(name string) bool {
+		for _, entry := range entries {
+			if entry.IsDir() && entry.Name() == name {
+				return true
+			}
+		}
+		return false
+	}
+	add := func(name string) {
+		if strings.TrimSpace(name) == "" {
+			return
+		}
+		if _, ok := seen[name]; ok {
+			return
+		}
+		seen[name] = struct{}{}
+		candidates = append(candidates, name)
+	}
+	switch backend {
+	case "spaces":
+		if hasDir("digitalocean") {
+			add("digitalocean")
+		}
+	case "s3":
+		if hasDir("aws") {
+			add("aws")
+		}
+	case "gcs":
+		if hasDir("gcp") {
+			add("gcp")
+		}
+	}
+	for _, entry := range entries {
+		if entry.IsDir() {
+			add(entry.Name())
+		}
+	}
+	return candidates
+}
+
 // ── Filesystem backend ─────────────────────────────────────────────────────────
 
 // fsWfctlStateStore persists ResourceState records as JSON files under a
diff --git a/cmd/wfctl/infra_state_store_test.go b/cmd/wfctl/infra_state_store_test.go
index 950261aa..65e1118d 100644
--- a/cmd/wfctl/infra_state_store_test.go
+++ b/cmd/wfctl/infra_state_store_test.go
@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"io"
+	"io/fs"
 	"os"
 	"path/filepath"
 	"strings"
@@ -10,6 +11,9 @@ import (
 
 	"github.com/GoCodeAlone/workflow/interfaces"
 	"github.com/GoCodeAlone/workflow/module"
+	"github.com/GoCodeAlone/workflow/plugin/external"
+	pb "github.com/GoCodeAlone/workflow/plugin/external/proto"
+	"google.golang.org/grpc"
 )
 
 // ── TestResolveStateStore_NoEnv_FallsBackToBase ────────────────────────────────
@@ -59,6 +63,217 @@ func TestResolveStateStore_ReturnsDiscoverErrors(t *testing.T) {
 	}
 }
 
+func TestResolveStateStore_SpacesUsesPluginLoader(t *testing.T) {
+	dir := t.TempDir()
+	cfgPath := filepath.Join(dir, "infra.yaml")
+	if err := os.WriteFile(cfgPath, []byte(`
+modules:
+  - name: iac-state
+    type: iac.state
+    config:
+      backend: spaces
+      bucket: bmw-iac-state
+      region: nyc3
+`), 0o600); err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+
+	pluginDir := filepath.Join(dir, "plugins")
+	if err := os.Mkdir(pluginDir, 0o750); err != nil {
+		t.Fatalf("mkdir plugin dir: %v", err)
+	}
+	currentInfraPluginDir = pluginDir
+	defer func() { currentInfraPluginDir = "" }()
+
+	_, err := resolveStateStore(cfgPath, "")
+	if err == nil {
+		t.Fatal("expected missing plugin error, got nil")
+	}
+	if strings.Contains(err.Error(), "direct-path commands no longer support in-tree spaces") {
+		t.Fatalf("resolveStateStore returned legacy hard-coded spaces error: %v", err)
+	}
+	if !strings.Contains(err.Error(), `no installed plugin`) || !strings.Contains(err.Error(), pluginDir) {
+		t.Fatalf("error = %v, want plugin-loader context", err)
+	}
+}
+
+func TestResolvePluginStateStore_ConfiguresAdvertisedBackend(t *testing.T) {
+	dir := t.TempDir()
+	pluginDir := filepath.Join(dir, "plugins")
+	for _, name := range []string{"auth", "digitalocean"} {
+		if err := os.MkdirAll(filepath.Join(pluginDir, name), 0o750); err != nil {
+			t.Fatalf("mkdir plugin %s: %v", name, err)
+		}
+	}
+	currentInfraPluginDir = pluginDir
+	t.Cleanup(func() { currentInfraPluginDir = "" })
+
+	client := &testIaCStateBackendClient{
+		states: []*pb.IaCState{{
+			ResourceId:   "site-vpc",
+			ResourceType: "infra.vpc",
+			Provider:     "digitalocean",
+			ProviderId:   "vpc-123",
+			ConfigJson:   []byte(`{"region":"nyc3"}`),
+			OutputsJson:  []byte(`{"id":"vpc-123"}`),
+		}},
+	}
+	var loaded []string
+	orig := loadPluginStateBackendClients
+	loadPluginStateBackendClients = func(_ *external.ExternalPluginManager, pluginName, backend string) (map[string]pb.IaCStateBackendClient, error) {
+		loaded = append(loaded, pluginName)
+		if pluginName != "digitalocean" {
+			return map[string]pb.IaCStateBackendClient{}, nil
+		}
+		return map[string]pb.IaCStateBackendClient{backend: client}, nil
+	}
+	t.Cleanup(func() { loadPluginStateBackendClients = orig })
+
+	store, err := resolvePluginStateStore(t.Context(), "spaces", map[string]any{
+		"backend": "spaces",
+		"bucket":  "bmw-iac-state",
+	})
+	if err != nil {
+		t.Fatalf("resolvePluginStateStore: %v", err)
+	}
+	t.Cleanup(func() {
+		if closer, ok := store.(interface{ Close() error }); ok {
+			_ = closer.Close()
+		}
+	})
+	if len(loaded) == 0 || loaded[0] != "digitalocean" {
+		t.Fatalf("loaded plugins = %#v, want digitalocean first", loaded)
+	}
+	if client.configureBackend != "spaces" || !strings.Contains(string(client.configureJSON), "bmw-iac-state") {
+		t.Fatalf("Configure backend/json = %q %s, want spaces bucket config", client.configureBackend, client.configureJSON)
+	}
+	states, err := store.ListResources(t.Context())
+	if err != nil {
+		t.Fatalf("ListResources: %v", err)
+	}
+	if len(states) != 1 || states[0].ProviderID != "vpc-123" {
+		t.Fatalf("states = %#v, want vpc-123 resource from plugin backend", states)
+	}
+}
+
+func TestLoadPluginStateBackendClients_ReturnsLoadContext(t *testing.T) {
+	mgr := external.NewExternalPluginManager(t.TempDir(), nil)
+	defer mgr.Shutdown()
+
+	_, err := loadPluginStateBackendClients(mgr, "missing-plugin", "spaces")
+	if err == nil {
+		t.Fatal("expected load error")
+	}
+	if !strings.Contains(err.Error(), `load plugin "missing-plugin"`) || !strings.Contains(err.Error(), `backend "spaces"`) {
+		t.Fatalf("error = %v, want plugin/backend context", err)
+	}
+}
+
+func TestPluginWfctlStateStore_RoundTripsResourceState(t *testing.T) {
+	store := &pluginWfctlStateStore{inner: module.NewMemoryIaCStateStore()}
+	state := interfaces.ResourceState{
+		ID:            "site-vpc",
+		Name:          "site-vpc",
+		Type:          "infra.vpc",
+		Provider:      "digitalocean",
+		ProviderRef:   "do-provider",
+		ProviderID:    "vpc-123",
+		ConfigHash:    "config-hash",
+		AppliedConfig: map[string]any{"region": "nyc3"},
+		Outputs:       map[string]any{"id": "vpc-123"},
+		Dependencies:  []string{"site-project"},
+	}
+
+	if err := store.SaveResource(t.Context(), state); err != nil {
+		t.Fatalf("SaveResource: %v", err)
+	}
+	states, err := store.ListResources(t.Context())
+	if err != nil {
+		t.Fatalf("ListResources: %v", err)
+	}
+	if len(states) != 1 {
+		t.Fatalf("states = %d, want 1", len(states))
+	}
+	got := states[0]
+	if got.ID != state.ID || got.ProviderID != state.ProviderID || got.ConfigHash != state.ConfigHash {
+		t.Fatalf("state = %#v, want ID %q ProviderID %q ConfigHash %q", got, state.ID, state.ProviderID, state.ConfigHash)
+	}
+	if len(got.Dependencies) != 1 || got.Dependencies[0] != "site-project" {
+		t.Fatalf("Dependencies = %#v, want [site-project]", got.Dependencies)
+	}
+	if got.AppliedConfig["region"] != "nyc3" || got.Outputs["id"] != "vpc-123" {
+		t.Fatalf("state config/output = %#v %#v, want region/output preserved", got.AppliedConfig, got.Outputs)
+	}
+
+	if err := store.DeleteResource(t.Context(), state.ID); err != nil {
+		t.Fatalf("DeleteResource: %v", err)
+	}
+	states, err = store.ListResources(t.Context())
+	if err != nil {
+		t.Fatalf("ListResources after delete: %v", err)
+	}
+	if len(states) != 0 {
+		t.Fatalf("states after delete = %d, want 0", len(states))
+	}
+	if err := store.Close(); err != nil {
+		t.Fatalf("Close without manager: %v", err)
+	}
+}
+
+func TestStateBackendPluginCandidates_PrioritizesKnownProviders(t *testing.T) {
+	dir := t.TempDir()
+	for _, name := range []string{"auth", "digitalocean", "aws", "gcp"} {
+		if err := os.Mkdir(filepath.Join(dir, name), 0o750); err != nil {
+			t.Fatalf("mkdir %s: %v", name, err)
+		}
+	}
+	if err := os.WriteFile(filepath.Join(dir, "README.md"), []byte("not a plugin"), 0o600); err != nil {
+		t.Fatalf("write file entry: %v", err)
+	}
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatalf("ReadDir: %v", err)
+	}
+
+	for _, tt := range []struct {
+		backend string
+		first   string
+	}{
+		{backend: "spaces", first: "digitalocean"},
+		{backend: "s3", first: "aws"},
+		{backend: "gcs", first: "gcp"},
+	} {
+		t.Run(tt.backend, func(t *testing.T) {
+			got := stateBackendPluginCandidates(tt.backend, entries)
+			if len(got) == 0 || got[0] != tt.first {
+				t.Fatalf("candidates = %#v, want first %q", got, tt.first)
+			}
+			seen := map[string]bool{}
+			for _, name := range got {
+				if name == "README.md" {
+					t.Fatalf("candidates included file entry: %#v", got)
+				}
+				if seen[name] {
+					t.Fatalf("candidates include duplicate %q: %#v", name, got)
+				}
+				seen[name] = true
+			}
+		})
+	}
+}
+
+func TestStateBackendPluginCandidates_SkipsBlankAndNonDirectories(t *testing.T) {
+	entries := []os.DirEntry{
+		testDirEntry{name: " ", dir: true},
+		testDirEntry{name: "notes.txt", dir: false},
+		testDirEntry{name: "custom", dir: true},
+	}
+	got := stateBackendPluginCandidates("custom", entries)
+	if len(got) != 1 || got[0] != "custom" {
+		t.Fatalf("candidates = %#v, want [custom]", got)
+	}
+}
+
 // ── TestResolveStateStore_EnvOverride_UsesEnvConfig ───────────────────────────
 
 // TestResolveStateStore_EnvOverride_UsesEnvConfig verifies that when envName
@@ -369,3 +584,53 @@ func TestFSStateStore_RoundTripsDependencies(t *testing.T) {
 		t.Fatalf("Dependencies = %#v, want [site-db site-dns]", states[0].Dependencies)
 	}
 }
+
+type testDirEntry struct {
+	name string
+	dir  bool
+}
+
+func (e testDirEntry) Name() string               { return e.name }
+func (e testDirEntry) IsDir() bool                { return e.dir }
+func (e testDirEntry) Type() fs.FileMode          { return 0 }
+func (e testDirEntry) Info() (fs.FileInfo, error) { return nil, nil }
+
+type testIaCStateBackendClient struct {
+	configureBackend string
+	configureJSON    []byte
+	states           []*pb.IaCState
+}
+
+func (c *testIaCStateBackendClient) Configure(_ context.Context, req *pb.ConfigureRequest, _ ...grpc.CallOption) (*pb.ConfigureResponse, error) {
+	c.configureBackend = req.BackendName
+	c.configureJSON = req.ConfigJson
+	return &pb.ConfigureResponse{}, nil
+}
+
+func (c *testIaCStateBackendClient) GetState(_ context.Context, _ *pb.GetStateRequest, _ ...grpc.CallOption) (*pb.GetStateResponse, error) {
+	return &pb.GetStateResponse{}, nil
+}
+
+func (c *testIaCStateBackendClient) SaveState(_ context.Context, _ *pb.SaveStateRequest, _ ...grpc.CallOption) (*pb.SaveStateResponse, error) {
+	return &pb.SaveStateResponse{}, nil
+}
+
+func (c *testIaCStateBackendClient) ListStates(_ context.Context, _ *pb.ListStatesRequest, _ ...grpc.CallOption) (*pb.ListStatesResponse, error) {
+	return &pb.ListStatesResponse{States: c.states}, nil
+}
+
+func (c *testIaCStateBackendClient) DeleteState(_ context.Context, _ *pb.DeleteStateRequest, _ ...grpc.CallOption) (*pb.DeleteStateResponse, error) {
+	return &pb.DeleteStateResponse{}, nil
+}
+
+func (c *testIaCStateBackendClient) Lock(_ context.Context, _ *pb.LockRequest, _ ...grpc.CallOption) (*pb.LockResponse, error) {
+	return &pb.LockResponse{}, nil
+}
+
+func (c *testIaCStateBackendClient) Unlock(_ context.Context, _ *pb.UnlockRequest, _ ...grpc.CallOption) (*pb.UnlockResponse, error) {
+	return &pb.UnlockResponse{}, nil
+}
+
+func (c *testIaCStateBackendClient) ListBackendNames(_ context.Context, _ *pb.ListBackendNamesRequest, _ ...grpc.CallOption) (*pb.ListBackendNamesResponse, error) {
+	return &pb.ListBackendNamesResponse{BackendNames: []string{"spaces"}}, nil
+}
diff --git a/module/iac_state_grpc_client.go b/module/iac_state_grpc_client.go
index 522277d5..6c0019e7 100644
--- a/module/iac_state_grpc_client.go
+++ b/module/iac_state_grpc_client.go
@@ -102,6 +102,13 @@ type grpcIaCStateStore struct {
 	client pb.IaCStateBackendClient
 }
 
+// NewGRPCIaCStateStore wraps an IaCStateBackendClient as an IaCStateStore.
+// It is exported for wfctl direct-path commands, which load plugin-served
+// state backends without constructing a full engine.
+func NewGRPCIaCStateStore(c pb.IaCStateBackendClient) *grpcIaCStateStore {
+	return newGRPCIaCStateStore(c)
+}
+
 // newGRPCIaCStateStore wraps an IaCStateBackendClient as an IaCStateStore.
 func newGRPCIaCStateStore(c pb.IaCStateBackendClient) *grpcIaCStateStore {
 	return &grpcIaCStateStore{client: c}
diff --git a/module/iac_state_grpc_client_test.go b/module/iac_state_grpc_client_test.go
index f5b0406c..ff7797de 100644
--- a/module/iac_state_grpc_client_test.go
+++ b/module/iac_state_grpc_client_test.go
@@ -67,6 +67,14 @@ func TestGRPCIaCStateStoreConfigure(t *testing.T) {
 	}
 }
 
+func TestNewGRPCIaCStateStoreWrapsClient(t *testing.T) {
+	fake := &captureStateBackendClient{}
+	store := NewGRPCIaCStateStore(fake)
+	if store.client != fake {
+		t.Fatalf("client = %p, want %p", store.client, fake)
+	}
+}
+
 func TestGRPCIaCStateStoreRoundTrip(t *testing.T) {
 	lis := bufconn.Listen(4 << 20)
 	t.Cleanup(func() { _ = lis.Close() })