diff --git a/internal/command/apply.go b/internal/command/apply.go
index 74e8e0378e..92ef5f4b69 100644
--- a/internal/command/apply.go
+++ b/internal/command/apply.go
@@ -210,17 +210,17 @@ func (c *ApplyCommand) PrepareBackend(planFile *planfile.WrappedPlanFile, args *
 			))
 			return nil, diags
 		}
-		if plan.Backend == nil {
+
+		if plan.Backend == nil && plan.StateStore == nil {
 			// Should never happen; always indicates a bug in the creation of the plan file
 			diags = diags.Append(tfdiags.Sourceless(
 				tfdiags.Error,
 				"Failed to read plan from plan file",
-				"The given plan file does not have a valid backend configuration. This is a bug in the Terraform command that generated this plan file.",
+				"The given plan file has neither a valid backend nor state store configuration. This is a bug in the Terraform command that generated this plan file.",
 			))
 			return nil, diags
 		}
-		// TODO: Update BackendForLocalPlan to use state storage, and plan to be able to contain State Store config details
-		be, beDiags = c.BackendForLocalPlan(*plan.Backend)
+		be, beDiags = c.BackendForLocalPlan(plan)
 	} else {
 
 		// Load the backend
diff --git a/internal/command/apply_test.go b/internal/command/apply_test.go
index 38b416f711..2e7e22a850 100644
--- a/internal/command/apply_test.go
+++ b/internal/command/apply_test.go
@@ -19,6 +19,8 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/hashicorp/cli"
+	version "github.com/hashicorp/go-version"
+	tfaddr "github.com/hashicorp/terraform-registry-address"
 	"github.com/zclconf/go-cty/cty"
 
 	"github.com/hashicorp/terraform/internal/addrs"
@@ -701,6 +703,269 @@ func TestApply_plan(t *testing.T) {
 	}
 }
 
+// Test the ability to apply a plan file with a state store.
+//
+// The state store's details (provider, config, etc) are supplied by the plan,
+// which allows this test to not use any configuration.
+func TestApply_plan_stateStore(t *testing.T) {
+	// Disable test mode so input would be asked
+	test = false
+	defer func() { test = true }()
+
+	// Set some default reader/writers for the inputs
+	defaultInputReader = new(bytes.Buffer)
+	defaultInputWriter = new(bytes.Buffer)
+
+	// Create the plan file that includes a state store
+	ver := version.Must(version.NewVersion("1.2.3"))
+	providerCfg := cty.ObjectVal(map[string]cty.Value{
+		"region": cty.StringVal("spain"),
+	})
+	providerCfgRaw, err := plans.NewDynamicValue(providerCfg, providerCfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	storeCfg := cty.ObjectVal(map[string]cty.Value{
+		"value": cty.StringVal("foobar"),
+	})
+	storeCfgRaw, err := plans.NewDynamicValue(storeCfg, storeCfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	plan := &plans.Plan{
+		Changes: plans.NewChangesSrc(),
+
+		// We'll default to the fake plan being both applyable and complete,
+		// since that's what most tests expect. Tests can override these
+		// back to false again afterwards if they need to.
+		Applyable: true,
+		Complete:  true,
+
+		StateStore: &plans.StateStore{
+			Type: "test_store",
+			Provider: &plans.Provider{
+				Version: ver,
+				Source: &tfaddr.Provider{
+					Hostname:  tfaddr.DefaultProviderRegistryHost,
+					Namespace: "hashicorp",
+					Type:      "test",
+				},
+				Config: providerCfgRaw,
+			},
+			Config:    storeCfgRaw,
+			Workspace: "default",
+		},
+	}
+
+	// Create a plan file on disk
+	//
+	// In this process we create a plan file describing the creation of a test_instance.foo resource.
+	state := testState() // State describes
+	_, snap := testModuleWithSnapshot(t, "apply")
+	planPath := testPlanFile(t, snap, state, plan)
+
+	// Create a mock, to be used as the pluggable state store described in the planfile
+	mock := testStateStoreMockWithChunkNegotiation(t, 1000)
+	view, done := testView(t)
+	c := &ApplyCommand{
+		Meta: Meta{
+			testingOverrides: &testingOverrides{
+				Providers: map[addrs.Provider]providers.Factory{
+					addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+				},
+			},
+			View: view,
+		},
+	}
+
+	args := []string{
+		planPath,
+	}
+	code := c.Run(args)
+	output := done(t)
+	if code != 0 {
+		t.Fatalf("bad: %d\n\n%s", code, output.Stderr())
+	}
+
+	if !mock.WriteStateBytesCalled {
+		t.Fatal("expected the test to write new state when applying the plan, but WriteStateBytesCalled is false on the mock provider.")
+	}
+}
+
+// Test unhappy paths when applying a plan file describing a state store.
+func TestApply_plan_stateStore_errorCases(t *testing.T) {
+	// Disable test mode so input would be asked
+	test = false
+	defer func() { test = true }()
+
+	t.Run("error when the provider doesn't include the state store named in the plan", func(t *testing.T) {
+		// Set some default reader/writers for the inputs
+		defaultInputReader = new(bytes.Buffer)
+		defaultInputWriter = new(bytes.Buffer)
+
+		// Create the plan file that includes a state store
+		ver := version.Must(version.NewVersion("1.2.3"))
+		providerCfg := cty.ObjectVal(map[string]cty.Value{
+			"region": cty.StringVal("spain"),
+		})
+		providerCfgRaw, err := plans.NewDynamicValue(providerCfg, providerCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+		storeCfg := cty.ObjectVal(map[string]cty.Value{
+			"value": cty.StringVal("foobar"),
+		})
+		storeCfgRaw, err := plans.NewDynamicValue(storeCfg, storeCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		plan := &plans.Plan{
+			Changes: plans.NewChangesSrc(),
+
+			// We'll default to the fake plan being both applyable and complete,
+			// since that's what most tests expect. Tests can override these
+			// back to false again afterwards if they need to.
+			Applyable: true,
+			Complete:  true,
+
+			StateStore: &plans.StateStore{
+				Type: "test_doesnt_exist", // Mismatched with test_store in the provider
+				Provider: &plans.Provider{
+					Version: ver,
+					Source: &tfaddr.Provider{
+						Hostname:  tfaddr.DefaultProviderRegistryHost,
+						Namespace: "hashicorp",
+						Type:      "test",
+					},
+					Config: providerCfgRaw,
+				},
+				Config:    storeCfgRaw,
+				Workspace: "default",
+			},
+		}
+
+		// Create a plan file on disk
+		//
+		// In this process we create a plan file describing the creation of a test_instance.foo resource.
+		state := testState() // State describes
+		_, snap := testModuleWithSnapshot(t, "apply")
+		planPath := testPlanFile(t, snap, state, plan)
+
+		// Create a mock, to be used as the pluggable state store described in the planfile
+		mock := testStateStoreMockWithChunkNegotiation(t, 1000)
+		view, done := testView(t)
+		c := &ApplyCommand{
+			Meta: Meta{
+				testingOverrides: &testingOverrides{
+					Providers: map[addrs.Provider]providers.Factory{
+						addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+					},
+				},
+				View: view,
+			},
+		}
+
+		args := []string{
+			planPath,
+			"-no-color",
+		}
+		code := c.Run(args)
+		output := done(t)
+		if code != 1 {
+			t.Fatalf("expected an error but got none: %d\n\n%s", code, output.Stdout())
+		}
+		expectedErr := "Error: State store not implemented by the provider"
+		if !strings.Contains(output.Stderr(), expectedErr) {
+			t.Fatalf("expected error to include %q, but got:\n%s", expectedErr, output.Stderr())
+		}
+	})
+
+	t.Run("error when the provider doesn't implement state stores", func(t *testing.T) {
+		// Set some default reader/writers for the inputs
+		defaultInputReader = new(bytes.Buffer)
+		defaultInputWriter = new(bytes.Buffer)
+
+		// Create the plan file that includes a state store
+		ver := version.Must(version.NewVersion("1.2.3"))
+		providerCfg := cty.ObjectVal(map[string]cty.Value{
+			"region": cty.StringVal("spain"),
+		})
+		providerCfgRaw, err := plans.NewDynamicValue(providerCfg, providerCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+		storeCfg := cty.ObjectVal(map[string]cty.Value{
+			"value": cty.StringVal("foobar"),
+		})
+		storeCfgRaw, err := plans.NewDynamicValue(storeCfg, storeCfg.Type())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		plan := &plans.Plan{
+			Changes: plans.NewChangesSrc(),
+
+			// We'll default to the fake plan being both applyable and complete,
+			// since that's what most tests expect. Tests can override these
+			// back to false again afterwards if they need to.
+			Applyable: true,
+			Complete:  true,
+
+			StateStore: &plans.StateStore{
+				Type: "test_store",
+				Provider: &plans.Provider{
+					Version: ver,
+					Source: &tfaddr.Provider{
+						Hostname:  tfaddr.DefaultProviderRegistryHost,
+						Namespace: "hashicorp",
+						Type:      "test",
+					},
+					Config: providerCfgRaw,
+				},
+				Config:    storeCfgRaw,
+				Workspace: "default",
+			},
+		}
+
+		// Create a plan file on disk
+		//
+		// In this process we create a plan file describing the creation of a test_instance.foo resource.
+		state := testState() // State describes
+		_, snap := testModuleWithSnapshot(t, "apply")
+		planPath := testPlanFile(t, snap, state, plan)
+
+		// Create a mock, to be used as the pluggable state store described in the planfile
+		mock := &testing_provider.MockProvider{}
+		view, done := testView(t)
+		c := &ApplyCommand{
+			Meta: Meta{
+				testingOverrides: &testingOverrides{
+					Providers: map[addrs.Provider]providers.Factory{
+						addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+					},
+				},
+				View: view,
+			},
+		}
+
+		args := []string{
+			planPath,
+			"-no-color",
+		}
+		code := c.Run(args)
+		output := done(t)
+		if code != 1 {
+			t.Fatalf("expected an error but got none: %d\n\n%s", code, output.Stdout())
+		}
+		expectedErr := "Error: Provider does not support pluggable state storage"
+		if !strings.Contains(output.Stderr(), expectedErr) {
+			t.Fatalf("expected error to include %q, but got:\n%s", expectedErr, output.Stderr())
+		}
+	})
+}
+
 func TestApply_plan_backup(t *testing.T) {
 	statePath := testTempFile(t)
 	backupPath := testTempFile(t)
diff --git a/internal/command/e2etest/primary_test.go b/internal/command/e2etest/primary_test.go
index ebffa9c37d..0da21067c0 100644
--- a/internal/command/e2etest/primary_test.go
+++ b/internal/command/e2etest/primary_test.go
@@ -314,6 +314,88 @@ func TestPrimary_stateStore(t *testing.T) {
 	}
 }
 
+func TestPrimary_stateStore_planFile(t *testing.T) {
+
+	if !canRunGoBuild {
+		// We're running in a separate-build-then-run context, so we can't
+		// currently execute this test which depends on being able to build
+		// new executable at runtime.
+		//
+		// (See the comment on canRunGoBuild's declaration for more information.)
+		t.Skip("can't run without building a new provider executable")
+	}
+
+	t.Setenv(e2e.TestExperimentFlag, "true")
+	terraformBin := e2e.GoBuild("github.com/hashicorp/terraform", "terraform")
+
+	fixturePath := filepath.Join("testdata", "full-workflow-with-state-store-fs")
+	tf := e2e.NewBinary(t, terraformBin, fixturePath)
+
+	// In order to test integration with PSS we need a provider plugin implementing a state store.
+	// Here will build the simple6 (built with protocol v6) provider, which implements PSS.
+	simple6Provider := filepath.Join(tf.WorkDir(), "terraform-provider-simple6")
+	simple6ProviderExe := e2e.GoBuild("github.com/hashicorp/terraform/internal/provider-simple-v6/main", simple6Provider)
+
+	// Move the provider binaries into a directory that we will point terraform
+	// to using the -plugin-dir cli flag.
+	platform := getproviders.CurrentPlatform.String()
+	hashiDir := "cache/registry.terraform.io/hashicorp/"
+	if err := os.MkdirAll(tf.Path(hashiDir, "simple6/0.0.1/", platform), os.ModePerm); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Rename(simple6ProviderExe, tf.Path(hashiDir, "simple6/0.0.1/", platform, "terraform-provider-simple6")); err != nil {
+		t.Fatal(err)
+	}
+
+	//// INIT
+	stdout, stderr, err := tf.Run("init", "-enable-pluggable-state-storage-experiment=true", "-plugin-dir=cache", "-no-color")
+	if err != nil {
+		t.Fatalf("unexpected init error: %s\nstderr:\n%s", err, stderr)
+	}
+
+	if !strings.Contains(stdout, "Terraform created an empty state file for the default workspace") {
+		t.Errorf("notice about creating the default workspace is missing from init output:\n%s", stdout)
+	}
+
+	//// PLAN
+	planFile := "testplan"
+	_, stderr, err = tf.Run("plan", "-out="+planFile, "-no-color")
+	if err != nil {
+		t.Fatalf("unexpected apply error: %s\nstderr:\n%s", err, stderr)
+	}
+
+	//// APPLY
+	stdout, stderr, err = tf.Run("apply", "-auto-approve", "-no-color", planFile)
+	if err != nil {
+		t.Fatalf("unexpected apply error: %s\nstderr:\n%s", err, stderr)
+	}
+
+	if !strings.Contains(stdout, "Resources: 1 added, 0 changed, 0 destroyed") {
+		t.Errorf("incorrect apply tally; want 1 added:\n%s", stdout)
+	}
+
+	// Check the statefile saved by the fs state store.
+	path := "states/default/terraform.tfstate"
+	f, err := tf.OpenFile(path)
+	if err != nil {
+		t.Fatalf("unexpected error opening state file %s: %s\nstderr:\n%s", path, err, stderr)
+	}
+	defer f.Close()
+
+	stateFile, err := statefile.Read(f)
+	if err != nil {
+		t.Fatalf("unexpected error reading statefile %s: %s\nstderr:\n%s", path, err, stderr)
+	}
+
+	r := stateFile.State.RootModule().Resources
+	if len(r) != 1 {
+		t.Fatalf("expected state to include one resource, but got %d", len(r))
+	}
+	if _, ok := r["terraform_data.my-data"]; !ok {
+		t.Fatalf("expected state to include terraform_data.my-data but it's missing")
+	}
+}
+
 func TestPrimary_stateStore_inMem(t *testing.T) {
 	if !canRunGoBuild {
 		// We're running in a separate-build-then-run context, so we can't
diff --git a/internal/command/meta_backend.go b/internal/command/meta_backend.go
index b26c70e538..e8e6aceb9c 100644
--- a/internal/command/meta_backend.go
+++ b/internal/command/meta_backend.go
@@ -327,40 +327,217 @@ func (m *Meta) selectWorkspace(b backend.Backend) error {
 	return m.SetWorkspace(workspace)
 }
 
-// BackendForLocalPlan is similar to Backend, but uses backend settings that were
-// stored in a plan.
+// BackendForLocalPlan is similar to Backend, but uses settings that were
+// stored in a plan when preparing the returned operations backend.
+// The plan's data may describe `backend` or `state_store` configuration.
 //
 // The current workspace name is also stored as part of the plan, and so this
 // method will check that it matches the currently-selected workspace name
 // and produce error diagnostics if not.
-func (m *Meta) BackendForLocalPlan(settings plans.Backend) (backendrun.OperationsBackend, tfdiags.Diagnostics) {
+func (m *Meta) BackendForLocalPlan(plan *plans.Plan) (backendrun.OperationsBackend, tfdiags.Diagnostics) {
 	var diags tfdiags.Diagnostics
 
-	f := backendInit.Backend(settings.Type)
-	if f == nil {
-		diags = diags.Append(errBackendSavedUnknown{settings.Type})
-		return nil, diags
-	}
-	b := f()
-	log.Printf("[TRACE] Meta.BackendForLocalPlan: instantiated backend of type %T", b)
+	var b backend.Backend
+	switch {
+	case plan.StateStore != nil:
+		settings := plan.StateStore
+
+		// BackendForLocalPlan is used in the context of an apply command using a plan file,
+		// so we can read locks directly from the lock file and trust it contains what we need.
+		locks, lockDiags := m.lockedDependencies()
+		diags = diags.Append(lockDiags)
+		if lockDiags.HasErrors() {
+			return nil, diags
+		}
 
-	schema := b.ConfigSchema()
-	configVal, err := settings.Config.Decode(schema.ImpliedType())
-	if err != nil {
-		diags = diags.Append(fmt.Errorf("saved backend configuration is invalid: %w", err))
-		return nil, diags
-	}
+		factories, err := m.ProviderFactoriesFromLocks(locks)
+		if err != nil {
+			// This may happen if the provider isn't present in the provider cache.
+			// This should be caught earlier by logic that diffs the config against the backend state file.
+			return nil, diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Provider unavailable",
+				Detail: fmt.Sprintf("Terraform experienced an error when trying to use provider %s (%q) to initialize the %q state store: %s",
+					settings.Provider.Source.Type,
+					settings.Provider.Source,
+					settings.Type,
+					err),
+			})
+		}
 
-	newVal, validateDiags := b.PrepareConfig(configVal)
-	diags = diags.Append(validateDiags)
-	if validateDiags.HasErrors() {
-		return nil, diags
-	}
+		factory, exists := factories[*settings.Provider.Source]
+		if !exists {
+			return nil, diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Provider unavailable",
+				Detail: fmt.Sprintf("The provider %s (%q) is required to initialize the %q state store, but the matching provider factory is missing. This is a bug in Terraform and should be reported.",
+					settings.Provider.Source.Type,
+					settings.Provider.Source,
+					settings.Type,
+				),
+			})
+		}
 
-	configureDiags := b.Configure(newVal)
-	diags = diags.Append(configureDiags)
-	if configureDiags.HasErrors() {
-		return nil, diags
+		provider, err := factory()
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("error when obtaining provider instance during state store initialization: %w", err))
+			return nil, diags
+		}
+		log.Printf("[TRACE] Meta.BackendForLocalPlan: launched instance of provider %s (%q)",
+			settings.Provider.Source.Type,
+			settings.Provider.Source,
+		)
+
+		// We purposefully don't have a deferred call to the provider's Close method here because the calling code needs a
+		// running provider instance inside the returned backend.Backend instance.
+		// Stopping the provider process is the responsibility of the calling code.
+
+		resp := provider.GetProviderSchema()
+
+		if len(resp.StateStores) == 0 {
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Provider does not support pluggable state storage",
+				Detail: fmt.Sprintf("There are no state stores implemented by provider %s (%q)",
+					settings.Provider.Source.Type,
+					settings.Provider.Source),
+			})
+			return nil, diags
+		}
+
+		stateStoreSchema, exists := resp.StateStores[settings.Type]
+		if !exists {
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "State store not implemented by the provider",
+				Detail: fmt.Sprintf("State store %q is not implemented by provider %s (%q)",
+					settings.Type,
+					settings.Provider.Source.Type,
+					settings.Provider.Source,
+				),
+			})
+			return nil, diags
+		}
+
+		// Get the provider config from the backend state file.
+		providerConfigVal, err := settings.Provider.Config.Decode(resp.Provider.Body.ImpliedType())
+		if err != nil {
+			diags = diags.Append(
+				&hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Error reading provider configuration state",
+					Detail: fmt.Sprintf("Terraform experienced an error reading provider configuration for provider %s (%q) while configuring state store %s: %s",
+						settings.Provider.Source.Type,
+						settings.Provider.Source,
+						settings.Type,
+						err,
+					),
+				},
+			)
+			return nil, diags
+		}
+
+		// Get the state store config from the backend state file.
+		stateStoreConfigVal, err := settings.Config.Decode(stateStoreSchema.Body.ImpliedType())
+		if err != nil {
+			diags = diags.Append(
+				&hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Error reading state store configuration state",
+					Detail: fmt.Sprintf("Terraform experienced an error reading state store configuration for state store %s in provider %s (%q): %s",
+						settings.Type,
+						settings.Provider.Source.Type,
+						settings.Provider.Source,
+						err,
+					),
+				},
+			)
+			return nil, diags
+		}
+
+		// Validate and configure the provider
+		//
+		// NOTE: there are no marks we need to remove at this point.
+		// We haven't added marks since the provider config from the backend state was used
+		// because the state-storage provider's config isn't going to be presented to the user via terminal output or diags.
+		validateResp := provider.ValidateProviderConfig(providers.ValidateProviderConfigRequest{
+			Config: providerConfigVal,
+		})
+		diags = diags.Append(validateResp.Diagnostics)
+		if diags.HasErrors() {
+			return nil, diags
+		}
+
+		configureResp := provider.ConfigureProvider(providers.ConfigureProviderRequest{
+			TerraformVersion: tfversion.SemVer.String(),
+			Config:           providerConfigVal,
+		})
+		diags = diags.Append(configureResp.Diagnostics)
+		if diags.HasErrors() {
+			return nil, diags
+		}
+
+		// Now that the provider is configured we can begin using the state store through
+		// the backend.Backend interface.
+		p, err := backendPluggable.NewPluggable(provider, settings.Type)
+		if err != nil {
+			diags = diags.Append(err)
+			return nil, diags
+		}
+
+		// Validate and configure the state store
+		//
+		// Note: we do not use the value returned from PrepareConfig for state stores,
+		// however that old approach is still used with backends for compatibility reasons.
+		_, validateDiags := p.PrepareConfig(stateStoreConfigVal)
+		diags = diags.Append(validateDiags)
+		if validateDiags.HasErrors() {
+			return nil, diags
+		}
+
+		configureDiags := p.Configure(stateStoreConfigVal)
+		diags = diags.Append(configureDiags)
+		if configureDiags.HasErrors() {
+			return nil, diags
+		}
+		log.Printf("[TRACE] Meta.BackendForLocalPlan: finished configuring state store %s in provider %s (%q)",
+			settings.Type,
+			settings.Provider.Source.Type,
+			settings.Provider.Source,
+		)
+
+		// The fully configured Pluggable is used as the instance of backend.Backend
+		b = p
+
+	default:
+		settings := plan.Backend
+
+		f := backendInit.Backend(settings.Type)
+		if f == nil {
+			diags = diags.Append(errBackendSavedUnknown{settings.Type})
+			return nil, diags
+		}
+		b = f()
+		log.Printf("[TRACE] Meta.BackendForLocalPlan: instantiated backend of type %T", b)
+
+		schema := b.ConfigSchema()
+		configVal, err := settings.Config.Decode(schema.ImpliedType())
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("saved backend configuration is invalid: %w", err))
+			return nil, diags
+		}
+
+		newVal, validateDiags := b.PrepareConfig(configVal)
+		diags = diags.Append(validateDiags)
+		if validateDiags.HasErrors() {
+			return nil, diags
+		}
+
+		configureDiags := b.Configure(newVal)
+		diags = diags.Append(configureDiags)
+		if configureDiags.HasErrors() {
+			return nil, diags
+		}
 	}
 
 	// If the backend supports CLI initialization, do it.
diff --git a/internal/command/meta_backend_test.go b/internal/command/meta_backend_test.go
index cd8069cf89..1c06899471 100644
--- a/internal/command/meta_backend_test.go
+++ b/internal/command/meta_backend_test.go
@@ -1560,7 +1560,7 @@ func TestMetaBackend_configuredBackendUnsetCopy(t *testing.T) {
 	}
 }
 
-// A plan that has uses the local backend
+// A plan that has uses the local backend and local state storage
 func TestMetaBackend_planLocal(t *testing.T) {
 	// Create a temporary working directory that is empty
 	td := t.TempDir()
@@ -1575,17 +1575,19 @@ func TestMetaBackend_planLocal(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	backendConfig := plans.Backend{
-		Type:      "local",
-		Config:    backendConfigRaw,
-		Workspace: "default",
+	plan := &plans.Plan{
+		Backend: &plans.Backend{
+			Type:      "local",
+			Config:    backendConfigRaw,
+			Workspace: "default",
+		},
 	}
 
 	// Setup the meta
 	m := testMetaBackend(t, nil)
 
 	// Get the backend
-	b, diags := m.BackendForLocalPlan(backendConfig)
+	b, diags := m.BackendForLocalPlan(plan)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}
@@ -1649,6 +1651,71 @@ func TestMetaBackend_planLocal(t *testing.T) {
 	}
 }
 
+// A plan that has uses the local backend and pluggable state storage
+func TestMetaBackend_planLocal_stateStore(t *testing.T) {
+	// Create a temporary working directory
+	td := t.TempDir()
+	testCopyDir(t, testFixturePath("state-store-unchanged"), td)
+	t.Chdir(td)
+
+	stateStoreConfigBlock := cty.ObjectVal(map[string]cty.Value{
+		"value": cty.StringVal("foobar"),
+	})
+	stateStoreConfigRaw, err := plans.NewDynamicValue(stateStoreConfigBlock, stateStoreConfigBlock.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	providerAddr := addrs.MustParseProviderSourceString("registry.terraform.io/hashicorp/test")
+
+	plan := &plans.Plan{
+		StateStore: &plans.StateStore{
+			Type:      "test_store",
+			Config:    stateStoreConfigRaw,
+			Workspace: backend.DefaultStateName,
+			Provider: &plans.Provider{
+				Version: version.Must(version.NewVersion("1.2.3")), // Matches lock file in the test fixtures
+				Source:  &providerAddr,
+				Config:  nil,
+			},
+		},
+	}
+
+	// Setup the meta, including a mock provider set up to mock PSS
+	m := testMetaBackend(t, nil)
+	mock := testStateStoreMockWithChunkNegotiation(t, 1000)
+	m.testingOverrides = &testingOverrides{
+		Providers: map[addrs.Provider]providers.Factory{
+			addrs.NewDefaultProvider("test"): providers.FactoryFixed(mock),
+		},
+	}
+
+	// Get the backend
+	b, diags := m.BackendForLocalPlan(plan)
+	if diags.HasErrors() {
+		t.Fatal(diags.Err())
+	}
+
+	// Check the state
+	s, sDiags := b.StateMgr(backend.DefaultStateName)
+	if sDiags.HasErrors() {
+		t.Fatalf("unexpected error: %s", sDiags.Err())
+	}
+	if err := s.RefreshState(); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	state := s.State()
+	if state != nil {
+		t.Fatalf("state should be nil: %#v", state)
+	}
+
+	// Write some state
+	state = states.NewState()
+	s.WriteState(state)
+	if err := s.PersistState(nil); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+}
+
 // A plan with a custom state save path
 func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	td := t.TempDir()
@@ -1666,10 +1733,12 @@ func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	plannedBackend := plans.Backend{
-		Type:      "local",
-		Config:    backendConfigRaw,
-		Workspace: "default",
+	plan := &plans.Plan{
+		Backend: &plans.Backend{
+			Type:      "local",
+			Config:    backendConfigRaw,
+			Workspace: "default",
+		},
 	}
 
 	// Create an alternate output path
@@ -1686,7 +1755,7 @@ func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	m.stateOutPath = statePath
 
 	// Get the backend
-	b, diags := m.BackendForLocalPlan(plannedBackend)
+	b, diags := m.BackendForLocalPlan(plan)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}
@@ -1765,17 +1834,19 @@ func TestMetaBackend_planLocalMatch(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	backendConfig := plans.Backend{
-		Type:      "local",
-		Config:    backendConfigRaw,
-		Workspace: "default",
+	plan := &plans.Plan{
+		Backend: &plans.Backend{
+			Type:      "local",
+			Config:    backendConfigRaw,
+			Workspace: "default",
+		},
 	}
 
 	// Setup the meta
 	m := testMetaBackend(t, nil)
 
 	// Get the backend
-	b, diags := m.BackendForLocalPlan(backendConfig)
+	b, diags := m.BackendForLocalPlan(plan)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}
diff --git a/internal/plans/plan.go b/internal/plans/plan.go
index 53b18c9515..0995c0f1ca 100644
--- a/internal/plans/plan.go
+++ b/internal/plans/plan.go
@@ -181,9 +181,30 @@ func (p *Plan) ProviderAddrs() []addrs.AbsProviderConfig {
 	}
 
 	m := map[string]addrs.AbsProviderConfig{}
+
+	// Get all provider requirements from resources.
 	for _, rc := range p.Changes.Resources {
 		m[rc.ProviderAddr.String()] = rc.ProviderAddr
 	}
+
+	// Get the provider required for pluggable state storage, if that's in use.
+	//
+	// This check should be redundant as:
+	// 1) Any provider used for state storage would be in required_providers, which is checked separately elsewhere.
+	// 2) An apply operation that uses the planfile only checks the providers needed for the plan _after_ the operations backend
+	//    for the operation is set up, and that process will detect if the provider needed for state storage is missing.
+	//
+	// However, for completeness when describing the providers needed by a plan, it is included here.
+	if p.StateStore != nil {
+		address := addrs.AbsProviderConfig{
+			Module:   addrs.RootModule, // A state_store block is only ever in the root module
+			Provider: *p.StateStore.Provider.Source,
+			// Alias: aliases are not permitted when using a provider for PSS.
+		}
+
+		m[p.StateStore.Provider.Source.String()] = address
+	}
+
 	if len(m) == 0 {
 		return nil
 	}
diff --git a/internal/plans/plan_test.go b/internal/plans/plan_test.go
index eca7e64d83..4b23aa7a33 100644
--- a/internal/plans/plan_test.go
+++ b/internal/plans/plan_test.go
@@ -9,12 +9,38 @@ import (
 	"github.com/go-test/deep"
 
 	"github.com/hashicorp/terraform/internal/addrs"
+	"github.com/zclconf/go-cty/cty"
 )
 
 func TestProviderAddrs(t *testing.T) {
+	// Inputs for plan
+	provider := &Provider{}
+	err := provider.SetSource("registry.terraform.io/hashicorp/pluggable")
+	if err != nil {
+		panic(err)
+	}
+	err = provider.SetVersion("9.9.9")
+	if err != nil {
+		panic(err)
+	}
+	config, err := NewDynamicValue(cty.ObjectVal(map[string]cty.Value{
+		"foo": cty.StringVal("bar"),
+	}), cty.Object(map[string]cty.Type{
+		"foo": cty.String,
+	}))
+	if err != nil {
+		panic(err)
+	}
+	provider.Config = config
 
 	// Prepare plan
 	plan := &Plan{
+		StateStore: &StateStore{
+			Type:      "pluggable_foobar",
+			Provider:  provider,
+			Config:    config,
+			Workspace: "default",
+		},
 		VariableValues: map[string]DynamicValue{},
 		Changes: &ChangesSrc{
 			Resources: []*ResourceInstanceChangeSrc{
@@ -67,6 +93,11 @@ func TestProviderAddrs(t *testing.T) {
 			Module:   addrs.RootModule,
 			Provider: addrs.NewDefaultProvider("test"),
 		},
+		// Provider used for pluggable state storage
+		{
+			Module:   addrs.RootModule,
+			Provider: addrs.NewDefaultProvider("pluggable"),
+		},
 	}
 
 	for _, problem := range deep.Equal(got, want) {