PSS: Implement initialisation of new working directory (or use of -reconfigure flag) while using state_store (#37732)

* Minor fixes in diagnostics

This can only be done once modules have been parsed and the required providers data is available. There are multiple places where config is parsed, into either Config or Module structs, so this needs to be implemented in multiple places.

* Rename test to make it specific to use of backend block in config

* Update initBackend to accept whole initArgs collection

* Only process --backend-config data, when setting up a `backend`, if that data isn't empty

* Simplify how mock provider factories are made in tests

* Update mock provider's default logic to track and manage existing workspaces

* Add `ProviderSchema` method to `Pluggable` structs. This allows calling code to access the provider schema when using provider configuration data.

* Add function for converting a providerreqs.Version to a hashicorp/go-version Version.

This is needed for using locks when creating the backend state file.

* Implement initial version of init new working directories using `stateStore_C_s`. Default to creating the default workspace if no workspaces exist.

* Update test fixtures to match the hashicorp/test mock provider used in PSS tests

* Allow tests to obtain locks that include `testingOverrides` providers.

The `testingOverrides` field will only be set in tests, so this should not impact end users.

* Add tests showing TF can initialize a working directory for the first time (and do the same when forced by -reconfigure flag). Remove replaced tests.

* Add -create-default-workspace flag, to be used to disable creating the default workspace by default when -input=false (i.e for use in CI). Refactor creation of default workspace logic. Add tests.

* Allow reattached providers to be used during init for PSS

* Rename variable to `backendHash` so relation to `backend` is clearer

* Allow `(m *Meta) Backend` to return warning diagnostics

* Protect against nil testingOverrides in providerFactoriesFromLocks

* Add test case seeing what happens if default workspace selected, doesn't exist, but other workspaces do exist.

The consequences here are due to using `selectWorkspace` in `stateStore_C_s`, matching what's done in `backend_C_r_s`.

* Address code consistency check failure on PR

* Refactor use of mock in test that's experiencing EOF error...

* Remove test that requires test to supply input for user prompt

This test passes when run in isolation but fails when run alongside other tests, even when skipping all other tests using `testStdinPipe`. I don't think the value of this test is great enough to start changing how we test stdin input.

* Allow -create-default-workspace to be used regardless of whether input is enabled or disabled

* Add TF_SKIP_CREATE_DEFAULT_WORKSPACE environment variable

* Responses to feedback, including making testStdinPipe helper log details of errors copying data to stdin.

Note: We cannot call t.Fatal from a non-test goroutine.

* Use Errorf instead

* Allow backend state files to not include version data when a builtin or reattached provider is in use.

* Add clarifying comment about re-attached providers when finding the matching entry in required_providers

* Report that the default workspace was created to the view

* Refactor: use error comparison via `errors.Is` to identify when no workspaces exist.

* Move handling of TF_ENABLE_PLUGGABLE_STATE_STORAGE into init's ParseInit func.

* Validate that PSS-related flags can only be used when experiments are enabled, enforce coupling of PSS-related flags when in use.

* Slight rewording of output message about default workspace

* Update test to assert new output about default workspace

internal/backend/pluggable/pluggable.go

@@ -68,6 +68,15 @@ func (p *Pluggable) ConfigSchema() *configschema.Block {
 	return val.Body
 }
 
+// ProviderSchema returns the schema for the provider implementing the state store.
+//
+// This isn't part of the backend.Backend interface but is needed in calling code.
+// When it's used the backend.Backend will need to be cast to a Pluggable.
+func (p *Pluggable) ProviderSchema() *configschema.Block {
+	schemaResp := p.provider.GetProviderSchema()
+	return schemaResp.Provider.Body
+}
+
 // PrepareConfig validates configuration for the state store in
 // the state storage provider. The configuration sent from Terraform core
 // will not include any values from environment variables; it is the

internal/backend/pluggable/pluggable_test.go

@@ -398,3 +398,70 @@ func TestPluggable_DeleteWorkspace(t *testing.T) {
 		t.Fatalf("expected error %q but got: %q", wantError, err)
 	}
 }
+
+func TestPluggable_ProviderSchema(t *testing.T) {
+	t.Run("Returns the expected provider schema", func(t *testing.T) {
+		mock := &testing_provider.MockProvider{
+			GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+				Provider: providers.Schema{
+					Body: &configschema.Block{
+						Attributes: map[string]*configschema.Attribute{
+							"custom_attr": {Type: cty.String, Optional: true},
+						},
+					},
+				},
+			},
+		}
+		p, err := NewPluggable(mock, "foobar")
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		// Calling code will need to case to Pluggable after using NewPluggable,
+		// so we do something similar in this test
+		var providerSchema *configschema.Block
+		if pluggable, ok := p.(*Pluggable); ok {
+			providerSchema = pluggable.ProviderSchema()
+		}
+
+		if !mock.GetProviderSchemaCalled {
+			t.Fatal("expected ProviderSchema to call the GetProviderSchema RPC")
+		}
+		if providerSchema == nil {
+			t.Fatal("ProviderSchema returned an unexpected nil schema")
+		}
+		if val := providerSchema.Attributes["custom_attr"]; val == nil {
+			t.Fatalf("expected the returned schema to include an attr called %q, but it was missing. Schema contains attrs: %v",
+				"custom_attr",
+				slices.Sorted(maps.Keys(providerSchema.Attributes)))
+		}
+	})
+
+	t.Run("Returns a nil schema when the provider has an empty schema", func(t *testing.T) {
+		mock := &testing_provider.MockProvider{
+			GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+				Provider: providers.Schema{
+					// empty schema
+				},
+			},
+		}
+		p, err := NewPluggable(mock, "foobar")
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		// Calling code will need to case to Pluggable after using NewPluggable,
+		// so we do something similar in this test
+		var providerSchema *configschema.Block
+		if pluggable, ok := p.(*Pluggable); ok {
+			providerSchema = pluggable.ProviderSchema()
+		}
+
+		if !mock.GetProviderSchemaCalled {
+			t.Fatal("expected ProviderSchema to call the GetProviderSchema RPC")
+		}
+		if providerSchema != nil {
+			t.Fatalf("expected ProviderSchema to return a nil schema but got: %#v", providerSchema)
+		}
+	})
+}

internal/command/arguments/init.go

@@ -4,6 +4,7 @@
 package arguments
 
 import (
+	"os"
 	"time"
 
 	"github.com/hashicorp/terraform/internal/tfdiags"
@@ -78,12 +79,16 @@ type Init struct {
 	// TODO(SarahFrench/radeksimko): Remove this once the feature is no longer
 	// experimental
 	EnablePssExperiment bool
+
+	// CreateDefaultWorkspace indicates whether the default workspace should be created by
+	// Terraform when initializing a state store for the first time.
+	CreateDefaultWorkspace bool
 }
 
 // ParseInit processes CLI arguments, returning an Init value and errors.
 // If errors are encountered, an Init value is still returned representing
 // the best effort interpretation of the arguments.
-func ParseInit(args []string) (*Init, tfdiags.Diagnostics) {
+func ParseInit(args []string, experimentsEnabled bool) (*Init, tfdiags.Diagnostics) {
 	var diags tfdiags.Diagnostics
 	init := &Init{
 		Vars: &Vars{},
@@ -111,6 +116,7 @@ func ParseInit(args []string) (*Init, tfdiags.Diagnostics) {
 	cmdFlags.BoolVar(&init.Json, "json", false, "json")
 	cmdFlags.Var(&init.BackendConfig, "backend-config", "")
 	cmdFlags.Var(&init.PluginPath, "plugin-dir", "plugin directory")
+	cmdFlags.BoolVar(&init.CreateDefaultWorkspace, "create-default-workspace", true, "when -input=false, use this flag to block creation of the default workspace")
 
 	// Used for enabling experimental code that's invoked before configuration is parsed.
 	cmdFlags.BoolVar(&init.EnablePssExperiment, "enable-pluggable-state-storage-experiment", false, "Enable the pluggable state storage experiment")
@@ -123,6 +129,46 @@ func ParseInit(args []string) (*Init, tfdiags.Diagnostics) {
 		))
 	}
 
+	if v := os.Getenv("TF_ENABLE_PLUGGABLE_STATE_STORAGE"); v != "" {
+		init.EnablePssExperiment = true
+	}
+
+	if v := os.Getenv("TF_SKIP_CREATE_DEFAULT_WORKSPACE"); v != "" {
+		// If TF_SKIP_CREATE_DEFAULT_WORKSPACE is set it will override
+		// a -create-default-workspace=true flag that's set explicitly,
+		// as that's indistinguishable from the default value being used.
+		init.CreateDefaultWorkspace = false
+	}
+
+	if !experimentsEnabled {
+		// If experiments aren't enabled then these flags should not be used.
+		if init.EnablePssExperiment {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Cannot use -enable-pluggable-state-storage-experiment flag without experiments enabled",
+				"Terraform cannot use the-enable-pluggable-state-storage-experiment flag (or TF_ENABLE_PLUGGABLE_STATE_STORAGE environment variable) unless experiments are enabled.",
+			))
+		}
+		if !init.CreateDefaultWorkspace {
+			// Can only be set to false by using the flag
+			// and we cannot identify if -create-default-workspace=true is set explicitly.
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Cannot use -create-default-workspace flag without experiments enabled",
+				"Terraform cannot use the -create-default-workspace flag (or TF_SKIP_CREATE_DEFAULT_WORKSPACE environment variable) unless experiments are enabled.",
+			))
+		}
+	} else {
+		// Errors using flags despite experiments being enabled.
+		if !init.CreateDefaultWorkspace && !init.EnablePssExperiment {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Cannot use -create-default-workspace=false flag unless the pluggable state storage experiment is enabled",
+				"Terraform cannot use the -create-default-workspace=false flag (or TF_SKIP_CREATE_DEFAULT_WORKSPACE environment variable) unless you also supply the -enable-pluggable-state-storage-experiment flag (or set the TF_ENABLE_PLUGGABLE_STATE_STORAGE environment variable).",
+			))
+		}
+	}
+
 	if init.MigrateState && init.Json {
 		diags = diags.Append(tfdiags.Sourceless(
 			tfdiags.Error,

internal/command/arguments/init_test.go

@@ -40,10 +40,11 @@ func TestParseInit_basicValid(t *testing.T) {
 					FlagName: "-backend-config",
 					Items:    &flagNameValue,
 				},
-				Vars:            &Vars{},
-				InputEnabled:    true,
-				CompactWarnings: false,
-				TargetFlags:     nil,
+				Vars:                   &Vars{},
+				InputEnabled:           true,
+				CompactWarnings:        false,
+				TargetFlags:            nil,
+				CreateDefaultWorkspace: true,
 			},
 		},
 		"setting multiple options": {
@@ -72,11 +73,12 @@ func TestParseInit_basicValid(t *testing.T) {
 					FlagName: "-backend-config",
 					Items:    &flagNameValue,
 				},
-				Vars:            &Vars{},
-				InputEnabled:    true,
-				Args:            []string{},
-				CompactWarnings: true,
-				TargetFlags:     nil,
+				Vars:                   &Vars{},
+				InputEnabled:           true,
+				Args:                   []string{},
+				CompactWarnings:        true,
+				TargetFlags:            nil,
+				CreateDefaultWorkspace: true,
 			},
 		},
 		"with cloud option": {
@@ -101,11 +103,12 @@ func TestParseInit_basicValid(t *testing.T) {
 					FlagName: "-backend-config",
 					Items:    &[]FlagNameValue{{Name: "-backend-config", Value: "backend.config"}},
 				},
-				Vars:            &Vars{},
-				InputEnabled:    false,
-				Args:            []string{},
-				CompactWarnings: false,
-				TargetFlags:     []string{"foo_bar.baz"},
+				Vars:                   &Vars{},
+				InputEnabled:           false,
+				Args:                   []string{},
+				CompactWarnings:        false,
+				TargetFlags:            []string{"foo_bar.baz"},
+				CreateDefaultWorkspace: true,
 			},
 		},
 	}
@@ -114,7 +117,8 @@ func TestParseInit_basicValid(t *testing.T) {
 
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
-			got, diags := ParseInit(tc.args)
+			experimentsEnabled := false
+			got, diags := ParseInit(tc.args, experimentsEnabled)
 			if len(diags) > 0 {
 				t.Fatalf("unexpected diags: %v", diags)
 			}
@@ -156,7 +160,8 @@ func TestParseInit_invalid(t *testing.T) {
 
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
-			got, diags := ParseInit(tc.args)
+			experimentsEnabled := false
+			got, diags := ParseInit(tc.args, experimentsEnabled)
 			if len(diags) == 0 {
 				t.Fatal("expected diags but got none")
 			}
@@ -170,6 +175,68 @@ func TestParseInit_invalid(t *testing.T) {
 	}
 }
 
+func TestParseInit_experimentalFlags(t *testing.T) {
+	testCases := map[string]struct {
+		args               []string
+		envs               map[string]string
+		wantErr            string
+		experimentsEnabled bool
+	}{
+		"error: -enable-pluggable-state-storage-experiment and experiments are disabled": {
+			args:               []string{"-enable-pluggable-state-storage-experiment"},
+			experimentsEnabled: false,
+			wantErr:            "Cannot use -enable-pluggable-state-storage-experiment flag without experiments enabled",
+		},
+		"error: TF_ENABLE_PLUGGABLE_STATE_STORAGE is set and experiments are disabled": {
+			envs: map[string]string{
+				"TF_ENABLE_PLUGGABLE_STATE_STORAGE": "1",
+			},
+			experimentsEnabled: false,
+			wantErr:            "Cannot use -enable-pluggable-state-storage-experiment flag without experiments enabled",
+		},
+		"error: -create-default-workspace=false and experiments are disabled": {
+			args:               []string{"-create-default-workspace=false"},
+			experimentsEnabled: false,
+			wantErr:            "Cannot use -create-default-workspace flag without experiments enabled",
+		},
+		"error: TF_SKIP_CREATE_DEFAULT_WORKSPACE is set and experiments are disabled": {
+			envs: map[string]string{
+				"TF_SKIP_CREATE_DEFAULT_WORKSPACE": "1",
+			},
+			experimentsEnabled: false,
+			wantErr:            "Cannot use -create-default-workspace flag without experiments enabled",
+		},
+		"error: -create-default-workspace=false used without -enable-pluggable-state-storage-experiment, while experiments are enabled": {
+			args:               []string{"-create-default-workspace=false"},
+			experimentsEnabled: true,
+			wantErr:            "Cannot use -create-default-workspace=false flag unless the pluggable state storage experiment is enabled",
+		},
+		"error: TF_SKIP_CREATE_DEFAULT_WORKSPACE used without -enable-pluggable-state-storage-experiment, while experiments are enabled": {
+			envs: map[string]string{
+				"TF_SKIP_CREATE_DEFAULT_WORKSPACE": "1",
+			},
+			experimentsEnabled: true,
+			wantErr:            "Cannot use -create-default-workspace=false flag unless the pluggable state storage experiment is enabled",
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			for k, v := range tc.envs {
+				t.Setenv(k, v)
+			}
+
+			_, diags := ParseInit(tc.args, tc.experimentsEnabled)
+			if len(diags) == 0 {
+				t.Fatal("expected diags but got none")
+			}
+			if got, want := diags.Err().Error(), tc.wantErr; !strings.Contains(got, want) {
+				t.Fatalf("wrong diags\n got: %s\nwant: %s", got, want)
+			}
+		})
+	}
+}
+
 func TestParseInit_vars(t *testing.T) {
 	testCases := map[string]struct {
 		args []string
@@ -207,7 +274,8 @@ func TestParseInit_vars(t *testing.T) {
 
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
-			got, diags := ParseInit(tc.args)
+			experimentsEnabled := false
+			got, diags := ParseInit(tc.args, experimentsEnabled)
 			if len(diags) > 0 {
 				t.Fatalf("unexpected diags: %v", diags)
 			}

internal/command/command_test.go

@@ -657,7 +657,10 @@ func testStdinPipe(t *testing.T, src io.Reader) func() {
 	// Copy the data from the reader to the pipe
 	go func() {
 		defer w.Close()
-		io.Copy(w, src)
+		_, err := io.Copy(w, src)
+		if err != nil {
+			t.Errorf("error when copying data from testStdinPipe reader argument to stdin: %s", err)
+		}
 	}()
 
 	return func() {

internal/command/init.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"log"
 	"maps"
-	"os"
 	"reflect"
 	"slices"
 	"sort"
@@ -49,7 +48,7 @@ type InitCommand struct {
 func (c *InitCommand) Run(args []string) int {
 	var diags tfdiags.Diagnostics
 	args = c.Meta.process(args)
-	initArgs, initDiags := arguments.ParseInit(args)
+	initArgs, initDiags := arguments.ParseInit(args, c.Meta.AllowExperimentalFeatures)
 
 	view := views.NewInit(initArgs.ViewType, c.View)
 
@@ -64,9 +63,6 @@ func (c *InitCommand) Run(args []string) int {
 	// 	> The user uses an experimental version of TF (alpha or built from source)
 	//  > Either the flag -enable-pluggable-state-storage-experiment is passed to the init command.
 	//  > Or, the environment variable TF_ENABLE_PLUGGABLE_STATE_STORAGE is set to any value.
-	if v := os.Getenv("TF_ENABLE_PLUGGABLE_STATE_STORAGE"); v != "" {
-		initArgs.EnablePssExperiment = true
-	}
 	if c.Meta.AllowExperimentalFeatures && initArgs.EnablePssExperiment {
 		// TODO(SarahFrench/radeksimko): Remove forked init logic once feature is no longer experimental
 		return c.runPssInit(initArgs, view)
@@ -159,7 +155,7 @@ func (c *InitCommand) initCloud(ctx context.Context, root *configs.Module, extra
 	return back, true, diags
 }
 
-func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, extraConfig arguments.FlagNameValueSlice, viewType arguments.ViewType, configLocks *depsfile.Locks, view views.Init) (be backend.Backend, output bool, diags tfdiags.Diagnostics) {
+func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, initArgs *arguments.Init, configLocks *depsfile.Locks, view views.Init) (be backend.Backend, output bool, diags tfdiags.Diagnostics) {
 	ctx, span := tracer.Start(ctx, "initialize backend")
 	_ = ctx // prevent staticcheck from complaining to avoid a maintenance hazard of having the wrong ctx in scope here
 	defer span.End()
@@ -195,7 +191,7 @@ func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, ext
 
 		// If overrides supplied by -backend-config CLI flag, process them
 		var configOverride hcl.Body
-		if !extraConfig.Empty() {
+		if !initArgs.BackendConfig.Empty() {
 			// We need to launch an instance of the provider to get the config of the state store for processing any overrides.
 			provider, err := factory()
 			defer provider.Close() // Stop the child process once we're done with it here.
@@ -238,7 +234,7 @@ func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, ext
 
 			// Handle any overrides supplied via -backend-config CLI flags
 			var overrideDiags tfdiags.Diagnostics
-			configOverride, overrideDiags = c.backendConfigOverrideBody(extraConfig, stateStoreSchema.Body)
+			configOverride, overrideDiags = c.backendConfigOverrideBody(initArgs.BackendConfig, stateStoreSchema.Body)
 			diags = diags.Append(overrideDiags)
 			if overrideDiags.HasErrors() {
 				return nil, true, diags
@@ -246,11 +242,13 @@ func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, ext
 		}
 
 		opts = &BackendOpts{
-			StateStoreConfig: root.StateStore,
-			ProviderFactory:  factory,
-			ConfigOverride:   configOverride,
-			Init:             true,
-			ViewType:         viewType,
+			StateStoreConfig:       root.StateStore,
+			Locks:                  configLocks,
+			ProviderFactory:        factory,
+			CreateDefaultWorkspace: initArgs.CreateDefaultWorkspace,
+			ConfigOverride:         configOverride,
+			Init:                   true,
+			ViewType:               initArgs.ViewType,
 		}
 
 	case root.Backend != nil:
@@ -286,17 +284,22 @@ func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, ext
 		backendSchema := b.ConfigSchema()
 		backendConfig := root.Backend
 
-		backendConfigOverride, overrideDiags := c.backendConfigOverrideBody(extraConfig, backendSchema)
-		diags = diags.Append(overrideDiags)
-		if overrideDiags.HasErrors() {
-			return nil, true, diags
+		// If overrides supplied by -backend-config CLI flag, process them
+		var configOverride hcl.Body
+		if !initArgs.BackendConfig.Empty() {
+			var overrideDiags tfdiags.Diagnostics
+			configOverride, overrideDiags = c.backendConfigOverrideBody(initArgs.BackendConfig, backendSchema)
+			diags = diags.Append(overrideDiags)
+			if overrideDiags.HasErrors() {
+				return nil, true, diags
+			}
 		}
 
 		opts = &BackendOpts{
 			BackendConfig:  backendConfig,
-			ConfigOverride: backendConfigOverride,
+			ConfigOverride: configOverride,
 			Init:           true,
-			ViewType:       viewType,
+			ViewType:       initArgs.ViewType,
 		}
 
 	default:
@@ -305,7 +308,7 @@ func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, ext
 		// If the user supplied a -backend-config on the CLI but no backend
 		// block was found in the configuration, it's likely - but not
 		// necessarily - a mistake. Return a warning.
-		if !extraConfig.Empty() {
+		if !initArgs.BackendConfig.Empty() {
 			diags = diags.Append(tfdiags.Sourceless(
 				tfdiags.Warning,
 				"Missing backend configuration",
@@ -328,7 +331,7 @@ the backend configuration is present and valid.
 
 		opts = &BackendOpts{
 			Init:     true,
-			ViewType: viewType,
+			ViewType: initArgs.ViewType,
 		}
 	}
 

internal/command/init_run.go

@@ -174,7 +174,7 @@ func (c *InitCommand) run(initArgs *arguments.Init, view views.Init) int {
 		// initBackend has new parameters that aren't relevant to the original (unpluggable) version of the init command logic here.
 		// So for this version of the init command, we pass in empty locks intentionally.
 		emptyLocks := depsfile.NewLocks()
-		back, backendOutput, backDiags = c.initBackend(ctx, rootModEarly, initArgs.BackendConfig, initArgs.ViewType, emptyLocks, view)
+		back, backendOutput, backDiags = c.initBackend(ctx, rootModEarly, initArgs, emptyLocks, view)
 	default:
 		// load the previously-stored backend config
 		back, backDiags = c.Meta.backendFromState(ctx)

internal/command/init_run_experiment.go

@@ -205,7 +205,7 @@ func (c *InitCommand) runPssInit(initArgs *arguments.Init, view views.Init) int
 	case initArgs.Cloud && rootModEarly.CloudConfig != nil:
 		back, backendOutput, backDiags = c.initCloud(ctx, rootModEarly, initArgs.BackendConfig, initArgs.ViewType, view)
 	case initArgs.Backend:
-		back, backendOutput, backDiags = c.initBackend(ctx, rootModEarly, initArgs.BackendConfig, initArgs.ViewType, configLocks, view)
+		back, backendOutput, backDiags = c.initBackend(ctx, rootModEarly, initArgs, configLocks, view)
 	default:
 		// load the previously-stored backend config
 		back, backDiags = c.Meta.backendFromState(ctx)

internal/command/init_test.go

@@ -17,11 +17,15 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"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"
+	"github.com/hashicorp/terraform/internal/backend"
 	"github.com/hashicorp/terraform/internal/command/arguments"
+	"github.com/hashicorp/terraform/internal/command/clistate"
 	"github.com/hashicorp/terraform/internal/command/views"
+	"github.com/hashicorp/terraform/internal/command/workdir"
 	"github.com/hashicorp/terraform/internal/configs"
 	"github.com/hashicorp/terraform/internal/configs/configschema"
 	"github.com/hashicorp/terraform/internal/depsfile"
@@ -3227,6 +3231,431 @@ func TestInit_testsWithModule(t *testing.T) {
 	}
 }
 
+// Testing init's behaviors with `state_store` when run in an empty working directory
+func TestInit_stateStore_newWorkingDir(t *testing.T) {
+	t.Run("the init command creates a backend state file, and creates the default workspace by default", func(t *testing.T) {
+		// Create a temporary, uninitialized working directory with configuration including a state store
+		td := t.TempDir()
+		testCopyDir(t, testFixturePath("init-with-state-store"), td)
+		t.Chdir(td)
+
+		mockProvider := mockPluggableStateStorageProvider()
+		mockProviderAddress := addrs.NewDefaultProvider("test")
+		providerSource, close := newMockProviderSource(t, map[string][]string{
+			"hashicorp/test": {"1.0.0"},
+		})
+		defer close()
+
+		ui := new(cli.MockUi)
+		view, done := testView(t)
+		meta := Meta{
+			Ui:                        ui,
+			View:                      view,
+			AllowExperimentalFeatures: true,
+			testingOverrides: &testingOverrides{
+				Providers: map[addrs.Provider]providers.Factory{
+					mockProviderAddress: providers.FactoryFixed(mockProvider),
+				},
+			},
+			ProviderSource: providerSource,
+		}
+		c := &InitCommand{
+			Meta: meta,
+		}
+
+		args := []string{"-enable-pluggable-state-storage-experiment=true"}
+		code := c.Run(args)
+		testOutput := done(t)
+		if code != 0 {
+			t.Fatalf("expected code 0 exit code, got %d, output: \n%s", code, testOutput.All())
+		}
+
+		// Check output
+		output := testOutput.All()
+		expectedOutputs := []string{
+			"Initializing the state store...",
+			"Terraform created an empty state file for the default workspace",
+			"Terraform has been successfully initialized!",
+		}
+		for _, expected := range expectedOutputs {
+			if !strings.Contains(output, expected) {
+				t.Fatalf("expected output to include %q, but got':\n %s", expected, output)
+			}
+		}
+
+		// Assert the default workspace was created
+		if _, exists := mockProvider.MockStates[backend.DefaultStateName]; !exists {
+			t.Fatal("expected the default workspace to be created during init, but it is missing")
+		}
+
+		// Assert contents of the backend state file
+		statePath := filepath.Join(meta.DataDir(), DefaultStateFilename)
+		sMgr := &clistate.LocalState{Path: statePath}
+		if err := sMgr.RefreshState(); err != nil {
+			t.Fatal("Failed to load state:", err)
+		}
+		s := sMgr.State()
+		if s == nil {
+			t.Fatal("expected backend state file to be created, but there isn't one")
+		}
+		v1_0_0, _ := version.NewVersion("1.0.0")
+		expectedState := &workdir.StateStoreConfigState{
+			Type:      "test_store",
+			ConfigRaw: []byte("{\n      \"value\": \"foobar\"\n    }"),
+			Hash:      uint64(2116468040), // Hash affected by config
+			Provider: &workdir.ProviderConfigState{
+				Version: v1_0_0,
+				Source: &tfaddr.Provider{
+					Hostname:  tfaddr.DefaultProviderRegistryHost,
+					Namespace: "hashicorp",
+					Type:      "test",
+				},
+				ConfigRaw: []byte("{\n        \"region\": null\n      }"),
+				Hash:      uint64(3976463117), // Hash of empty config
+			},
+		}
+		if diff := cmp.Diff(s.StateStore, expectedState); diff != "" {
+			t.Fatalf("unexpected diff in backend state file's description of state store:\n%s", diff)
+		}
+	})
+
+	t.Run("an init command with the flag -create-default-workspace=false will not make the default workspace by default", func(t *testing.T) {
+		// Create a temporary, uninitialized working directory with configuration including a state store
+		td := t.TempDir()
+		testCopyDir(t, testFixturePath("init-with-state-store"), td)
+		t.Chdir(td)
+
+		mockProvider := mockPluggableStateStorageProvider()
+		mockProviderAddress := addrs.NewDefaultProvider("test")
+		providerSource, close := newMockProviderSource(t, map[string][]string{
+			"hashicorp/test": {"1.0.0"},
+		})
+		defer close()
+
+		ui := new(cli.MockUi)
+		view, done := testView(t)
+		c := &InitCommand{
+			Meta: Meta{
+				Ui:                        ui,
+				View:                      view,
+				AllowExperimentalFeatures: true,
+				testingOverrides: &testingOverrides{
+					Providers: map[addrs.Provider]providers.Factory{
+						mockProviderAddress: providers.FactoryFixed(mockProvider),
+					},
+				},
+				ProviderSource: providerSource,
+			},
+		}
+
+		args := []string{"-enable-pluggable-state-storage-experiment=true", "-create-default-workspace=false"}
+		code := c.Run(args)
+		testOutput := done(t)
+		if code != 0 {
+			t.Fatalf("expected code 0 exit code, got %d, output: \n%s", code, testOutput.All())
+		}
+
+		// Check output
+		output := testOutput.All()
+		expectedOutput := `Terraform has been configured to skip creation of the default workspace`
+		if !strings.Contains(output, expectedOutput) {
+			t.Fatalf("expected output to include %q, but got':\n %s", expectedOutput, output)
+		}
+
+		// Assert the default workspace was created
+		if _, exists := mockProvider.MockStates[backend.DefaultStateName]; exists {
+			t.Fatal("expected Terraform to skip creating the default workspace, but it has been created")
+		}
+	})
+
+	t.Run("an init command with TF_SKIP_CREATE_DEFAULT_WORKSPACE set will not make the default workspace by default", func(t *testing.T) {
+		// Create a temporary, uninitialized working directory with configuration including a state store
+		td := t.TempDir()
+		testCopyDir(t, testFixturePath("init-with-state-store"), td)
+		t.Chdir(td)
+
+		mockProvider := mockPluggableStateStorageProvider()
+		mockProviderAddress := addrs.NewDefaultProvider("test")
+		providerSource, close := newMockProviderSource(t, map[string][]string{
+			"hashicorp/test": {"1.0.0"},
+		})
+		defer close()
+
+		ui := new(cli.MockUi)
+		view, done := testView(t)
+		c := &InitCommand{
+			Meta: Meta{
+				Ui:                        ui,
+				View:                      view,
+				AllowExperimentalFeatures: true,
+				testingOverrides: &testingOverrides{
+					Providers: map[addrs.Provider]providers.Factory{
+						mockProviderAddress: providers.FactoryFixed(mockProvider),
+					},
+				},
+				ProviderSource: providerSource,
+			},
+		}
+
+		t.Setenv("TF_SKIP_CREATE_DEFAULT_WORKSPACE", "1") // any value
+		args := []string{"-enable-pluggable-state-storage-experiment=true"}
+		code := c.Run(args)
+		testOutput := done(t)
+		if code != 0 {
+			t.Fatalf("expected code 0 exit code, got %d, output: \n%s", code, testOutput.All())
+		}
+
+		// Check output
+		output := testOutput.All()
+		expectedOutput := `Terraform has been configured to skip creation of the default workspace`
+		if !strings.Contains(output, expectedOutput) {
+			t.Fatalf("expected output to include %q, but got':\n %s", expectedOutput, output)
+		}
+
+		// Assert the default workspace was created
+		if _, exists := mockProvider.MockStates[backend.DefaultStateName]; exists {
+			t.Fatal("expected Terraform to skip creating the default workspace, but it has been created")
+		}
+	})
+
+	// This scenario would be rare, but protecting against it is easy and avoids assumptions.
+	t.Run("if a custom workspace is selected but no workspaces exist an error is returned", func(t *testing.T) {
+		// Create a temporary, uninitialized working directory with configuration including a state store
+		td := t.TempDir()
+		testCopyDir(t, testFixturePath("init-with-state-store"), td)
+		t.Chdir(td)
+
+		// Select a custom workspace (which will not exist)
+		customWorkspace := "my-custom-workspace"
+		t.Setenv(WorkspaceNameEnvVar, customWorkspace)
+
+		mockProvider := mockPluggableStateStorageProvider()
+		mockProviderAddress := addrs.NewDefaultProvider("test")
+		providerSource, close := newMockProviderSource(t, map[string][]string{
+			"hashicorp/test": {"1.0.0"},
+		})
+		defer close()
+
+		ui := new(cli.MockUi)
+		view, done := testView(t)
+		meta := Meta{
+			Ui:                        ui,
+			View:                      view,
+			AllowExperimentalFeatures: true,
+			testingOverrides: &testingOverrides{
+				Providers: map[addrs.Provider]providers.Factory{
+					mockProviderAddress: providers.FactoryFixed(mockProvider),
+				},
+			},
+			ProviderSource: providerSource,
+		}
+		c := &InitCommand{
+			Meta: meta,
+		}
+
+		args := []string{"-enable-pluggable-state-storage-experiment=true"}
+		code := c.Run(args)
+		testOutput := done(t)
+		if code != 1 {
+			t.Fatalf("expected code 1 exit code, got %d, output: \n%s", code, testOutput.All())
+		}
+
+		// Check output
+		output := testOutput.All()
+		expectedOutputs := []string{
+			fmt.Sprintf("Workspace %q has not been created yet", customWorkspace),
+			fmt.Sprintf("To create the custom workspace %q use the command `terraform workspace new %s`", customWorkspace, customWorkspace),
+		}
+		for _, expected := range expectedOutputs {
+			if !strings.Contains(cleanString(output), expected) {
+				t.Fatalf("expected output to include %q, but got':\n %s", expected, cleanString(output))
+			}
+		}
+
+		// Assert no workspaces exist
+		if len(mockProvider.MockStates) != 0 {
+			t.Fatalf("expected no workspaces, but got: %#v", mockProvider.MockStates)
+		}
+
+		// Assert no backend state file made due to the error
+		statePath := filepath.Join(meta.DataDir(), DefaultStateFilename)
+		_, err := os.Stat(statePath)
+		if pathErr, ok := err.(*os.PathError); !ok || !os.IsNotExist(pathErr.Err) {
+			t.Fatalf("expected backend state file to not be created, but it exists")
+		}
+	})
+
+	// Tests outcome when input enabled and disabled
+	t.Run("if the default workspace is selected and doesn't exist, but other custom workspaces do exist and input is disabled, an error is returned", func(t *testing.T) {
+		// Create a temporary, uninitialized working directory with configuration including a state store
+		td := t.TempDir()
+		testCopyDir(t, testFixturePath("init-with-state-store"), td)
+		t.Chdir(td)
+
+		mockProvider := mockPluggableStateStorageProvider()
+		mockProvider.GetStatesResponse = &providers.GetStatesResponse{
+			States: []string{
+				"foobar1",
+				"foobar2",
+				// Force provider to report workspaces exist
+				// But default workspace doesn't exist
+			},
+		}
+
+		mockProviderAddress := addrs.NewDefaultProvider("test")
+		providerSource, close := newMockProviderSource(t, map[string][]string{
+			"hashicorp/test": {"1.0.0"},
+		})
+		defer close()
+
+		ui := new(cli.MockUi)
+		view, done := testView(t)
+		meta := Meta{
+			Ui:                        ui,
+			View:                      view,
+			AllowExperimentalFeatures: true,
+			testingOverrides: &testingOverrides{
+				Providers: map[addrs.Provider]providers.Factory{
+					mockProviderAddress: providers.FactoryFixed(mockProvider),
+				},
+			},
+			ProviderSource: providerSource,
+		}
+		c := &InitCommand{
+			Meta: meta,
+		}
+
+		// If input is disabled users receive an error about the missing workspace
+		args := []string{
+			"-enable-pluggable-state-storage-experiment=true",
+			"-input=false",
+		}
+		code := c.Run(args)
+		testOutput := done(t)
+		if code != 1 {
+			t.Fatalf("expected code 1 exit code, got %d, output: \n%s", code, testOutput.All())
+		}
+		output := testOutput.All()
+		expectedOutput := "Failed to select a workspace: Currently selected workspace \"default\" does not exist"
+		if !strings.Contains(cleanString(output), expectedOutput) {
+			t.Fatalf("expected output to include %q, but got':\n %s", expectedOutput, cleanString(output))
+		}
+		statePath := filepath.Join(meta.DataDir(), DefaultStateFilename)
+		_, err := os.Stat(statePath)
+		if _, ok := err.(*os.PathError); !ok {
+			if err == nil {
+				t.Fatalf("expected backend state file to not be created, but it exists")
+			}
+
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	// TODO(SarahFrench/radeksimko): Add test cases below:
+	// 1) "during a non-init command, the command ends in with an error telling the user to run an init command"
+	// >>> Currently this is handled at a lower level in `internal/command/meta_backend_test.go`
+}
+
+// Testing init's behaviors with `state_store` when run in a working directory where the configuration
+// doesn't match the backend state file.
+func TestInit_stateStore_configChanges(t *testing.T) {
+	t.Run("the -reconfigure flag makes Terraform ignore the backend state file during initialization", func(t *testing.T) {
+		// Create a temporary working directory with state store configuration
+		// that doesn't match the backend state file
+		td := t.TempDir()
+		testCopyDir(t, testFixturePath("state-store-reconfigure"), td)
+		t.Chdir(td)
+
+		mockProvider := mockPluggableStateStorageProvider()
+		mockProviderAddress := addrs.NewDefaultProvider("test")
+		providerSource, close := newMockProviderSource(t, map[string][]string{
+			"hashicorp/test": {"1.0.0"},
+		})
+		defer close()
+
+		ui := new(cli.MockUi)
+		view, done := testView(t)
+		meta := Meta{
+			Ui:                        ui,
+			View:                      view,
+			AllowExperimentalFeatures: true,
+			testingOverrides: &testingOverrides{
+				Providers: map[addrs.Provider]providers.Factory{
+					mockProviderAddress: providers.FactoryFixed(mockProvider),
+				},
+			},
+			ProviderSource: providerSource,
+		}
+		c := &InitCommand{
+			Meta: meta,
+		}
+
+		args := []string{
+			"-enable-pluggable-state-storage-experiment=true",
+			"-reconfigure",
+		}
+		code := c.Run(args)
+		testOutput := done(t)
+		if code != 0 {
+			t.Fatalf("expected code 0 exit code, got %d, output: \n%s", code, testOutput.All())
+		}
+
+		// Check output
+		output := testOutput.All()
+		expectedOutputs := []string{
+			"Initializing the state store...",
+			"Terraform has been successfully initialized!",
+		}
+		for _, expected := range expectedOutputs {
+			if !strings.Contains(output, expected) {
+				t.Fatalf("expected output to include %q, but got':\n %s", expected, output)
+			}
+		}
+
+		// Assert the default workspace was created
+		if _, exists := mockProvider.MockStates[backend.DefaultStateName]; !exists {
+			t.Fatal("expected the default workspace to be created during init, but it is missing")
+		}
+
+		// Assert contents of the backend state file
+		statePath := filepath.Join(meta.DataDir(), DefaultStateFilename)
+		sMgr := &clistate.LocalState{Path: statePath}
+		if err := sMgr.RefreshState(); err != nil {
+			t.Fatal("Failed to load state:", err)
+		}
+		s := sMgr.State()
+		if s == nil {
+			t.Fatal("expected backend state file to be created, but there isn't one")
+		}
+		v1_0_0, _ := version.NewVersion("1.0.0")
+		expectedState := &workdir.StateStoreConfigState{
+			Type:      "test_store",
+			ConfigRaw: []byte("{\n      \"value\": \"changed-value\"\n    }"),
+			Hash:      uint64(1417640992), // Hash affected by config
+			Provider: &workdir.ProviderConfigState{
+				Version: v1_0_0,
+				Source: &tfaddr.Provider{
+					Hostname:  tfaddr.DefaultProviderRegistryHost,
+					Namespace: "hashicorp",
+					Type:      "test",
+				},
+				ConfigRaw: []byte("{\n        \"region\": null\n      }"),
+				Hash:      uint64(3976463117), // Hash of empty config
+			},
+		}
+		if diff := cmp.Diff(s.StateStore, expectedState); diff != "" {
+			t.Fatalf("unexpected diff in backend state file's description of state store:\n%s", diff)
+		}
+	})
+
+	// TODO(SarahFrench/radeksimko): Add more test cases related to changing the
+	// configuration and the forced need for state migration.
+	// More complicated situations might benefit from being separate tests altogether.
+	// Simpler scenarios that make sense to keep here are:
+	// 1) Changing config of the same state_store type
+	// 2) Changing config of the same provider (and version) used for PSS
+}
+
 // newMockProviderSource is a helper to succinctly construct a mock provider
 // source that contains a set of packages matching the given provider versions
 // that are available for installation (from temporary local files).
@@ -3367,3 +3796,72 @@ func expectedPackageInstallPath(name, version string, exe bool) string {
 		baseDir, fmt.Sprintf("registry.terraform.io/hashicorp/%s/%s/%s", name, version, platform),
 	))
 }
+
+func mockPluggableStateStorageProvider() *testing_provider.MockProvider {
+	// Create a mock provider to use for PSS
+	// Get mock provider factory to be used during init
+	//
+	// This imagines a provider called `test` that contains
+	// a pluggable state store implementation called `store`.
+	pssName := "test_store"
+	mock := testing_provider.MockProvider{
+		GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+			Provider: providers.Schema{
+				Body: &configschema.Block{
+					Attributes: map[string]*configschema.Attribute{
+						"region": {Type: cty.String, Optional: true},
+					},
+				},
+			},
+			DataSources:       map[string]providers.Schema{},
+			ResourceTypes:     map[string]providers.Schema{},
+			ListResourceTypes: map[string]providers.Schema{},
+			StateStores: map[string]providers.Schema{
+				pssName: {
+					Body: &configschema.Block{
+						Attributes: map[string]*configschema.Attribute{
+							"value": {
+								Type:     cty.String,
+								Required: true,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	mock.ConfigureStateStoreFn = func(req providers.ConfigureStateStoreRequest) providers.ConfigureStateStoreResponse {
+		return providers.ConfigureStateStoreResponse{
+			Capabilities: providers.StateStoreServerCapabilities{
+				ChunkSize: 1234, // arbitrary number that isn't 0
+			},
+		}
+	}
+	mock.WriteStateBytesFn = func(req providers.WriteStateBytesRequest) providers.WriteStateBytesResponse {
+		// Workspaces exist once the artefact representing it is written
+		if _, exist := mock.MockStates[req.StateId]; !exist {
+			// Ensure non-nil map
+			if mock.MockStates == nil {
+				mock.MockStates = make(map[string]interface{})
+			}
+
+			mock.MockStates[req.StateId] = req.Bytes
+		}
+		return providers.WriteStateBytesResponse{
+			Diagnostics: nil, // success
+		}
+	}
+	mock.ReadStateBytesFn = func(req providers.ReadStateBytesRequest) providers.ReadStateBytesResponse {
+		state := []byte{}
+		if v, exist := mock.MockStates[req.StateId]; exist {
+			if s, ok := v.([]byte); ok {
+				state = s
+			}
+		}
+		return providers.ReadStateBytesResponse{
+			Bytes:       state,
+			Diagnostics: nil, // success
+		}
+	}
+	return &mock
+}

internal/command/meta_backend.go

@@ -14,15 +14,18 @@ import (
 	"fmt"
 	"log"
 	"maps"
+	"os"
 	"path/filepath"
 	"slices"
 	"strconv"
 	"strings"
 
+	version "github.com/hashicorp/go-version"
 	"github.com/hashicorp/hcl/v2"
 	"github.com/hashicorp/hcl/v2/hcldec"
 	"github.com/zclconf/go-cty/cty"
 
+	"github.com/hashicorp/terraform/internal/addrs"
 	"github.com/hashicorp/terraform/internal/backend"
 	"github.com/hashicorp/terraform/internal/backend/backendrun"
 	backendInit "github.com/hashicorp/terraform/internal/backend/init"
@@ -36,8 +39,11 @@ import (
 	"github.com/hashicorp/terraform/internal/configs"
 	"github.com/hashicorp/terraform/internal/depsfile"
 	"github.com/hashicorp/terraform/internal/didyoumean"
+	"github.com/hashicorp/terraform/internal/getproviders/providerreqs"
+	"github.com/hashicorp/terraform/internal/getproviders/reattach"
 	"github.com/hashicorp/terraform/internal/plans"
 	"github.com/hashicorp/terraform/internal/providers"
+	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
 	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
@@ -75,6 +81,13 @@ type BackendOpts struct {
 	// This will only be set if the configuration contains a state_store block.
 	ProviderFactory providers.Factory
 
+	// Locks allows state-migration logic to detect when the provider used for pluggable state storage
+	// during the last init (i.e. what's in the backend state file) is mismatched with the provider
+	// version in use currently.
+	//
+	// This will only be set if the configuration contains a state_store block.
+	Locks *depsfile.Locks
+
 	// ConfigOverride is an hcl.Body that, if non-nil, will be used with
 	// configs.MergeBodies to override the type-specific backend configuration
 	// arguments in Config.
@@ -92,6 +105,10 @@ type BackendOpts struct {
 	// ViewType will set console output format for the
 	// initialization operation (JSON or human-readable).
 	ViewType arguments.ViewType
+
+	// CreateDefaultWorkspace signifies whether the operations backend should create
+	// the default workspace or not
+	CreateDefaultWorkspace bool
 }
 
 // BackendWithRemoteTerraformVersion is a shared interface between the 'remote' and 'cloud' backends
@@ -239,7 +256,7 @@ func (m *Meta) Backend(opts *BackendOpts) (backendrun.OperationsBackend, tfdiags
 		}
 	}
 
-	return local, nil
+	return local, diags
 }
 
 // selectWorkspace gets a list of existing workspaces and then checks
@@ -656,11 +673,13 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 	// Get the local 'backend' or 'state_store' configuration.
 	var backendConfig *configs.Backend
 	var stateStoreConfig *configs.StateStore
-	var cHash int
+	var backendHash int
+	var stateStoreHash int
+	var stateStoreProviderHash int
 	if opts.StateStoreConfig != nil {
 		// state store has been parsed from config and is included in opts
 		var ssDiags tfdiags.Diagnostics
-		stateStoreConfig, cHash, _, ssDiags = m.stateStoreConfig(opts)
+		stateStoreConfig, stateStoreHash, stateStoreProviderHash, ssDiags = m.stateStoreConfig(opts)
 		diags = diags.Append(ssDiags)
 		if ssDiags.HasErrors() {
 			return nil, diags
@@ -669,7 +688,7 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 		// backend config may or may not have been parsed and included in opts,
 		// or may not exist in config at all (default/implied local backend)
 		var beDiags tfdiags.Diagnostics
-		backendConfig, cHash, beDiags = m.backendConfig(opts)
+		backendConfig, backendHash, beDiags = m.backendConfig(opts)
 		diags = diags.Append(beDiags)
 		if beDiags.HasErrors() {
 			return nil, diags
@@ -699,7 +718,7 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 	statePath := filepath.Join(m.DataDir(), DefaultStateFilename)
 	sMgr := &clistate.LocalState{Path: statePath}
 	if err := sMgr.RefreshState(); err != nil {
-		diags = diags.Append(fmt.Errorf("Failed to load state: %s", err))
+		diags = diags.Append(fmt.Errorf("Failed to load the backend state file: %s", err))
 		return nil, diags
 	}
 
@@ -760,7 +779,7 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 			return nil, diags
 		}
 
-		return m.backend_c_r_S(backendConfig, cHash, sMgr, true, opts)
+		return m.backend_c_r_S(backendConfig, backendHash, sMgr, true, opts)
 
 	// We're unsetting a state_store (moving from state_store => local)
 	case stateStoreConfig == nil && !s.StateStore.Empty() &&
@@ -790,7 +809,7 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 			}
 			return nil, diags
 		}
-		return m.backend_C_r_s(backendConfig, cHash, sMgr, opts)
+		return m.backend_C_r_s(backendConfig, backendHash, sMgr, opts)
 
 	// Configuring a state store for the first time or -reconfigure flag was used
 	case stateStoreConfig != nil && s.StateStore.Empty() &&
@@ -800,11 +819,18 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 			stateStoreConfig.Provider.Name,
 			stateStoreConfig.ProviderAddr,
 		)
-		return nil, diags.Append(&hcl.Diagnostic{
-			Severity: hcl.DiagError,
-			Summary:  "Not implemented yet",
-			Detail:   "Configuring a state store for the first time is not implemented yet",
-		})
+
+		if !opts.Init {
+			initReason := fmt.Sprintf("Initial configuration of the requested state_store %q in provider %s (%q)",
+				stateStoreConfig.Type,
+				stateStoreConfig.Provider.Name,
+				stateStoreConfig.ProviderAddr,
+			)
+			diags = diags.Append(errStateStoreInitDiag(initReason))
+			return nil, diags
+		}
+
+		return m.stateStore_C_s(stateStoreConfig, stateStoreHash, stateStoreProviderHash, sMgr, opts)
 
 	// Migration from state store to backend
 	case backendConfig != nil && s.Backend.Empty() &&
@@ -844,7 +870,7 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 		// We're not initializing
 		// AND the backend cache hash values match, indicating that the stored config is valid and completely unchanged.
 		// AND we're not providing any overrides. An override can mean a change overriding an unchanged backend block (indicated by the hash value).
-		if (uint64(cHash) == s.Backend.Hash) && (!opts.Init || opts.ConfigOverride == nil) {
+		if (uint64(backendHash) == s.Backend.Hash) && (!opts.Init || opts.ConfigOverride == nil) {
 			log.Printf("[TRACE] Meta.Backend: using already-initialized, unchanged %q backend configuration", backendConfig.Type)
 			savedBackend, diags := m.savedBackend(sMgr)
 			// Verify that selected workspace exist. Otherwise prompt user to create one
@@ -872,7 +898,7 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 			// It's possible for a backend to be unchanged, and the config itself to
 			// have changed by moving a parameter from the config to `-backend-config`
 			// In this case, we update the Hash.
-			moreDiags = m.updateSavedBackendHash(cHash, sMgr)
+			moreDiags = m.updateSavedBackendHash(backendHash, sMgr)
 			if moreDiags.HasErrors() {
 				return nil, diags
 			}
@@ -903,7 +929,7 @@ func (m *Meta) backendFromConfig(opts *BackendOpts) (backend.Backend, tfdiags.Di
 		}
 
 		log.Printf("[WARN] backend config has changed since last init")
-		return m.backend_C_r_S_changed(backendConfig, cHash, sMgr, true, opts)
+		return m.backend_C_r_S_changed(backendConfig, backendHash, sMgr, true, opts)
 
 	// Potentially changing a state store configuration
 	case backendConfig == nil && s.Backend.Empty() &&
@@ -1528,6 +1554,275 @@ func (m *Meta) updateSavedBackendHash(cHash int, sMgr *clistate.LocalState) tfdi
 	return diags
 }
 
+//-------------------------------------------------------------------
+// State Store Config Scenarios
+// The functions below cover handling all the various scenarios that
+// can exist when loading a state store. They are named in the format of
+// "stateStore_C_S" where C and S may be upper or lowercase. Lowercase
+// means it is false, uppercase means it is true.
+//
+// The fields are:
+//
+//   * C - State store configuration is set and changed in TF files
+//   * S - State store configuration is set in the state
+//
+//-------------------------------------------------------------------
+
+// Configuring a state_store for the first time.
+func (m *Meta) stateStore_C_s(c *configs.StateStore, stateStoreHash int, providerHash int, backendSMgr *clistate.LocalState, opts *BackendOpts) (backend.Backend, tfdiags.Diagnostics) {
+	var diags tfdiags.Diagnostics
+
+	vt := arguments.ViewJSON
+	// Set default viewtype if none was set as the StateLocker needs to know exactly
+	// what viewType we want to have.
+	if opts == nil || opts.ViewType != vt {
+		vt = arguments.ViewHuman
+	}
+
+	// Grab a purely local backend to get the local state if it exists
+	localB, localBDiags := m.Backend(&BackendOpts{ForceLocal: true, Init: true})
+	if localBDiags.HasErrors() {
+		diags = diags.Append(localBDiags)
+		return nil, diags
+	}
+
+	workspaces, wDiags := localB.Workspaces()
+	if wDiags.HasErrors() {
+		diags = diags.Append(&errBackendLocalRead{wDiags.Err()})
+		return nil, diags
+	}
+
+	var localStates []statemgr.Full
+	for _, workspace := range workspaces {
+		localState, sDiags := localB.StateMgr(workspace)
+		if sDiags.HasErrors() {
+			diags = diags.Append(&errBackendLocalRead{sDiags.Err()})
+			return nil, diags
+		}
+		if err := localState.RefreshState(); err != nil {
+			diags = diags.Append(&errBackendLocalRead{err})
+			return nil, diags
+		}
+
+		// We only care about non-empty states.
+		if localS := localState.State(); !localS.Empty() {
+			log.Printf("[TRACE] Meta.Backend: will need to migrate workspace states because of existing %q workspace", workspace)
+			localStates = append(localStates, localState)
+		} else {
+			log.Printf("[TRACE] Meta.Backend: ignoring local %q workspace because its state is empty", workspace)
+		}
+	}
+
+	// Get the state store as an instance of backend.Backend
+	b, storeConfigVal, providerConfigVal, moreDiags := m.stateStoreInitFromConfig(c, opts.ProviderFactory)
+	diags = diags.Append(moreDiags)
+	if diags.HasErrors() {
+		return nil, diags
+	}
+
+	if len(localStates) > 0 {
+		// Migrate any local states into the new state store
+		err := m.backendMigrateState(&backendMigrateOpts{
+			SourceType:      "local",
+			DestinationType: c.Type,
+			Source:          localB,
+			Destination:     b,
+			ViewType:        vt,
+		})
+		if err != nil {
+			diags = diags.Append(err)
+			return nil, diags
+		}
+
+		// We remove the local state after migration to prevent confusion
+		// As we're migrating to a state store we don't have insight into whether it stores
+		// files locally at all, and whether those local files conflict with the location of
+		// the old local state.
+		log.Printf("[TRACE] Meta.Backend: removing old state snapshots from old backend")
+		for _, localState := range localStates {
+			// We always delete the local state, unless that was our new state too.
+			if err := localState.WriteState(nil); err != nil {
+				diags = diags.Append(&errBackendMigrateLocalDelete{err})
+				return nil, diags
+			}
+			if err := localState.PersistState(nil); err != nil {
+				diags = diags.Append(&errBackendMigrateLocalDelete{err})
+				return nil, diags
+			}
+		}
+	}
+
+	if m.stateLock {
+		view := views.NewStateLocker(vt, m.View)
+		stateLocker := clistate.NewLocker(m.stateLockTimeout, view)
+		if err := stateLocker.Lock(backendSMgr, "init is initializing state_store first time"); err != nil {
+			diags = diags.Append(fmt.Errorf("Error locking state: %s", err))
+			return nil, diags
+		}
+		defer stateLocker.Unlock()
+	}
+
+	// Store the state_store metadata in our saved state location
+	s := backendSMgr.State()
+	if s == nil {
+		s = workdir.NewBackendStateFile()
+	}
+
+	var pVersion *version.Version // This will remain nil for builtin providers or unmanaged providers.
+	if c.ProviderAddr.Hostname == addrs.BuiltInProviderHost {
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagWarning,
+			Summary:  "State storage is using a builtin provider",
+			Detail:   "Terraform is using a builtin provider for initializing state storage. Terraform will be less able to detect when state migrations are required in future init commands.",
+		})
+	} else {
+		isReattached, err := reattach.IsProviderReattached(c.ProviderAddr, os.Getenv("TF_REATTACH_PROVIDERS"))
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("Error determining if the state storage provider is reattached or not. This is a bug in Terraform and should be reported: %w",
+				err))
+			return nil, diags
+		}
+		if isReattached {
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagWarning,
+				Summary:  "State storage provider is not managed by Terraform",
+				Detail:   "Terraform is using a provider supplied via TF_REATTACH_PROVIDERS for initializing state storage. Terraform will be less able to detect when state migrations are required in future init commands.",
+			})
+		} else {
+			// The provider is not built in and is being managed by Terraform
+			// This is the most common scenario, by far.
+			pLock := opts.Locks.Provider(c.ProviderAddr)
+			if pLock == nil {
+				diags = diags.Append(fmt.Errorf("The provider %s (%q) is not present in the lockfile, despite being used for state store %q. This is a bug in Terraform and should be reported.",
+					c.Provider.Name,
+					c.ProviderAddr,
+					c.Type))
+				return nil, diags
+			}
+			var err error
+			pVersion, err = providerreqs.GoVersionFromVersion(pLock.Version())
+			if err != nil {
+				diags = diags.Append(fmt.Errorf("Failed obtain the in-use version of provider %s (%q) when recording backend state for state store %q. This is a bug in Terraform and should be reported: %w",
+					c.Provider.Name,
+					c.ProviderAddr,
+					c.Type,
+					err))
+				return nil, diags
+			}
+		}
+	}
+	s.StateStore = &workdir.StateStoreConfigState{
+		Type: c.Type,
+		Hash: uint64(stateStoreHash),
+		Provider: &workdir.ProviderConfigState{
+			Source:  &c.ProviderAddr,
+			Version: pVersion,
+			Hash:    uint64(providerHash),
+		},
+	}
+	s.StateStore.SetConfig(storeConfigVal, b.ConfigSchema())
+	if plug, ok := b.(*backendPluggable.Pluggable); ok {
+		// We need to convert away from backend.Backend interface to use the method
+		// for accessing the provider schema.
+		s.StateStore.Provider.SetConfig(providerConfigVal, plug.ProviderSchema())
+	}
+
+	// Verify that selected workspace exists in the state store.
+	if opts.Init && b != nil {
+		err := m.selectWorkspace(b)
+		if err != nil {
+			if errors.Is(err, &errBackendNoExistingWorkspaces{}) {
+				// If there are no workspaces, Terraform either needs to create the default workspace here
+				// or instruct the user to run a `terraform workspace new` command.
+				ws, err := m.Workspace()
+				if err != nil {
+					diags = diags.Append(fmt.Errorf("Failed to check current workspace: %w", err))
+					return nil, diags
+				}
+
+				if ws == backend.DefaultStateName {
+					// Users control if the default workspace is created through the -create-default-workspace flag (defaults to true)
+					if opts.CreateDefaultWorkspace {
+						diags = diags.Append(m.createDefaultWorkspace(c, b))
+						if !diags.HasErrors() {
+							// Report workspace creation to the view
+							view := views.NewInit(vt, m.View)
+							view.Output(views.DefaultWorkspaceCreatedMessage)
+						}
+					} else {
+						diags = diags.Append(&hcl.Diagnostic{
+							Severity: hcl.DiagWarning,
+							Summary:  "The default workspace does not exist",
+							Detail:   "Terraform has been configured to skip creation of the default workspace in the state store. To create it, either remove the `-create-default-workspace=false` flag and re-run the 'init' command, or create it using a 'workspace new' command",
+						})
+					}
+				} else {
+					// User needs to run a `terraform workspace new` command to create the missing custom workspace.
+					diags = append(diags, tfdiags.Sourceless(
+						tfdiags.Error,
+						fmt.Sprintf("Workspace %q has not been created yet", ws),
+						fmt.Sprintf("State store %q in provider %s (%q) reports that no workspaces currently exist. To create the custom workspace %q use the command `terraform workspace new %s`.",
+							c.Type,
+							c.Provider.Name,
+							c.ProviderAddr,
+							ws,
+							ws,
+						),
+					))
+					return nil, diags
+				}
+			} else {
+				// For all other errors, report via diagnostics
+				diags = diags.Append(fmt.Errorf("Failed to select a workspace: %w", err))
+			}
+		}
+	}
+	if diags.HasErrors() {
+		return nil, diags
+	}
+
+	// Update backend state file
+	if err := backendSMgr.WriteState(s); err != nil {
+		diags = diags.Append(errBackendWriteSavedDiag(err))
+		return nil, diags
+	}
+	if err := backendSMgr.PersistState(); err != nil {
+		diags = diags.Append(errBackendWriteSavedDiag(err))
+		return nil, diags
+	}
+
+	return b, diags
+}
+
+// createDefaultWorkspace receives a backend made using a pluggable state store, and details about that store's config,
+// and persists an empty state file in the default workspace. By creating this artifact we ensure that the default
+// workspace is created and usable by Terraform in later operations.
+func (m *Meta) createDefaultWorkspace(c *configs.StateStore, b backend.Backend) tfdiags.Diagnostics {
+	var diags tfdiags.Diagnostics
+
+	defaultSMgr, sDiags := b.StateMgr(backend.DefaultStateName)
+	diags = diags.Append(sDiags)
+	if sDiags.HasErrors() {
+		diags = diags.Append(fmt.Errorf("Failed to create a state manager for state store %q in provider %s (%q). This is a bug in Terraform and should be reported: %w",
+			c.Type,
+			c.Provider.Name,
+			c.ProviderAddr,
+			sDiags.Err()))
+		return diags
+	}
+	emptyState := states.NewState()
+	if err := defaultSMgr.WriteState(emptyState); err != nil {
+		diags = diags.Append(errStateStoreWorkspaceCreateDiag(err, c.Type))
+		return diags
+	}
+	if err := defaultSMgr.PersistState(nil); err != nil {
+		diags = diags.Append(errStateStoreWorkspaceCreateDiag(err, c.Type))
+		return diags
+	}
+
+	return diags
+}
+
 // Initializing a saved state store from the backend state file (aka 'cache file', aka 'legacy state file')
 func (m *Meta) savedStateStore(sMgr *clistate.LocalState, factory providers.Factory) (backend.Backend, tfdiags.Diagnostics) {
 	// We're preparing a state_store version of backend.Backend.

internal/command/meta_backend_errors.go

@@ -123,6 +123,35 @@ configuration or state have been made.`, initReason)
 	)
 }
 
+// errStateStoreInitDiag creates a diagnostic to present to users when
+// users attempt to run a non-init command after making a change to their
+// state_store configuration.
+//
+// An init reason should be provided as an argument.
+func errStateStoreInitDiag(initReason string) tfdiags.Diagnostic {
+	msg := fmt.Sprintf(`Reason: %s
+
+The "state store" is the interface that Terraform uses to store state when
+performing operations on the local machine. If this message is showing up,
+it means that the Terraform configuration you're using is using a custom
+configuration for state storage in Terraform.
+
+Changes to state store configurations require reinitialization. This allows
+Terraform to set up the new configuration, copy existing state, etc. Please run
+"terraform init" with either the "-reconfigure" or "-migrate-state" flags to
+use the current configuration.
+
+If the change reason above is incorrect, please verify your configuration
+hasn't changed and try again. At this point, no changes to your existing
+configuration or state have been made.`, initReason)
+
+	return tfdiags.Sourceless(
+		tfdiags.Error,
+		"State store initialization required, please run \"terraform init\"",
+		msg,
+	)
+}
+
 // errBackendInitCloudDiag creates a diagnostic to present to users when
 // an init command encounters config changes in a `cloud` block.
 //
@@ -176,6 +205,23 @@ If the backend already contains existing workspaces, you may need to update
 the backend configuration.`
 }
 
+func errStateStoreWorkspaceCreateDiag(innerError error, storeType string) tfdiags.Diagnostic {
+	msg := fmt.Sprintf(`Error creating the default workspace using pluggable state store %s: %s
+
+This could be a bug in the provider used for state storage, or a bug in
+Terraform. Please file an issue with the provider developers before reporting
+a bug for Terraform.`,
+		storeType,
+		innerError,
+	)
+
+	return tfdiags.Sourceless(
+		tfdiags.Error,
+		"Cannot create the default workspace",
+		msg,
+	)
+}
+
 // migrateOrReconfigDiag creates a diagnostic to present to users when
 // an init command encounters a mismatch in backend state and the current config
 // and Terraform needs users to provide additional instructions about how Terraform

internal/command/meta_backend_test.go

@@ -636,7 +636,7 @@ func TestMetaBackend_configureNewBackendWithStateExistingNoMigrate(t *testing.T)
 }
 
 // Saved backend state matching config
-func TestMetaBackend_configuredUnchanged(t *testing.T) {
+func TestMetaBackend_configuredBackendUnchanged(t *testing.T) {
 	td := t.TempDir()
 	testCopyDir(t, testFixturePath("backend-unchanged"), td)
 	t.Chdir(td)
@@ -2086,50 +2086,6 @@ func Test_determineInitReason(t *testing.T) {
 	}
 }
 
-// Newly configured state store
-//
-// TODO(SarahFrench/radeksimko): currently this test only confirms that we're hitting the switch
-// case for this scenario, and will need to be updated when that init feature is implemented.
-func TestMetaBackend_configureNewStateStore(t *testing.T) {
-	td := t.TempDir()
-	testCopyDir(t, testFixturePath("state-store-new"), td)
-	t.Chdir(td)
-
-	// Setup the meta
-	m := testMetaBackend(t, nil)
-	m.AllowExperimentalFeatures = true
-
-	// Get the state store's config
-	mod, loadDiags := m.loadSingleModule(td)
-	if loadDiags.HasErrors() {
-		t.Fatalf("unexpected error when loading test config: %s", loadDiags.Err())
-	}
-
-	// Get mock provider factory to be used during init
-	//
-	// This imagines a provider called foo that contains
-	// a pluggable state store implementation called bar.
-	mock := testStateStoreMock(t)
-	factory := func() (providers.Interface, error) {
-		return mock, nil
-	}
-
-	// Get the operations backend
-	_, beDiags := m.Backend(&BackendOpts{
-		Init:             true,
-		StateStoreConfig: mod.StateStore,
-		ProviderFactory:  factory,
-	})
-	if !beDiags.HasErrors() {
-		t.Fatal("expected an error to be returned during partial implementation of PSS")
-	}
-	wantErr := "Configuring a state store for the first time is not implemented yet"
-	if !strings.Contains(beDiags.Err().Error(), wantErr) {
-		t.Fatalf("expected the returned error to contain %q, but got: %s", wantErr, beDiags.Err())
-	}
-
-}
-
 // Unsetting a saved state store
 //
 // TODO(SarahFrench/radeksimko): currently this test only confirms that we're hitting the switch
@@ -2168,59 +2124,6 @@ func TestMetaBackend_configuredStateStoreUnset(t *testing.T) {
 	}
 }
 
-// Reconfiguring with an already configured state store.
-// This should ignore the existing state_store config, and configure the new
-// state store is if this is the first time.
-//
-// TODO(SarahFrench/radeksimko): currently this test only confirms that we're hitting the switch
-// case for this scenario, and will need to be updated when that init feature is implemented.
-func TestMetaBackend_reconfigureStateStoreChange(t *testing.T) {
-	td := t.TempDir()
-	testCopyDir(t, testFixturePath("state-store-reconfigure"), td)
-	t.Chdir(td)
-
-	// Setup the meta
-	m := testMetaBackend(t, nil)
-	m.AllowExperimentalFeatures = true
-
-	// this should not ask for input
-	m.input = false
-
-	// cli flag -reconfigure
-	m.reconfigure = true
-
-	// Get the state store's config
-	mod, loadDiags := m.loadSingleModule(td)
-	if loadDiags.HasErrors() {
-		t.Fatalf("unexpected error when loading test config: %s", loadDiags.Err())
-	}
-
-	// Get mock provider factory to be used during init
-	//
-	// This imagines a provider called foo that contains
-	// a pluggable state store implementation called bar.
-	mock := testStateStoreMock(t)
-	factory := func() (providers.Interface, error) {
-		return mock, nil
-	}
-
-	// Get the operations backend
-	_, beDiags := m.Backend(&BackendOpts{
-		Init:             true,
-		StateStoreConfig: mod.StateStore,
-		ProviderFactory:  factory,
-	})
-
-	if !beDiags.HasErrors() {
-		t.Fatal("expected an error to be returned during partial implementation of PSS")
-	}
-	wantErr := "Configuring a state store for the first time is not implemented yet"
-	if !strings.Contains(beDiags.Err().Error(), wantErr) {
-		t.Fatalf("expected the returned error to contain %q, but got: %s", wantErr, beDiags.Err())
-	}
-
-}
-
 // Changing a configured state store
 //
 // TODO(SarahFrench/radeksimko): currently this test only confirms that we're hitting the switch
@@ -2241,20 +2144,17 @@ func TestMetaBackend_changeConfiguredStateStore(t *testing.T) {
 		t.Fatalf("unexpected error when loading test config: %s", loadDiags.Err())
 	}
 
-	// Get mock provider factory to be used during init
+	// Get mock provider to be used during init
 	//
-	// This imagines a provider called foo that contains
-	// a pluggable state store implementation called bar.
+	// This imagines a provider called "test" that contains
+	// a pluggable state store implementation called "store".
 	mock := testStateStoreMock(t)
-	factory := func() (providers.Interface, error) {
-		return mock, nil
-	}
 
 	// Get the operations backend
 	_, beDiags := m.Backend(&BackendOpts{
 		Init:             true,
 		StateStoreConfig: mod.StateStore,
-		ProviderFactory:  factory,
+		ProviderFactory:  providers.FactoryFixed(mock),
 	})
 	if !beDiags.HasErrors() {
 		t.Fatal("expected an error to be returned during partial implementation of PSS")
@@ -2284,20 +2184,17 @@ func TestMetaBackend_configuredBackendToStateStore(t *testing.T) {
 		t.Fatalf("unexpected error when loading test config: %s", loadDiags.Err())
 	}
 
-	// Get mock provider factory to be used during init
+	// Get mock provider to be used during init
 	//
-	// This imagines a provider called foo that contains
-	// a pluggable state store implementation called bar.
+	// This imagines a provider called "test" that contains
+	// a pluggable state store implementation called "store".
 	mock := testStateStoreMock(t)
-	factory := func() (providers.Interface, error) {
-		return mock, nil
-	}
 
 	// Get the operations backend
 	_, beDiags := m.Backend(&BackendOpts{
 		Init:             true,
 		StateStoreConfig: mod.StateStore,
-		ProviderFactory:  factory,
+		ProviderFactory:  providers.FactoryFixed(mock),
 	})
 	if !beDiags.HasErrors() {
 		t.Fatal("expected an error to be returned during partial implementation of PSS")
@@ -2381,20 +2278,17 @@ func TestMetaBackend_configureStateStoreVariableUse(t *testing.T) {
 				t.Fatalf("unexpected error when loading test config: %s", loadDiags.Err())
 			}
 
-			// Get mock provider factory to be used during init
+			// Get mock provider to be used during init
 			//
-			// This imagines a provider called foo that contains
-			// a pluggable state store implementation called bar.
+			// This imagines a provider called "test" that contains
+			// a pluggable state store implementation called "store".
 			mock := testStateStoreMock(t)
-			factory := func() (providers.Interface, error) {
-				return mock, nil
-			}
 
 			// Get the operations backend
 			_, err := m.Backend(&BackendOpts{
 				Init:             true,
 				StateStoreConfig: mod.StateStore,
-				ProviderFactory:  factory,
+				ProviderFactory:  providers.FactoryFixed(mock),
 			})
 			if err == nil {
 				t.Fatal("should error")

internal/command/meta_providers.go

@@ -382,6 +382,12 @@ func (m *Meta) providerFactoriesFromLocks(locks *depsfile.Locks) (map[addrs.Prov
 	for provider, reattach := range unmanagedProviders {
 		factories[provider] = unmanagedProviderFactory(provider, reattach)
 	}
+	if m.testingOverrides != nil {
+		// Allow tests, where testingOverrides is set, to see test providers in locks
+		for provider, factory := range m.testingOverrides.Providers {
+			factories[provider] = factory
+		}
+	}
 
 	var err error
 	if len(errs) > 0 {

internal/command/testdata/init-with-state-store/main.tf

@@ -1,11 +1,14 @@
 terraform {
 
   required_providers {
-    foo = {
-      source = "my-org/foo"
+    test = {
+      source = "hashicorp/test"
     }
   }
-  state_store "foo_foo" {
-    provider "foo" {}
+  state_store "test_store" {
+    provider "test" {
+    }
+
+    value = "foobar"
   }
 }

internal/command/testdata/state-store-reconfigure/.terraform/terraform.tfstate

@@ -9,7 +9,7 @@
         },
         "provider": {
             "version": "1.2.3",
-            "source": "registry.terraform.io/my-org/foo",
+            "source": "registry.terraform.io/hashicorp/test",
             "config": {},
             "hash": 12345
         },

internal/command/testdata/state-store-to-backend/.terraform/terraform.tfstate

@@ -9,7 +9,7 @@
         },
         "provider": {
             "version": "1.2.3",
-            "source": "registry.terraform.io/my-org/foo",
+            "source": "registry.terraform.io/hashicorp/test",
             "config": {},
             "hash": 12345
         },

internal/command/testdata/state-store-to-backend/main.tf

@@ -1,7 +1,7 @@
 terraform {
   required_providers {
-    foo = {
-      source = "my-org/foo"
+    test = {
+      source = "hashicorp/test"
     }
   }
 

internal/command/testdata/state-store-unset/.terraform/terraform.tfstate

@@ -9,7 +9,7 @@
         },
         "provider": {
             "version": "1.2.3",
-            "source": "registry.terraform.io/my-org/foo",
+            "source": "registry.terraform.io/hashicorp/test",
             "config": {},
             "hash": 12345
         },

internal/command/testdata/state-store-unset/main.tf

@@ -1,7 +1,7 @@
 terraform {
   required_providers {
-    foo = {
-      source = "my-org/foo"
+    test = {
+      source = "hashicorp/test"
     }
   }
 

internal/command/views/init.go

@@ -198,6 +198,10 @@ var MessageRegistry map[InitMessageCode]InitMessage = map[InitMessageCode]InitMe
 		HumanValue: "\n[reset][bold]Initializing the state store...",
 		JSONValue:  "Initializing the state store...",
 	},
+	"default_workspace_created_message": {
+		HumanValue: defaultWorkspaceCreatedInfo,
+		JSONValue:  defaultWorkspaceCreatedInfo,
+	},
 	"dependencies_lock_changes_info": {
 		HumanValue: dependenciesLockChangesInfo,
 		JSONValue:  dependenciesLockChangesInfo,
@@ -278,6 +282,7 @@ const (
 	InitializingModulesMessage        InitMessageCode = "initializing_modules_message"
 	InitializingBackendMessage        InitMessageCode = "initializing_backend_message"
 	InitializingStateStoreMessage     InitMessageCode = "initializing_state_store_message"
+	DefaultWorkspaceCreatedMessage    InitMessageCode = "default_workspace_created_message"
 	InitializingProviderPluginMessage InitMessageCode = "initializing_provider_plugin_message"
 	LockInfo                          InitMessageCode = "lock_info"
 	DependenciesLockChangesInfo       InitMessageCode = "dependencies_lock_changes_info"
@@ -393,6 +398,13 @@ selections it made above. Include this file in your version control repository
 so that Terraform can guarantee to make the same selections by default when
 you run "terraform init" in the future.`
 
+const defaultWorkspaceCreatedInfo = `
+Terraform created an empty state file for the default workspace in your state store
+because it didn't exist. If this was not intended, read the init command's documentation for
+more guidance:
+https://developer.hashicorp.com/terraform/cli/commands/init
+`
+
 const dependenciesLockChangesInfo = `
 Terraform has made some changes to the provider dependency selections recorded
 in the .terraform.lock.hcl file. Review those changes and commit them to your

internal/command/workdir/backend_state_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	tfaddr "github.com/hashicorp/terraform-registry-address"
 	"github.com/hashicorp/terraform/version"
 )
 
@@ -160,9 +161,12 @@ func TestParseBackendStateFile(t *testing.T) {
 }
 
 func TestEncodeBackendStateFile(t *testing.T) {
+	noVersionData := ""
+
 	tfVersion := version.Version
 	tests := map[string]struct {
 		Input   *BackendStateFile
+		Envs    map[string]string
 		Want    []byte
 		WantErr string
 	}{
@@ -177,11 +181,58 @@ func TestEncodeBackendStateFile(t *testing.T) {
 			},
 			Want: []byte("{\n  \"version\": 3,\n  \"terraform_version\": \"" + tfVersion + "\",\n  \"state_store\": {\n    \"type\": \"foobar_baz\",\n    \"provider\": {\n      \"version\": \"1.2.3\",\n      \"source\": \"registry.terraform.io/my-org/foobar\",\n      \"config\": {\n        \"foo\": \"bar\"\n      },\n      \"hash\": 12345\n    },\n    \"config\": {\n      \"foo\": \"bar\"\n    },\n    \"hash\": 123\n  }\n}"),
 		},
-		"it returns an error when neither backend nor state_store config state are present": {
+		"it's valid to record no version data when a builtin provider used for state store": {
+			Input: &BackendStateFile{
+				StateStore: &StateStoreConfigState{
+					Type:      "foobar_baz",
+					Provider:  getTestProviderState(t, noVersionData, string(tfaddr.BuiltInProviderHost), string(tfaddr.BuiltInProviderNamespace), "foobar", `{"foo": "bar"}`),
+					ConfigRaw: json.RawMessage([]byte(`{"foo":"bar"}`)),
+					Hash:      123,
+				},
+			},
+			Want: []byte("{\n  \"version\": 3,\n  \"terraform_version\": \"" + tfVersion + "\",\n  \"state_store\": {\n    \"type\": \"foobar_baz\",\n    \"provider\": {\n      \"version\": null,\n      \"source\": \"terraform.io/builtin/foobar\",\n      \"config\": {\n        \"foo\": \"bar\"\n      },\n      \"hash\": 12345\n    },\n    \"config\": {\n      \"foo\": \"bar\"\n    },\n    \"hash\": 123\n  }\n}"),
+		},
+		"it's valid to record no version data when a re-attached provider used for state store": {
+			Input: &BackendStateFile{
+				StateStore: &StateStoreConfigState{
+					Type:      "foobar_baz",
+					Provider:  getTestProviderState(t, noVersionData, "registry.terraform.io", "hashicorp", "foobar", `{"foo": "bar"}`),
+					ConfigRaw: json.RawMessage([]byte(`{"foo":"bar"}`)),
+					Hash:      123,
+				},
+			},
+			Envs: map[string]string{
+				"TF_REATTACH_PROVIDERS": `{
+				"foobar": {
+					"Protocol": "grpc",
+					"ProtocolVersion": 6,
+					"Pid": 12345,
+					"Test": true,
+					"Addr": {
+						"Network": "unix",
+						"String":"/var/folders/xx/abcde12345/T/plugin12345"
+					}
+				}
+			}`,
+			},
+			Want: []byte("{\n  \"version\": 3,\n  \"terraform_version\": \"" + tfVersion + "\",\n  \"state_store\": {\n    \"type\": \"foobar_baz\",\n    \"provider\": {\n      \"version\": null,\n      \"source\": \"registry.terraform.io/hashicorp/foobar\",\n      \"config\": {\n        \"foo\": \"bar\"\n      },\n      \"hash\": 12345\n    },\n    \"config\": {\n      \"foo\": \"bar\"\n    },\n    \"hash\": 123\n  }\n}"),
+		},
+		"error when neither backend nor state_store config state are present": {
 			Input: &BackendStateFile{},
 			Want:  []byte("{\n  \"version\": 3,\n  \"terraform_version\": \"" + tfVersion + "\"\n}"),
 		},
-		"it returns an error when the provider source's hostname is missing": {
+		"error when the provider is neither builtin nor reattached and the provider version is missing": {
+			Input: &BackendStateFile{
+				StateStore: &StateStoreConfigState{
+					Type:      "foobar_baz",
+					Provider:  getTestProviderState(t, noVersionData, "registry.terraform.io", "my-org", "foobar", ""),
+					ConfigRaw: json.RawMessage([]byte(`{"foo":"bar"}`)),
+					Hash:      123,
+				},
+			},
+			WantErr: `state store is not valid: provider version data is missing`,
+		},
+		"error when the provider source's hostname is missing": {
 			Input: &BackendStateFile{
 				StateStore: &StateStoreConfigState{
 					Type:      "foobar_baz",
@@ -192,7 +243,7 @@ func TestEncodeBackendStateFile(t *testing.T) {
 			},
 			WantErr: `state store is not valid: Unknown hostname: Expected hostname in the provider address to be set`,
 		},
-		"it returns an error when the provider source's hostname and namespace are missing ": {
+		"error when the provider source's hostname and namespace are missing ": {
 			Input: &BackendStateFile{
 				StateStore: &StateStoreConfigState{
 					Type:      "foobar_baz",
@@ -203,7 +254,7 @@ func TestEncodeBackendStateFile(t *testing.T) {
 			},
 			WantErr: `state store is not valid: Unknown hostname: Expected hostname in the provider address to be set`,
 		},
-		"it returns an error when the provider source is completely missing ": {
+		"error when the provider source is completely missing ": {
 			Input: &BackendStateFile{
 				StateStore: &StateStoreConfigState{
 					Type:      "foobar_baz",
@@ -214,7 +265,7 @@ func TestEncodeBackendStateFile(t *testing.T) {
 			},
 			WantErr: `state store is not valid: Empty provider address: Expected address composed of hostname, provider namespace and name`,
 		},
-		"it returns an error when both backend and state_store config state are present": {
+		"error when both backend and state_store config state are present": {
 			Input: &BackendStateFile{
 				Backend: &BackendConfigState{
 					Type:      "foobar",
@@ -234,6 +285,11 @@ func TestEncodeBackendStateFile(t *testing.T) {
 
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
+			// Some test cases depend on ENVs, not all
+			for k, v := range test.Envs {
+				t.Setenv(k, v)
+			}
+
 			got, err := EncodeBackendStateFile(test.Input)
 
 			if test.WantErr != "" {

internal/command/workdir/statestore_config_state.go

@@ -7,10 +7,12 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
 
 	version "github.com/hashicorp/go-version"
 	tfaddr "github.com/hashicorp/terraform-registry-address"
 	"github.com/hashicorp/terraform/internal/configs/configschema"
+	"github.com/hashicorp/terraform/internal/getproviders/reattach"
 	"github.com/hashicorp/terraform/internal/plans"
 	"github.com/zclconf/go-cty/cty"
 	ctyjson "github.com/zclconf/go-cty/cty/json"
@@ -40,13 +42,13 @@ func (s *StateStoreConfigState) Validate() error {
 
 	// Are any bits of data totally missing?
 	if s.Empty() {
-		return fmt.Errorf("state store is not valid: data is empty")
+		return fmt.Errorf("attempted to encode a malformed backend state file; data is empty")
+	}
+	if s.Type == "" {
+		return fmt.Errorf("attempted to encode a malformed backend state file; state store type is missing")
 	}
 	if s.Provider == nil {
-		return fmt.Errorf("state store is not valid: provider data is missing")
-	}
-	if s.Provider.Version == nil {
-		return fmt.Errorf("state store is not valid: version data is missing")
+		return fmt.Errorf("attempted to encode a malformed backend state file; provider data is missing")
 	}
 	if s.ConfigRaw == nil {
 		return fmt.Errorf("attempted to encode a malformed backend state file; state_store configuration data is missing")
@@ -58,6 +60,18 @@ func (s *StateStoreConfigState) Validate() error {
 		return fmt.Errorf("state store is not valid: %w", err)
 	}
 
+	// Version information is required if the provider isn't builtin or unmanaged by Terraform
+	isReattached, err := reattach.IsProviderReattached(*s.Provider.Source, os.Getenv("TF_REATTACH_PROVIDERS"))
+	if err != nil {
+		return fmt.Errorf("error determining if state storage provider is reattached: %w", err)
+	}
+	if (s.Provider.Source.Hostname != tfaddr.BuiltInProviderHost) &&
+		!isReattached {
+		if s.Provider.Version == nil {
+			return fmt.Errorf("state store is not valid: provider version data is missing")
+		}
+	}
+
 	return nil
 }
 

internal/command/workdir/testing.go

@@ -17,9 +17,16 @@ import (
 func getTestProviderState(t *testing.T, semVer, hostname, namespace, typeName, config string) *ProviderConfigState {
 	t.Helper()
 
-	ver, err := version.NewSemver(semVer)
-	if err != nil {
-		t.Fatalf("test setup failed when creating version.Version: %s", err)
+	var ver *version.Version
+	if semVer == "" {
+		// Allow passing no version in; leave ver nil
+		ver = nil
+	} else {
+		var err error
+		ver, err = version.NewSemver(semVer)
+		if err != nil {
+			t.Fatalf("test setup failed when creating version.Version: %s", err)
+		}
 	}
 
 	return &ProviderConfigState{

internal/configs/state_store.go

@@ -116,13 +116,16 @@ func resolveStateStoreProviderType(requiredProviders map[string]*RequiredProvide
 		diags = append(diags, &hcl.Diagnostic{
 			Severity: hcl.DiagError,
 			Summary:  "Missing entry in required_providers",
-			Detail: fmt.Sprintf("The provider used for state storage must have a matching entry in required_providers. Please add an entry for provider %q",
-				stateStore.Provider.Name),
+			Detail: fmt.Sprintf("The provider used for state storage must have a matching entry in required_providers. Please add an entry for provider %s",
+				stateStore.Provider.Name,
+			),
 			Subject: &stateStore.DeclRange,
 		})
 		return tfaddr.Provider{}, diags
 	default:
 		// We've got a required_providers entry to use
+		// This code path is used for both re-attached providers
+		// providers that are fully managed by Terraform.
 		return addr.Type, nil
 	}
 }

internal/configs/state_store_test.go

@@ -274,7 +274,7 @@ func configBodyForTest(t *testing.T, config string) hcl.Body {
 	t.Helper()
 	f, diags := hclsyntax.ParseConfig([]byte(config), "", hcl.Pos{Line: 1, Column: 1})
 	if diags.HasErrors() {
-		t.Fatalf("failure creating hcl.Body during test setup")
+		t.Fatalf("failure creating hcl.Body during test setup: %s", diags.Error())
 	}
 	return f.Body
 }

internal/getproviders/providerreqs/version.go

@@ -20,6 +20,7 @@ import (
 	"github.com/apparentlymart/go-versions/versions"
 	"github.com/apparentlymart/go-versions/versions/constraints"
 
+	"github.com/hashicorp/go-version"
 	"github.com/hashicorp/terraform/internal/addrs"
 )
 
@@ -86,6 +87,12 @@ func ParseVersion(str string) (Version, error) {
 	return versions.ParseVersion(str)
 }
 
+// GoVersionFromVersion converts a Version from the providerreqs package
+// into a Version from the hashicorp/go-version module.
+func GoVersionFromVersion(v Version) (*version.Version, error) {
+	return version.NewVersion(v.String())
+}
+
 // MustParseVersion is a variant of ParseVersion that panics if it encounters
 // an error while parsing.
 func MustParseVersion(str string) Version {

internal/getproviders/providerreqs/version_test.go

@@ -5,6 +5,8 @@ package providerreqs
 
 import (
 	"testing"
+
+	"github.com/hashicorp/go-version"
 )
 
 func TestVersionConstraintsString(t *testing.T) {
@@ -97,3 +99,20 @@ func TestVersionConstraintsString(t *testing.T) {
 		})
 	}
 }
+
+func TestGoVersionFromVersion(t *testing.T) {
+	versionString := "1.0.0"
+	v := MustParseVersion(versionString)
+
+	var goV *version.Version
+	goV, err := GoVersionFromVersion(v)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if goV.String() != versionString {
+		t.Fatalf("unexpected version, expected string representation to be %q but got %q",
+			versionString,
+			goV.String(),
+		)
+	}
+}

internal/getproviders/reattach/reattach.go

@@ -9,6 +9,7 @@ import (
 	"net"
 
 	"github.com/hashicorp/go-plugin"
+	tfaddr "github.com/hashicorp/terraform-registry-address"
 	"github.com/hashicorp/terraform/internal/addrs"
 )
 
@@ -88,7 +89,7 @@ func ParseReattachProviders(in string) (map[addrs.Provider]*plugin.ReattachConfi
 // environment variable.
 //
 // Calling code is expected to pass in a provider address and the value of os.Getenv("TF_REATTACH_PROVIDERS")
-func IsProviderReattached(provider addrs.Provider, in string) (bool, error) {
+func IsProviderReattached(provider tfaddr.Provider, in string) (bool, error) {
 	providers, err := ParseReattachProviders(in)
 	if err != nil {
 		return false, err

internal/providers/testing/provider_mock.go

@@ -5,12 +5,15 @@ package testing
 
 import (
 	"fmt"
+	"maps"
+	"slices"
 	"sync"
 
 	"github.com/zclconf/go-cty/cty"
 	ctyjson "github.com/zclconf/go-cty/cty/json"
 	"github.com/zclconf/go-cty/cty/msgpack"
 
+	"github.com/hashicorp/hcl/v2"
 	"github.com/hashicorp/terraform/internal/configs/hcl2shim"
 	"github.com/hashicorp/terraform/internal/providers"
 )
@@ -171,6 +174,10 @@ type MockProvider struct {
 	UnlockStateRequest  providers.UnlockStateRequest
 	UnlockStateFn       func(providers.UnlockStateRequest) providers.UnlockStateResponse
 
+	// MockStates is an internal field that tracks which workspaces have been created in a test
+	// The map keys are state ids (workspaces) and the value depends on the test.
+	MockStates map[string]interface{}
+
 	GetStatesCalled   bool
 	GetStatesResponse *providers.GetStatesResponse
 	GetStatesRequest  providers.GetStatesRequest
@@ -352,6 +359,10 @@ func (p *MockProvider) WriteStateBytes(r providers.WriteStateBytesRequest) (resp
 		return p.WriteStateBytesFn(r)
 	}
 
+	// If we haven't already, record in the mock that
+	// the matching workspace exists
+	p.MockStates[r.StateId] = true
+
 	return p.WriteStateBytesResponse
 }
 
@@ -1073,11 +1084,8 @@ func (p *MockProvider) GetStates(r providers.GetStatesRequest) (resp providers.G
 		return p.GetStatesFn(r)
 	}
 
-	// If the mock has no further inputs, return an empty list.
-	// The state store should be reporting a minimum of the default workspace usually,
-	// but this should be achieved by querying data storage and identifying the artifact
-	// for that workspace, and reporting that the workspace exists.
-	resp.States = []string{}
+	// When no custom logic is provided to the mock, return the internal states list
+	resp.States = slices.Sorted(maps.Keys(p.MockStates))
 
 	return resp
 }
@@ -1104,7 +1112,16 @@ func (p *MockProvider) DeleteState(r providers.DeleteStateRequest) (resp provide
 		return p.DeleteStateFn(r)
 	}
 
-	// There's no logic we can include here in the absence of other fields on the mock.
+	// When no custom logic is provided to the mock, delete matching internal state
+	if _, match := p.MockStates[r.StateId]; match {
+		delete(p.MockStates, r.StateId)
+	} else {
+		resp.Diagnostics.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Workspace cannot be deleted",
+			Detail:   fmt.Sprintf("The workspace %q does not exist, so cannot be deleted", r.StateId),
+		})
+	}
 
 	// If the response contains no diagnostics then the deletion is assumed to be successful.
 	return resp