diff --git a/internal/command/cloud_test.go b/internal/command/cloud_test.go
index 3661e4b98b..cf59acfbdf 100644
--- a/internal/command/cloud_test.go
+++ b/internal/command/cloud_test.go
@@ -135,7 +135,7 @@ func TestCloud_withBackendConfig(t *testing.T) {
 
 	// Initialize the backend
 	ic := &InitCommand{
-		Meta{
+		Meta: Meta{
 			Ui:               ui,
 			View:             view,
 			testingOverrides: metaOverridesForProvider(testProvider()),
diff --git a/internal/command/init.go b/internal/command/init.go
index 3618f6f287..114ce62a29 100644
--- a/internal/command/init.go
+++ b/internal/command/init.go
@@ -28,8 +28,10 @@ import (
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/configs"
 	"github.com/hashicorp/terraform/internal/configs/configschema"
+	"github.com/hashicorp/terraform/internal/depsfile"
 	"github.com/hashicorp/terraform/internal/didyoumean"
 	"github.com/hashicorp/terraform/internal/getproviders"
+	"github.com/hashicorp/terraform/internal/getproviders/providerreqs"
 	"github.com/hashicorp/terraform/internal/providercache"
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/tfdiags"
@@ -40,6 +42,8 @@ import (
 // module and clones it to the working directory.
 type InitCommand struct {
 	Meta
+
+	incompleteProviders []string
 }
 
 func (c *InitCommand) Run(args []string) int {
@@ -65,7 +69,7 @@ func (c *InitCommand) Run(args []string) int {
 	}
 	if c.Meta.AllowExperimentalFeatures && initArgs.EnablePssExperiment {
 		// TODO(SarahFrench/radeksimko): Remove forked init logic once feature is no longer experimental
-		panic("This experiment is not available yet")
+		return c.runPssInit(initArgs, view)
 	} else {
 		return c.run(initArgs, view)
 	}
@@ -209,7 +213,7 @@ func (c *InitCommand) initBackend(ctx context.Context, root *configs.Module, ext
 					root.StateStore.ProviderAddr,
 					root.StateStore.Type,
 				),
-				Subject: &root.Backend.TypeRange,
+				Subject: &root.StateStore.TypeRange,
 			})
 			return nil, true, diags
 		}
@@ -358,8 +362,11 @@ the backend configuration is present and valid.
 	return back, true, diags
 }
 
-// Load the complete module tree, and fetch any missing providers.
-// This method outputs its own Ui.
+// getProviders determines what providers are required given configuration and state data. The method downloads any missing providers
+// and replaces the contents of the dependency lock file if any changes happen.
+// The calling code is expected to have loaded the complete module tree and read the state file, and passes that data into this method.
+//
+// This method outputs to the provided view. The returned `output` boolean lets calling code know if anything has been rendered via the view.
 func (c *InitCommand) getProviders(ctx context.Context, config *configs.Config, state *states.State, upgrade bool, pluginDirs []string, flagLockfile string, view views.Init) (output, abort bool, diags tfdiags.Diagnostics) {
 	ctx, span := tracer.Start(ctx, "install providers")
 	defer span.End()
@@ -808,6 +815,577 @@ func (c *InitCommand) getProviders(ctx context.Context, config *configs.Config,
 	return true, false, diags
 }
 
+// getProvidersFromConfig determines what providers are required by the given configuration data.
+// The method downloads any missing providers that aren't already downloaded and then returns
+// dependency lock data based on the configuration.
+// The dependency lock file itself isn't updated here.
+func (c *InitCommand) getProvidersFromConfig(ctx context.Context, config *configs.Config, upgrade bool, pluginDirs []string, flagLockfile string, view views.Init) (output bool, resultingLocks *depsfile.Locks, diags tfdiags.Diagnostics) {
+	ctx, span := tracer.Start(ctx, "install providers from config")
+	defer span.End()
+
+	// Dev overrides cause the result of "terraform init" to be irrelevant for
+	// any overridden providers, so we'll warn about it to avoid later
+	// confusion when Terraform ends up using a different provider than the
+	// lock file called for.
+	diags = diags.Append(c.providerDevOverrideInitWarnings())
+
+	// Collect the provider dependencies from the configuration.
+	reqs, hclDiags := config.ProviderRequirements()
+	diags = diags.Append(hclDiags)
+	if hclDiags.HasErrors() {
+		return false, nil, diags
+	}
+
+	for providerAddr := range reqs {
+		if providerAddr.IsLegacy() {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Invalid legacy provider address",
+				fmt.Sprintf(
+					"This configuration or its associated state refers to the unqualified provider %q.\n\nYou must complete the Terraform 0.13 upgrade process before upgrading to later versions.",
+					providerAddr.Type,
+				),
+			))
+		}
+	}
+	if diags.HasErrors() {
+		return false, nil, diags
+	}
+
+	var inst *providercache.Installer
+	if len(pluginDirs) == 0 {
+		// By default we use a source that looks for providers in all of the
+		// standard locations, possibly customized by the user in CLI config.
+		inst = c.providerInstaller()
+	} else {
+		// If the user passes at least one -plugin-dir then that circumvents
+		// the usual sources and forces Terraform to consult only the given
+		// directories. Anything not available in one of those directories
+		// is not available for installation.
+		source := c.providerCustomLocalDirectorySource(pluginDirs)
+		inst = c.providerInstallerCustomSource(source)
+
+		// The default (or configured) search paths are logged earlier, in provider_source.go
+		// Log that those are being overridden by the `-plugin-dir` command line options
+		log.Println("[DEBUG] init: overriding provider plugin search paths")
+		log.Printf("[DEBUG] will search for provider plugins in %s", pluginDirs)
+	}
+
+	evts := c.prepareInstallerEvents(ctx, reqs, diags, inst, view, views.InitializingProviderPluginFromConfigMessage, views.ReusingPreviousVersionInfo)
+	ctx = evts.OnContext(ctx)
+
+	mode := providercache.InstallNewProvidersOnly
+	if upgrade {
+		if flagLockfile == "readonly" {
+			diags = diags.Append(fmt.Errorf("The -upgrade flag conflicts with -lockfile=readonly."))
+			view.Diagnostics(diags)
+			return true, nil, diags
+		}
+
+		mode = providercache.InstallUpgrades
+	}
+
+	// Previous locks from dep locks file are needed so we don't re-download any providers
+	previousLocks, moreDiags := c.lockedDependencies()
+	diags = diags.Append(moreDiags)
+	if diags.HasErrors() {
+		return false, nil, diags
+	}
+
+	// Determine which required providers are already downloaded, and download any
+	// new providers or newer versions of providers
+	configLocks, err := inst.EnsureProviderVersions(ctx, previousLocks, reqs, mode)
+	if ctx.Err() == context.Canceled {
+		diags = diags.Append(fmt.Errorf("Provider installation was canceled by an interrupt signal."))
+		view.Diagnostics(diags)
+		return true, nil, diags
+	}
+	if err != nil {
+		// The errors captured in "err" should be redundant with what we
+		// received via the InstallerEvents callbacks above, so we'll
+		// just return those as long as we have some.
+		if !diags.HasErrors() {
+			diags = diags.Append(err)
+		}
+
+		return true, nil, diags
+	}
+
+	return true, configLocks, diags
+}
+
+// getProvidersFromState determines what providers are required by the given state data.
+// The method downloads any missing providers that aren't already downloaded and then returns
+// dependency lock data based on the state.
+// The calling code is assumed to have already called getProvidersFromConfig, which is used to
+// supply the configLocks argument.
+// The dependency lock file itself isn't updated here.
+func (c *InitCommand) getProvidersFromState(ctx context.Context, state *states.State, configLocks *depsfile.Locks, upgrade bool, pluginDirs []string, flagLockfile string, view views.Init) (output bool, resultingLocks *depsfile.Locks, diags tfdiags.Diagnostics) {
+	ctx, span := tracer.Start(ctx, "install providers from state")
+	defer span.End()
+
+	// Dev overrides cause the result of "terraform init" to be irrelevant for
+	// any overridden providers, so we'll warn about it to avoid later
+	// confusion when Terraform ends up using a different provider than the
+	// lock file called for.
+	diags = diags.Append(c.providerDevOverrideInitWarnings())
+
+	if state == nil {
+		// if there is no state there are no providers to get
+		return true, depsfile.NewLocks(), nil
+	}
+	reqs := state.ProviderRequirements()
+
+	for providerAddr := range reqs {
+		if providerAddr.IsLegacy() {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Invalid legacy provider address",
+				fmt.Sprintf(
+					"This configuration or its associated state refers to the unqualified provider %q.\n\nYou must complete the Terraform 0.13 upgrade process before upgrading to later versions.",
+					providerAddr.Type,
+				),
+			))
+		}
+	}
+	if diags.HasErrors() {
+		return false, nil, diags
+	}
+
+	// The locks below are used to avoid re-downloading any providers in the
+	// second download step.
+	// We combine any locks from the dependency lock file and locks identified
+	// from the configuration
+	var moreDiags tfdiags.Diagnostics
+	previousLocks, moreDiags := c.lockedDependencies()
+	diags = diags.Append(moreDiags)
+	if diags.HasErrors() {
+		return false, nil, diags
+	}
+	inProgressLocks := c.mergeLockedDependencies(configLocks, previousLocks)
+
+	var inst *providercache.Installer
+	if len(pluginDirs) == 0 {
+		// By default we use a source that looks for providers in all of the
+		// standard locations, possibly customized by the user in CLI config.
+		inst = c.providerInstaller()
+	} else {
+		// If the user passes at least one -plugin-dir then that circumvents
+		// the usual sources and forces Terraform to consult only the given
+		// directories. Anything not available in one of those directories
+		// is not available for installation.
+		source := c.providerCustomLocalDirectorySource(pluginDirs)
+		inst = c.providerInstallerCustomSource(source)
+
+		// The default (or configured) search paths are logged earlier, in provider_source.go
+		// Log that those are being overridden by the `-plugin-dir` command line options
+		log.Println("[DEBUG] init: overriding provider plugin search paths")
+		log.Printf("[DEBUG] will search for provider plugins in %s", pluginDirs)
+	}
+
+	// Because we're currently just streaming a series of events sequentially
+	// into the terminal, we're showing only a subset of the events to keep
+	// things relatively concise. Later it'd be nice to have a progress UI
+	// where statuses update in-place, but we can't do that as long as we
+	// are shimming our vt100 output to the legacy console API on Windows.
+	evts := c.prepareInstallerEvents(ctx, reqs, diags, inst, view, views.InitializingProviderPluginFromStateMessage, views.ReusingVersionIdentifiedFromConfig)
+	ctx = evts.OnContext(ctx)
+
+	mode := providercache.InstallNewProvidersOnly
+
+	// We don't handle upgrade flags here, i.e. what happens at this point in getProvidersFromConfig:
+	// > We cannot upgrade a provider used only by the state, as there are no version constraints in state.
+	//    > Given the overlap between providers in the config and state, using the upgrade mode here
+	//      would remove the effects of version constraints from the config.
+	// > Any validation of CLI flag usage is already done in getProvidersFromConfig
+
+	newLocks, err := inst.EnsureProviderVersions(ctx, inProgressLocks, reqs, mode)
+	if ctx.Err() == context.Canceled {
+		diags = diags.Append(fmt.Errorf("Provider installation was canceled by an interrupt signal."))
+		view.Diagnostics(diags)
+		return true, nil, diags
+	}
+	if err != nil {
+		// The errors captured in "err" should be redundant with what we
+		// received via the InstallerEvents callbacks above, so we'll
+		// just return those as long as we have some.
+		if !diags.HasErrors() {
+			diags = diags.Append(err)
+		}
+
+		return true, nil, diags
+	}
+
+	return true, newLocks, diags
+}
+
+// saveDependencyLockFile overwrites the contents of the dependency lock file.
+// The calling code is expected to provide the previous locks (if any) and the two sets of locks determined from
+// configuration and state data.
+func (c *InitCommand) saveDependencyLockFile(previousLocks, configLocks, stateLocks *depsfile.Locks, flagLockfile string, view views.Init) (output bool, diags tfdiags.Diagnostics) {
+
+	// Get the combination of config and state locks
+	newLocks := c.mergeLockedDependencies(configLocks, stateLocks)
+
+	// If the provider dependencies have changed since the last run then we'll
+	// say a little about that in case the reader wasn't expecting a change.
+	// (When we later integrate module dependencies into the lock file we'll
+	// probably want to refactor this so that we produce one lock-file related
+	// message for all changes together, but this is here for now just because
+	// it's the smallest change relative to what came before it, which was
+	// a hidden JSON file specifically for tracking providers.)
+	if !newLocks.Equal(previousLocks) {
+		// if readonly mode
+		if flagLockfile == "readonly" {
+			// check if required provider dependencies change
+			if !newLocks.EqualProviderAddress(previousLocks) {
+				diags = diags.Append(tfdiags.Sourceless(
+					tfdiags.Error,
+					`Provider dependency changes detected`,
+					`Changes to the required provider dependencies were detected, but the lock file is read-only. To use and record these requirements, run "terraform init" without the "-lockfile=readonly" flag.`,
+				))
+				return output, diags
+			}
+			// suppress updating the file to record any new information it learned,
+			// such as a hash using a new scheme.
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Warning,
+				`Provider lock file not updated`,
+				`Changes to the provider selections were detected, but not saved in the .terraform.lock.hcl file. To record these selections, run "terraform init" without the "-lockfile=readonly" flag.`,
+			))
+			return output, diags
+		}
+		// Jump in here and add a warning if any of the providers are incomplete.
+		if len(c.incompleteProviders) > 0 {
+			// We don't really care about the order here, we just want the
+			// output to be deterministic.
+			sort.Slice(c.incompleteProviders, func(i, j int) bool {
+				return c.incompleteProviders[i] < c.incompleteProviders[j]
+			})
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Warning,
+				incompleteLockFileInformationHeader,
+				fmt.Sprintf(
+					incompleteLockFileInformationBody,
+					strings.Join(c.incompleteProviders, "\n  - "),
+					getproviders.CurrentPlatform.String())))
+		}
+		if previousLocks.Empty() {
+			// A change from empty to non-empty is special because it suggests
+			// we're running "terraform init" for the first time against a
+			// new configuration. In that case we'll take the opportunity to
+			// say a little about what the dependency lock file is, for new
+			// users or those who are upgrading from a previous Terraform
+			// version that didn't have dependency lock files.
+			view.Output(views.LockInfo)
+			output = true
+		} else {
+			view.Output(views.DependenciesLockChangesInfo)
+			output = true
+		}
+		lockFileDiags := c.replaceLockedDependencies(newLocks)
+		diags = diags.Append(lockFileDiags)
+	}
+	return output, diags
+}
+
+// prepareInstallerEvents returns an instance of *providercache.InstallerEvents. This struct defines callback functions that will be executed
+// when a specific type of event occurs during provider installation.
+// The calling code needs to provide a tfdiags.Diagnostics collection, so that provider installation code returns diags to the calling code using closures
+func (c *InitCommand) prepareInstallerEvents(ctx context.Context, reqs providerreqs.Requirements, diags tfdiags.Diagnostics, inst *providercache.Installer, view views.Init, initMsg views.InitMessageCode, reuseMsg views.InitMessageCode) *providercache.InstallerEvents {
+
+	// Because we're currently just streaming a series of events sequentially
+	// into the terminal, we're showing only a subset of the events to keep
+	// things relatively concise. Later it'd be nice to have a progress UI
+	// where statuses update in-place, but we can't do that as long as we
+	// are shimming our vt100 output to the legacy console API on Windows.
+	events := &providercache.InstallerEvents{
+		PendingProviders: func(reqs map[addrs.Provider]getproviders.VersionConstraints) {
+			view.Output(initMsg)
+		},
+		ProviderAlreadyInstalled: func(provider addrs.Provider, selectedVersion getproviders.Version) {
+			view.LogInitMessage(views.ProviderAlreadyInstalledMessage, provider.ForDisplay(), selectedVersion)
+		},
+		BuiltInProviderAvailable: func(provider addrs.Provider) {
+			view.LogInitMessage(views.BuiltInProviderAvailableMessage, provider.ForDisplay())
+		},
+		BuiltInProviderFailure: func(provider addrs.Provider, err error) {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Invalid dependency on built-in provider",
+				fmt.Sprintf("Cannot use %s: %s.", provider.ForDisplay(), err),
+			))
+		},
+		QueryPackagesBegin: func(provider addrs.Provider, versionConstraints getproviders.VersionConstraints, locked bool) {
+			if locked {
+				view.LogInitMessage(reuseMsg, provider.ForDisplay())
+			} else {
+				if len(versionConstraints) > 0 {
+					view.LogInitMessage(views.FindingMatchingVersionMessage, provider.ForDisplay(), getproviders.VersionConstraintsString(versionConstraints))
+				} else {
+					view.LogInitMessage(views.FindingLatestVersionMessage, provider.ForDisplay())
+				}
+			}
+		},
+		LinkFromCacheBegin: func(provider addrs.Provider, version getproviders.Version, cacheRoot string) {
+			view.LogInitMessage(views.UsingProviderFromCacheDirInfo, provider.ForDisplay(), version)
+		},
+		FetchPackageBegin: func(provider addrs.Provider, version getproviders.Version, location getproviders.PackageLocation) {
+			view.LogInitMessage(views.InstallingProviderMessage, provider.ForDisplay(), version)
+		},
+		QueryPackagesFailure: func(provider addrs.Provider, err error) {
+			switch errorTy := err.(type) {
+			case getproviders.ErrProviderNotFound:
+				sources := errorTy.Sources
+				displaySources := make([]string, len(sources))
+				for i, source := range sources {
+					displaySources[i] = fmt.Sprintf("  - %s", source)
+				}
+				diags = diags.Append(tfdiags.Sourceless(
+					tfdiags.Error,
+					"Failed to query available provider packages",
+					fmt.Sprintf("Could not retrieve the list of available versions for provider %s: %s\n\n%s",
+						provider.ForDisplay(), err, strings.Join(displaySources, "\n"),
+					),
+				))
+			case getproviders.ErrRegistryProviderNotKnown:
+				// We might be able to suggest an alternative provider to use
+				// instead of this one.
+				suggestion := fmt.Sprintf("\n\nAll modules should specify their required_providers so that external consumers will get the correct providers when using a module. To see which modules are currently depending on %s, run the following command:\n    terraform providers", provider.ForDisplay())
+				alternative := getproviders.MissingProviderSuggestion(ctx, provider, inst.ProviderSource(), reqs)
+				if alternative != provider {
+					suggestion = fmt.Sprintf(
+						"\n\nDid you intend to use %s? If so, you must specify that source address in each module which requires that provider. To see which modules are currently depending on %s, run the following command:\n    terraform providers",
+						alternative.ForDisplay(), provider.ForDisplay(),
+					)
+				}
+
+				diags = diags.Append(tfdiags.Sourceless(
+					tfdiags.Error,
+					"Failed to query available provider packages",
+					fmt.Sprintf("Could not retrieve the list of available versions for provider %s: %s%s",
+						provider.ForDisplay(), err, suggestion,
+					),
+				))
+			case getproviders.ErrHostNoProviders:
+				switch {
+				case errorTy.Hostname == svchost.Hostname("github.com") && !errorTy.HasOtherVersion:
+					// If a user copies the URL of a GitHub repository into
+					// the source argument and removes the schema to make it
+					// provider-address-shaped then that's one way we can end up
+					// here. We'll use a specialized error message in anticipation
+					// of that mistake. We only do this if github.com isn't a
+					// provider registry, to allow for the (admittedly currently
+					// rather unlikely) possibility that github.com starts being
+					// a real Terraform provider registry in the future.
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						"Invalid provider registry host",
+						fmt.Sprintf("The given source address %q specifies a GitHub repository rather than a Terraform provider. Refer to the documentation of the provider to find the correct source address to use.",
+							provider.String(),
+						),
+					))
+
+				case errorTy.HasOtherVersion:
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						"Invalid provider registry host",
+						fmt.Sprintf("The host %q given in provider source address %q does not offer a Terraform provider registry that is compatible with this Terraform version, but it may be compatible with a different Terraform version.",
+							errorTy.Hostname, provider.String(),
+						),
+					))
+
+				default:
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						"Invalid provider registry host",
+						fmt.Sprintf("The host %q given in provider source address %q does not offer a Terraform provider registry.",
+							errorTy.Hostname, provider.String(),
+						),
+					))
+				}
+
+			case getproviders.ErrRequestCanceled:
+				// We don't attribute cancellation to any particular operation,
+				// but rather just emit a single general message about it at
+				// the end, by checking ctx.Err().
+
+			default:
+				suggestion := fmt.Sprintf("\n\nTo see which modules are currently depending on %s and what versions are specified, run the following command:\n    terraform providers", provider.ForDisplay())
+				diags = diags.Append(tfdiags.Sourceless(
+					tfdiags.Error,
+					"Failed to query available provider packages",
+					fmt.Sprintf("Could not retrieve the list of available versions for provider %s: %s%s",
+						provider.ForDisplay(), err, suggestion,
+					),
+				))
+			}
+
+		},
+		QueryPackagesWarning: func(provider addrs.Provider, warnings []string) {
+			displayWarnings := make([]string, len(warnings))
+			for i, warning := range warnings {
+				displayWarnings[i] = fmt.Sprintf("- %s", warning)
+			}
+
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Warning,
+				"Additional provider information from registry",
+				fmt.Sprintf("The remote registry returned warnings for %s:\n%s",
+					provider.String(),
+					strings.Join(displayWarnings, "\n"),
+				),
+			))
+		},
+		LinkFromCacheFailure: func(provider addrs.Provider, version getproviders.Version, err error) {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Failed to install provider from shared cache",
+				fmt.Sprintf("Error while importing %s v%s from the shared cache directory: %s.", provider.ForDisplay(), version, err),
+			))
+		},
+		FetchPackageFailure: func(provider addrs.Provider, version getproviders.Version, err error) {
+			const summaryIncompatible = "Incompatible provider version"
+			switch err := err.(type) {
+			case getproviders.ErrProtocolNotSupported:
+				closestAvailable := err.Suggestion
+				switch {
+				case closestAvailable == getproviders.UnspecifiedVersion:
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						summaryIncompatible,
+						fmt.Sprintf(errProviderVersionIncompatible, provider.String()),
+					))
+				case version.GreaterThan(closestAvailable):
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						summaryIncompatible,
+						fmt.Sprintf(providerProtocolTooNew, provider.ForDisplay(),
+							version, tfversion.String(), closestAvailable, closestAvailable,
+							getproviders.VersionConstraintsString(reqs[provider]),
+						),
+					))
+				default: // version is less than closestAvailable
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						summaryIncompatible,
+						fmt.Sprintf(providerProtocolTooOld, provider.ForDisplay(),
+							version, tfversion.String(), closestAvailable, closestAvailable,
+							getproviders.VersionConstraintsString(reqs[provider]),
+						),
+					))
+				}
+			case getproviders.ErrPlatformNotSupported:
+				switch {
+				case err.MirrorURL != nil:
+					// If we're installing from a mirror then it may just be
+					// the mirror lacking the package, rather than it being
+					// unavailable from upstream.
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						summaryIncompatible,
+						fmt.Sprintf(
+							"Your chosen provider mirror at %s does not have a %s v%s package available for your current platform, %s.\n\nProvider releases are separate from Terraform CLI releases, so this provider might not support your current platform. Alternatively, the mirror itself might have only a subset of the plugin packages available in the origin registry, at %s.",
+							err.MirrorURL, err.Provider, err.Version, err.Platform,
+							err.Provider.Hostname,
+						),
+					))
+				default:
+					diags = diags.Append(tfdiags.Sourceless(
+						tfdiags.Error,
+						summaryIncompatible,
+						fmt.Sprintf(
+							"Provider %s v%s does not have a package available for your current platform, %s.\n\nProvider releases are separate from Terraform CLI releases, so not all providers are available for all platforms. Other versions of this provider may have different platforms supported.",
+							err.Provider, err.Version, err.Platform,
+						),
+					))
+				}
+
+			case getproviders.ErrRequestCanceled:
+				// We don't attribute cancellation to any particular operation,
+				// but rather just emit a single general message about it at
+				// the end, by checking ctx.Err().
+
+			default:
+				// We can potentially end up in here under cancellation too,
+				// in spite of our getproviders.ErrRequestCanceled case above,
+				// because not all of the outgoing requests we do under the
+				// "fetch package" banner are source metadata requests.
+				// In that case we will emit a redundant error here about
+				// the request being cancelled, but we'll still detect it
+				// as a cancellation after the installer returns and do the
+				// normal cancellation handling.
+
+				diags = diags.Append(tfdiags.Sourceless(
+					tfdiags.Error,
+					"Failed to install provider",
+					fmt.Sprintf("Error while installing %s v%s: %s", provider.ForDisplay(), version, err),
+				))
+			}
+		},
+		FetchPackageSuccess: func(provider addrs.Provider, version getproviders.Version, localDir string, authResult *getproviders.PackageAuthenticationResult) {
+			var keyID string
+			if authResult != nil && authResult.ThirdPartySigned() {
+				keyID = authResult.KeyID
+			}
+			if keyID != "" {
+				keyID = view.PrepareMessage(views.KeyID, keyID)
+			}
+
+			view.LogInitMessage(views.InstalledProviderVersionInfo, provider.ForDisplay(), version, authResult, keyID)
+		},
+		ProvidersLockUpdated: func(provider addrs.Provider, version getproviders.Version, localHashes []getproviders.Hash, signedHashes []getproviders.Hash, priorHashes []getproviders.Hash) {
+			// We're going to use this opportunity to track if we have any
+			// "incomplete" installs of providers. An incomplete install is
+			// when we are only going to write the local hashes into our lock
+			// file which means a `terraform init` command will fail in future
+			// when used on machines of a different architecture.
+			//
+			// We want to print a warning about this.
+
+			if len(signedHashes) > 0 {
+				// If we have any signedHashes hashes then we don't worry - as
+				// we know we retrieved all available hashes for this version
+				// anyway.
+				return
+			}
+
+			// If local hashes and prior hashes are exactly the same then
+			// it means we didn't record any signed hashes previously, and
+			// we know we're not adding any extra in now (because we already
+			// checked the signedHashes), so that's a problem.
+			//
+			// In the actual check here, if we have any priorHashes and those
+			// hashes are not the same as the local hashes then we're going to
+			// accept that this provider has been configured correctly.
+			if len(priorHashes) > 0 && !reflect.DeepEqual(localHashes, priorHashes) {
+				return
+			}
+
+			// Now, either signedHashes is empty, or priorHashes is exactly the
+			// same as our localHashes which means we never retrieved the
+			// signedHashes previously.
+			//
+			// Either way, this is bad. Let's complain/warn.
+			c.incompleteProviders = append(c.incompleteProviders, provider.ForDisplay())
+		},
+		ProvidersFetched: func(authResults map[addrs.Provider]*getproviders.PackageAuthenticationResult) {
+			thirdPartySigned := false
+			for _, authResult := range authResults {
+				if authResult.ThirdPartySigned() {
+					thirdPartySigned = true
+					break
+				}
+			}
+			if thirdPartySigned {
+				view.LogInitMessage(views.PartnerAndCommunityProvidersMessage)
+			}
+		},
+	}
+
+	return events
+}
+
 // backendConfigOverrideBody interprets the raw values of -backend-config
 // arguments into a hcl Body that should override the backend settings given
 // in the configuration.
diff --git a/internal/command/init_run.go b/internal/command/init_run.go
index 5d3e6d172a..1e866bc63c 100644
--- a/internal/command/init_run.go
+++ b/internal/command/init_run.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/hashicorp/hcl/v2"
 	"github.com/hashicorp/terraform/internal/backend"
 	"github.com/hashicorp/terraform/internal/cloud"
 	"github.com/hashicorp/terraform/internal/command/arguments"
@@ -141,6 +142,22 @@ func (c *InitCommand) run(initArgs *arguments.Init, view views.Init) int {
 
 		return 1
 	}
+	if !c.Meta.AllowExperimentalFeatures && rootModEarly.StateStore != nil {
+		// TODO(SarahFrench/radeksimko) - remove when this feature isn't experimental.
+		// This approach for making the feature experimental is required
+		// to let us assert the feature is gated behind an experiment in tests.
+		// See https://github.com/hashicorp/terraform/pull/37350#issuecomment-3168555619
+		diags = diags.Append(earlyConfDiags)
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Unsupported block type",
+			Detail:   "Blocks of type \"state_store\" are not expected here.",
+			Subject:  &rootModEarly.StateStore.TypeRange,
+		})
+		view.Diagnostics(diags)
+
+		return 1
+	}
 
 	var back backend.Backend
 
diff --git a/internal/command/init_run_experiment.go b/internal/command/init_run_experiment.go
new file mode 100644
index 0000000000..b62b40f515
--- /dev/null
+++ b/internal/command/init_run_experiment.go
@@ -0,0 +1,346 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package command
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/internal/cloud"
+	"github.com/hashicorp/terraform/internal/command/arguments"
+	"github.com/hashicorp/terraform/internal/command/views"
+	"github.com/hashicorp/terraform/internal/configs"
+	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
+	"github.com/hashicorp/terraform/internal/tfdiags"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// `runPssInit` is an altered version of the logic in `run` that contains changes
+// related to the PSS project. This is used by the (InitCommand.Run method only if Terraform has
+// experimental features enabled.
+func (c *InitCommand) runPssInit(initArgs *arguments.Init, view views.Init) int {
+	var diags tfdiags.Diagnostics
+
+	c.forceInitCopy = initArgs.ForceInitCopy
+	c.Meta.stateLock = initArgs.StateLock
+	c.Meta.stateLockTimeout = initArgs.StateLockTimeout
+	c.reconfigure = initArgs.Reconfigure
+	c.migrateState = initArgs.MigrateState
+	c.Meta.ignoreRemoteVersion = initArgs.IgnoreRemoteVersion
+	c.Meta.input = initArgs.InputEnabled
+	c.Meta.targetFlags = initArgs.TargetFlags
+	c.Meta.compactWarnings = initArgs.CompactWarnings
+
+	varArgs := initArgs.Vars.All()
+	items := make([]arguments.FlagNameValue, len(varArgs))
+	for i := range varArgs {
+		items[i].Name = varArgs[i].Name
+		items[i].Value = varArgs[i].Value
+	}
+	c.Meta.variableArgs = arguments.FlagNameValueSlice{Items: &items}
+
+	// Copying the state only happens during backend migration, so setting
+	// -force-copy implies -migrate-state
+	if c.forceInitCopy {
+		c.migrateState = true
+	}
+
+	if len(initArgs.PluginPath) > 0 {
+		c.pluginPath = initArgs.PluginPath
+	}
+
+	// Validate the arg count and get the working directory
+	path, err := ModulePath(initArgs.Args)
+	if err != nil {
+		diags = diags.Append(err)
+		view.Diagnostics(diags)
+		return 1
+	}
+
+	if err := c.storePluginPath(c.pluginPath); err != nil {
+		diags = diags.Append(fmt.Errorf("Error saving -plugin-dir to workspace directory: %s", err))
+		view.Diagnostics(diags)
+		return 1
+	}
+
+	// Initialization can be aborted by interruption signals
+	ctx, done := c.InterruptibleContext(c.CommandContext())
+	defer done()
+
+	// This will track whether we outputted anything so that we know whether
+	// to output a newline before the success message
+	var header bool
+
+	if initArgs.FromModule != "" {
+		src := initArgs.FromModule
+
+		empty, err := configs.IsEmptyDir(path, initArgs.TestsDirectory)
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("Error validating destination directory: %s", err))
+			view.Diagnostics(diags)
+			return 1
+		}
+		if !empty {
+			diags = diags.Append(errors.New(strings.TrimSpace(errInitCopyNotEmpty)))
+			view.Diagnostics(diags)
+			return 1
+		}
+
+		view.Output(views.CopyingConfigurationMessage, src)
+		header = true
+
+		hooks := uiModuleInstallHooks{
+			Ui:             c.Ui,
+			ShowLocalPaths: false, // since they are in a weird location for init
+			View:           view,
+		}
+
+		ctx, span := tracer.Start(ctx, "-from-module=...", trace.WithAttributes(
+			attribute.String("module_source", src),
+		))
+
+		initDirFromModuleAbort, initDirFromModuleDiags := c.initDirFromModule(ctx, path, src, hooks)
+		diags = diags.Append(initDirFromModuleDiags)
+		if initDirFromModuleAbort || initDirFromModuleDiags.HasErrors() {
+			view.Diagnostics(diags)
+			span.SetStatus(codes.Error, "module installation failed")
+			span.End()
+			return 1
+		}
+		span.End()
+
+		view.Output(views.EmptyMessage)
+	}
+
+	// If our directory is empty, then we're done. We can't get or set up
+	// the backend with an empty directory.
+	empty, err := configs.IsEmptyDir(path, initArgs.TestsDirectory)
+	if err != nil {
+		diags = diags.Append(fmt.Errorf("Error checking configuration: %s", err))
+		view.Diagnostics(diags)
+		return 1
+	}
+	if empty {
+		view.Output(views.OutputInitEmptyMessage)
+		return 0
+	}
+
+	// Load just the root module to begin backend and module initialization
+	rootModEarly, earlyConfDiags := c.loadSingleModuleWithTests(path, initArgs.TestsDirectory)
+
+	// There may be parsing errors in config loading but these will be shown later _after_
+	// checking for core version requirement errors. Not meeting the version requirement should
+	// be the first error displayed if that is an issue, but other operations are required
+	// before being able to check core version requirements.
+	if rootModEarly == nil {
+		diags = diags.Append(errors.New(view.PrepareMessage(views.InitConfigError)), earlyConfDiags)
+		view.Diagnostics(diags)
+
+		return 1
+	}
+
+	if initArgs.Get {
+		modsOutput, modsAbort, modsDiags := c.getModules(ctx, path, initArgs.TestsDirectory, rootModEarly, initArgs.Upgrade, view)
+		diags = diags.Append(modsDiags)
+		if modsAbort || modsDiags.HasErrors() {
+			view.Diagnostics(diags)
+			return 1
+		}
+		if modsOutput {
+			header = true
+		}
+	}
+
+	// With all of the modules (hopefully) installed, we can now try to load the
+	// whole configuration tree.
+	config, confDiags := c.loadConfigWithTests(path, initArgs.TestsDirectory)
+	// configDiags will be handled after the version constraint check, since an
+	// incorrect version of terraform may be producing errors for configuration
+	// constructs added in later versions.
+
+	// Before we go further, we'll check to make sure none of the modules in
+	// the configuration declare that they don't support this Terraform
+	// version, so we can produce a version-related error message rather than
+	// potentially-confusing downstream errors.
+	versionDiags := terraform.CheckCoreVersionRequirements(config)
+	if versionDiags.HasErrors() {
+		view.Diagnostics(versionDiags)
+		return 1
+	}
+
+	// Now the full configuration is loaded, we can download the providers specified in the configuration.
+	// This is step one of a two-step provider download process
+	// Providers may be downloaded by this code, but the dependency lock file is only updated later in `init`
+	// after step two of provider download is complete.
+	previousLocks, moreDiags := c.lockedDependencies()
+	diags = diags.Append(moreDiags)
+
+	configProvidersOutput, configLocks, configProviderDiags := c.getProvidersFromConfig(ctx, config, initArgs.Upgrade, initArgs.PluginPath, initArgs.Lockfile, view)
+	diags = diags.Append(configProviderDiags)
+	if configProviderDiags.HasErrors() {
+		view.Diagnostics(diags)
+		return 1
+	}
+	if configProvidersOutput {
+		header = true
+	}
+
+	// If we outputted information, then we need to output a newline
+	// so that our success message is nicely spaced out from prior text.
+	if header {
+		view.Output(views.EmptyMessage)
+	}
+
+	var back backend.Backend
+
+	var backDiags tfdiags.Diagnostics
+	var backendOutput bool
+	switch {
+	case initArgs.Cloud && rootModEarly.CloudConfig != nil:
+		back, backendOutput, backDiags = c.initCloud(ctx, rootModEarly, initArgs.BackendConfig, initArgs.ViewType, view)
+	case initArgs.Backend:
+		// TODO(SarahFrench/radeksimko) - pass information about config locks (`configLocks`) into initBackend to
+		// enable PSS
+		back, backendOutput, backDiags = c.initBackend(ctx, rootModEarly, initArgs.BackendConfig, initArgs.ViewType, view)
+	default:
+		// load the previously-stored backend config
+		back, backDiags = c.Meta.backendFromState(ctx)
+	}
+	if backendOutput {
+		header = true
+	}
+	if header {
+		// If we outputted information, then we need to output a newline
+		// so that our success message is nicely spaced out from prior text.
+		view.Output(views.EmptyMessage)
+	}
+
+	var state *states.State
+
+	// If we have a functional backend (either just initialized or initialized
+	// on a previous run) we'll use the current state as a potential source
+	// of provider dependencies.
+	if back != nil {
+		c.ignoreRemoteVersionConflict(back)
+		workspace, err := c.Workspace()
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("Error selecting workspace: %s", err))
+			view.Diagnostics(diags)
+			return 1
+		}
+		sMgr, err := back.StateMgr(workspace)
+		if err != nil {
+			diags = diags.Append(fmt.Errorf("Error loading state: %s", err))
+			view.Diagnostics(diags)
+			return 1
+		}
+
+		if err := sMgr.RefreshState(); err != nil {
+			diags = diags.Append(fmt.Errorf("Error refreshing state: %s", err))
+			view.Diagnostics(diags)
+			return 1
+		}
+
+		state = sMgr.State()
+	}
+
+	// Now the resource state is loaded, we can download the providers specified in the state but not the configuration.
+	// This is step two of a two-step provider download process
+	stateProvidersOutput, stateLocks, stateProvidersDiags := c.getProvidersFromState(ctx, state, configLocks, initArgs.Upgrade, initArgs.PluginPath, initArgs.Lockfile, view)
+	diags = diags.Append(configProviderDiags)
+	if stateProvidersDiags.HasErrors() {
+		view.Diagnostics(diags)
+		return 1
+	}
+	if stateProvidersOutput {
+		header = true
+	}
+	if header {
+		// If we outputted information, then we need to output a newline
+		// so that our success message is nicely spaced out from prior text.
+		view.Output(views.EmptyMessage)
+	}
+
+	// Now the two steps of provider download have happened, update the dependency lock file if it has changed.
+	lockFileOutput, lockFileDiags := c.saveDependencyLockFile(previousLocks, configLocks, stateLocks, initArgs.Lockfile, view)
+	diags = diags.Append(lockFileDiags)
+	if lockFileDiags.HasErrors() {
+		view.Diagnostics(diags)
+		return 1
+	}
+	if lockFileOutput {
+		header = true
+	}
+	if header {
+		// If we outputted information, then we need to output a newline
+		// so that our success message is nicely spaced out from prior text.
+		view.Output(views.EmptyMessage)
+	}
+
+	// As Terraform version-related diagnostics are handled above, we can now
+	// check the diagnostics from the early configuration and the backend.
+	diags = diags.Append(earlyConfDiags)
+	diags = diags.Append(backDiags)
+	if earlyConfDiags.HasErrors() {
+		diags = diags.Append(errors.New(view.PrepareMessage(views.InitConfigError)))
+		view.Diagnostics(diags)
+		return 1
+	}
+
+	// Now, we can show any errors from initializing the backend, but we won't
+	// show the InitConfigError preamble as we didn't detect problems with
+	// the early configuration.
+	if backDiags.HasErrors() {
+		view.Diagnostics(diags)
+		return 1
+	}
+
+	// If everything is ok with the core version check and backend initialization,
+	// show other errors from loading the full configuration tree.
+	diags = diags.Append(confDiags)
+	if confDiags.HasErrors() {
+		diags = diags.Append(errors.New(view.PrepareMessage(views.InitConfigError)))
+		view.Diagnostics(diags)
+		return 1
+	}
+
+	if cb, ok := back.(*cloud.Cloud); ok {
+		if c.RunningInAutomation {
+			if err := cb.AssertImportCompatible(config); err != nil {
+				diags = diags.Append(tfdiags.Sourceless(tfdiags.Error, "Compatibility error", err.Error()))
+				view.Diagnostics(diags)
+				return 1
+			}
+		}
+	}
+
+	// If we accumulated any warnings along the way that weren't accompanied
+	// by errors then we'll output them here so that the success message is
+	// still the final thing shown.
+	view.Diagnostics(diags)
+	_, cloud := back.(*cloud.Cloud)
+	output := views.OutputInitSuccessMessage
+	if cloud {
+		output = views.OutputInitSuccessCloudMessage
+	}
+
+	view.Output(output)
+
+	if !c.RunningInAutomation {
+		// If we're not running in an automation wrapper, give the user
+		// some more detailed next steps that are appropriate for interactive
+		// shell usage.
+		output = views.OutputInitSuccessCLIMessage
+		if cloud {
+			output = views.OutputInitSuccessCLICloudMessage
+		}
+		view.Output(output)
+	}
+	return 0
+}
diff --git a/internal/command/init_test.go b/internal/command/init_test.go
index d243344f52..9de0c76f8e 100644
--- a/internal/command/init_test.go
+++ b/internal/command/init_test.go
@@ -111,6 +111,130 @@ func TestInit_only_test_files(t *testing.T) {
 	}
 }
 
+func TestInit_two_step_provider_download(t *testing.T) {
+	cases := map[string]struct {
+		workDirPath          string
+		flags                []string
+		expectedDownloadMsgs []string
+	}{
+		"providers required by only the state file": {
+			// TODO - should the output indicate that no providers were found in config?
+			workDirPath: "init-provider-download/state-file-only",
+			expectedDownloadMsgs: []string{
+				views.MessageRegistry[views.OutputInitSuccessCLIMessage].JSONValue,
+				`Initializing provider plugins found in the configuration...
+				Initializing the backend...`, // No providers found in the configuration so next output is backend-related
+				`Initializing provider plugins found in the state...
+				- Finding latest version of hashicorp/random...
+				- Installing hashicorp/random v9.9.9...`, // The latest version is expected, as state has no version constraints
+			},
+		},
+		"different providers required by config and state": {
+			workDirPath: "init-provider-download/config-and-state-different-providers",
+			expectedDownloadMsgs: []string{
+				views.MessageRegistry[views.OutputInitSuccessCLIMessage].JSONValue,
+
+				// Config - this provider is affected by a version constraint
+				`Initializing provider plugins found in the configuration...
+				- Finding hashicorp/null versions matching "< 9.0.0"...
+				- Installing hashicorp/null v1.0.0...
+				- Installed hashicorp/null v1.0.0`,
+
+				// State - the latest version of this provider is expected, as state has no version constraints
+				`Initializing provider plugins found in the state...
+				- Finding latest version of hashicorp/random...
+				- Installing hashicorp/random v9.9.9...`,
+			},
+		},
+		"does not re-download providers that are present in both config and state": {
+			workDirPath: "init-provider-download/config-and-state-same-providers",
+			expectedDownloadMsgs: []string{
+				// Config
+				`Initializing provider plugins found in the configuration...
+				- Finding hashicorp/random versions matching "< 9.0.0"...
+				- Installing hashicorp/random v1.0.0...
+				- Installed hashicorp/random v1.0.0`,
+				// State
+				`Initializing provider plugins found in the state...
+				- Reusing previous version of hashicorp/random
+				- Using previously-installed hashicorp/random v1.0.0`,
+			},
+		},
+		"reuses providers already represented in a dependency lock file": {
+			workDirPath: "init-provider-download/config-state-file-and-lockfile",
+			expectedDownloadMsgs: []string{
+				// Config
+				`Initializing provider plugins found in the configuration...
+				- Reusing previous version of hashicorp/random from the dependency lock file
+				- Installing hashicorp/random v1.0.0...
+				- Installed hashicorp/random v1.0.0`,
+				// State
+				`Initializing provider plugins found in the state...
+				- Reusing previous version of hashicorp/random
+				- Using previously-installed hashicorp/random v1.0.0`,
+			},
+		},
+		"using the -upgrade flag causes provider download to ignore the lock file": {
+			workDirPath: "init-provider-download/config-state-file-and-lockfile",
+			flags:       []string{"-upgrade"},
+			expectedDownloadMsgs: []string{
+				// Config - lock file is not mentioned due to the -upgrade flag
+				`Initializing provider plugins found in the configuration...
+				- Finding hashicorp/random versions matching "< 9.0.0"...
+				- Installing hashicorp/random v1.0.0...
+				- Installed hashicorp/random v1.0.0`,
+				// State - reuses the provider download from the config
+				`Initializing provider plugins found in the state...
+				- Reusing previous version of hashicorp/random
+				- Using previously-installed hashicorp/random v1.0.0`,
+			},
+		},
+	}
+
+	for tn, tc := range cases {
+		t.Run(tn, func(t *testing.T) {
+
+			// Create a temporary working directory no tf configuration but has state
+			td := t.TempDir()
+			testCopyDir(t, testFixturePath(tc.workDirPath), td)
+			os.MkdirAll(td, 0755)
+			t.Chdir(td)
+
+			// A provider source containing the random and null providers
+			providerSource, close := newMockProviderSource(t, map[string][]string{
+				"hashicorp/random": {"1.0.0", "9.9.9"},
+				"hashicorp/null":   {"1.0.0", "9.9.9"},
+			})
+			defer close()
+
+			ui := new(cli.MockUi)
+			view, done := testView(t)
+			c := &InitCommand{
+				Meta: Meta{
+					testingOverrides: metaOverridesForProvider(testProvider()),
+					Ui:               ui,
+					View:             view,
+					ProviderSource:   providerSource,
+
+					AllowExperimentalFeatures: true, // Needed to test init changes for PSS project
+				},
+			}
+
+			args := append(tc.flags, "-enable-pluggable-state-storage-experiment") // Needed to test init changes for PSS project
+			if code := c.Run(args); code != 0 {
+				t.Fatalf("bad: \n%s", done(t).All())
+			}
+
+			actual := cleanString(done(t).All())
+			for _, downloadMsg := range tc.expectedDownloadMsgs {
+				if !strings.Contains(cleanString(actual), cleanString(downloadMsg)) {
+					t.Fatalf("expected output to contain %q\n, got %q", cleanString(downloadMsg), cleanString(actual))
+				}
+			}
+		})
+	}
+}
+
 func TestInit_multipleArgs(t *testing.T) {
 	// Create a temporary working directory that is empty
 	td := t.TempDir()
diff --git a/internal/command/meta_dependencies.go b/internal/command/meta_dependencies.go
index e7f213e43d..b84afecf3a 100644
--- a/internal/command/meta_dependencies.go
+++ b/internal/command/meta_dependencies.go
@@ -66,6 +66,43 @@ func (m *Meta) replaceLockedDependencies(new *depsfile.Locks) tfdiags.Diagnostic
 	return depsfile.SaveLocksToFile(new, dependencyLockFilename)
 }
 
+// mergeLockedDependencies combines two sets of locks. The 'base' locks are copied, and any providers
+// present in the additional locks that aren't present in the base are added to that copy. The merged
+// combination is returned.
+//
+// If you're combining locks derived from config with other locks (from state or deps locks file), then
+// the config locks need to be the first argument to ensure that the merged locks contain any
+// version constraints. Version constraint data is only present in configuration.
+// This allows code in the init command to download providers in separate phases and
+// keep the lock file updated accurately after each phase.
+//
+// This method supports downloading providers in 2 steps, and is used during the second download step and
+// while updating the dependency lock file.
+func (m *Meta) mergeLockedDependencies(baseLocks, additionalLocks *depsfile.Locks) *depsfile.Locks {
+
+	mergedLocks := baseLocks.DeepCopy()
+
+	// Append locks derived from the state to locks derived from config.
+	for _, lock := range additionalLocks.AllProviders() {
+		match := mergedLocks.Provider(lock.Provider())
+		if match != nil {
+			log.Printf("[TRACE] Ignoring provider %s version %s in appendLockedDependencies; lock file contains %s version %s already",
+				lock.Provider(),
+				lock.Version(),
+				match.Provider(),
+				match.Version(),
+			)
+		} else {
+			// This is a new provider now present in the lockfile yet
+			log.Printf("[DEBUG] Appending provider %s to the lock file", lock.Provider())
+			mergedLocks.SetProvider(lock.Provider(), lock.Version(), lock.VersionConstraints(), lock.AllHashes())
+		}
+	}
+
+	// Override the locks file with the new combination of locks
+	return mergedLocks
+}
+
 // annotateDependencyLocksWithOverrides modifies the given Locks object in-place
 // to track as overridden any provider address that's subject to testing
 // overrides, development overrides, or "unmanaged provider" status.
diff --git a/internal/command/meta_dependencies_test.go b/internal/command/meta_dependencies_test.go
new file mode 100644
index 0000000000..b245648867
--- /dev/null
+++ b/internal/command/meta_dependencies_test.go
@@ -0,0 +1,151 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package command
+
+import (
+	"testing"
+
+	"github.com/apparentlymart/go-versions/versions"
+	"github.com/google/go-cmp/cmp"
+	"github.com/hashicorp/cli"
+	tfaddr "github.com/hashicorp/terraform-registry-address"
+	"github.com/hashicorp/terraform/internal/depsfile"
+	"github.com/hashicorp/terraform/internal/getproviders/providerreqs"
+)
+
+// This tests combining locks from config and state. Locks derived from state are always unconstrained, i.e. no version constraint data,
+// so this test
+func Test_mergeLockedDependencies_config_and_state(t *testing.T) {
+
+	providerA := tfaddr.NewProvider(tfaddr.DefaultProviderRegistryHost, "my-org", "providerA")
+	providerB := tfaddr.NewProvider(tfaddr.DefaultProviderRegistryHost, "my-org", "providerB")
+	v1_0_0 := providerreqs.MustParseVersion("1.0.0")
+	versionConstraintv1, _ := providerreqs.ParseVersionConstraints("1.0.0")
+	hashesProviderA := []providerreqs.Hash{providerreqs.MustParseHash("providerA:this-is-providerA")}
+	hashesProviderB := []providerreqs.Hash{providerreqs.MustParseHash("providerB:this-is-providerB")}
+
+	var versionUnconstrained providerreqs.VersionConstraints = nil
+	noVersion := versions.Version{}
+
+	cases := map[string]struct {
+		configLocks   *depsfile.Locks
+		stateLocks    *depsfile.Locks
+		expectedLocks *depsfile.Locks
+	}{
+		"no locks when all inputs empty": {
+			configLocks:   depsfile.NewLocks(),
+			stateLocks:    depsfile.NewLocks(),
+			expectedLocks: depsfile.NewLocks(),
+		},
+		"when provider only described in config, output locks have matching constraints": {
+			configLocks: func() *depsfile.Locks {
+				configLocks := depsfile.NewLocks()
+				configLocks.SetProvider(providerA, v1_0_0, versionConstraintv1, hashesProviderA)
+				return configLocks
+			}(),
+			stateLocks: depsfile.NewLocks(),
+			expectedLocks: func() *depsfile.Locks {
+				combinedLocks := depsfile.NewLocks()
+				combinedLocks.SetProvider(providerA, v1_0_0, versionConstraintv1, hashesProviderA)
+				return combinedLocks
+			}(),
+		},
+		"when provider only described in state, output locks are unconstrained": {
+			configLocks: depsfile.NewLocks(),
+			stateLocks: func() *depsfile.Locks {
+				stateLocks := depsfile.NewLocks()
+				stateLocks.SetProvider(providerA, noVersion, versionUnconstrained, hashesProviderA)
+				return stateLocks
+			}(),
+			expectedLocks: func() *depsfile.Locks {
+				combinedLocks := depsfile.NewLocks()
+				combinedLocks.SetProvider(providerA, noVersion, versionUnconstrained, hashesProviderA)
+				return combinedLocks
+			}(),
+		},
+		"different providers present in state and config are combined, with version constraints kept on config providers": {
+			configLocks: func() *depsfile.Locks {
+				configLocks := depsfile.NewLocks()
+				configLocks.SetProvider(providerA, v1_0_0, versionConstraintv1, hashesProviderA)
+				return configLocks
+			}(),
+			stateLocks: func() *depsfile.Locks {
+				stateLocks := depsfile.NewLocks()
+
+				// Imagine that the state locks contain:
+				// 1) provider for resources in the config
+				stateLocks.SetProvider(providerA, noVersion, versionUnconstrained, hashesProviderA)
+
+				// 2) also, a provider that's deleted from the config and only present in state
+				stateLocks.SetProvider(providerB, noVersion, versionUnconstrained, hashesProviderB)
+
+				return stateLocks
+			}(),
+			expectedLocks: func() *depsfile.Locks {
+				combinedLocks := depsfile.NewLocks()
+				combinedLocks.SetProvider(providerA, v1_0_0, versionConstraintv1, hashesProviderA)     // version constraint preserved
+				combinedLocks.SetProvider(providerB, noVersion, versionUnconstrained, hashesProviderB) // sourced from state only
+				return combinedLocks
+			}(),
+		},
+	}
+
+	for tn, tc := range cases {
+		t.Run(tn, func(t *testing.T) {
+			// Use tmp dir as we're creating lock files in the test
+			td := t.TempDir()
+			t.Chdir(td)
+
+			ui := new(cli.MockUi)
+			view, _ := testView(t)
+
+			m := Meta{
+				Ui:   ui,
+				View: view,
+			}
+
+			// Code under test - combine deps from state with prior deps from config
+			// mergedLocks := m.mergeLockedDependencies(tc.configLocks, tc.stateLocks)
+			mergedLocks := m.mergeLockedDependencies(tc.configLocks, tc.stateLocks)
+
+			// We cannot use (l *depsfile.Locks) Equal here as it doesn't compare version constraints
+			// Instead, inspect entries directly
+			if len(mergedLocks.AllProviders()) != len(tc.expectedLocks.AllProviders()) {
+				t.Fatalf("expected merged dependencies to include %d providers, but got %d:\n %#v",
+					len(tc.expectedLocks.AllProviders()),
+					len(mergedLocks.AllProviders()),
+					mergedLocks.AllProviders(),
+				)
+			}
+			for _, lock := range tc.expectedLocks.AllProviders() {
+				match := mergedLocks.Provider(lock.Provider())
+				if match == nil {
+					t.Fatalf("expected merged dependencies to include provider %s, but it's missing", lock.Provider())
+				}
+				if len(match.VersionConstraints()) != len(lock.VersionConstraints()) {
+					t.Fatalf("detected a problem with version constraints for provider %s, got: %d, want %d",
+						lock.Provider(),
+						len(match.VersionConstraints()),
+						len(lock.VersionConstraints()),
+					)
+				}
+				if len(match.VersionConstraints()) > 0 && len(lock.VersionConstraints()) > 0 {
+					gotConstraints := match.VersionConstraints()[0]
+					wantConstraints := lock.VersionConstraints()[0]
+
+					if gotConstraints.Boundary.String() != wantConstraints.Boundary.String() {
+						t.Fatalf("expected merged dependencies to include provider %s with version constraint %v, but instead got %v",
+							lock.Provider(),
+							gotConstraints.Boundary.String(),
+							wantConstraints.Boundary.String(),
+						)
+					}
+				}
+			}
+			if diff := cmp.Diff(tc.expectedLocks, mergedLocks); diff != "" {
+				t.Errorf("difference in file contents detected\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/internal/command/testdata/init-provider-download/config-and-state-different-providers/main.tf b/internal/command/testdata/init-provider-download/config-and-state-different-providers/main.tf
new file mode 100644
index 0000000000..f0687030bb
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/config-and-state-different-providers/main.tf
@@ -0,0 +1,14 @@
+terraform {
+  required_providers {
+    null = {
+      source  = "hashicorp/null"
+      version = "<9.0.0"
+    }
+  }
+
+  backend "local" {
+    path = "./state-using-random-provider.tfstate"
+  }
+}
+
+resource "null_resource" "null" {}
diff --git a/internal/command/testdata/init-provider-download/config-and-state-different-providers/state-using-random-provider.tfstate b/internal/command/testdata/init-provider-download/config-and-state-different-providers/state-using-random-provider.tfstate
new file mode 100644
index 0000000000..76b543b06a
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/config-and-state-different-providers/state-using-random-provider.tfstate
@@ -0,0 +1,28 @@
+{
+    "version": 4,
+    "terraform_version": "1.14.0",
+    "serial": 1,
+    "lineage": "foobar",
+    "resources": [
+        {
+            "mode": "managed",
+            "type": "random_pet",
+            "name": "maurice",
+            "provider": "provider[\"registry.terraform.io/hashicorp/random\"]",
+            "instances": [
+                {
+                    "schema_version": 0,
+                    "attributes": {
+                        "id": "sassy-ferret",
+                        "keepers": null,
+                        "length": 2,
+                        "prefix": null,
+                        "separator": "-"
+                    },
+                    "sensitive_attributes": [],
+                    "identity_schema_version": 0
+                }
+            ]
+        }
+    ]
+}
\ No newline at end of file
diff --git a/internal/command/testdata/init-provider-download/config-and-state-same-providers/main.tf b/internal/command/testdata/init-provider-download/config-and-state-same-providers/main.tf
new file mode 100644
index 0000000000..d5857489d6
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/config-and-state-same-providers/main.tf
@@ -0,0 +1,14 @@
+terraform {
+  required_providers {
+    random = {
+      source  = "hashicorp/random"
+      version = "<9.0.0"
+    }
+  }
+
+  backend "local" {
+    path = "./state-using-random-provider.tfstate"
+  }
+}
+
+resource "random_pet" "maurice" {}
diff --git a/internal/command/testdata/init-provider-download/config-and-state-same-providers/state-using-random-provider.tfstate b/internal/command/testdata/init-provider-download/config-and-state-same-providers/state-using-random-provider.tfstate
new file mode 100644
index 0000000000..76b543b06a
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/config-and-state-same-providers/state-using-random-provider.tfstate
@@ -0,0 +1,28 @@
+{
+    "version": 4,
+    "terraform_version": "1.14.0",
+    "serial": 1,
+    "lineage": "foobar",
+    "resources": [
+        {
+            "mode": "managed",
+            "type": "random_pet",
+            "name": "maurice",
+            "provider": "provider[\"registry.terraform.io/hashicorp/random\"]",
+            "instances": [
+                {
+                    "schema_version": 0,
+                    "attributes": {
+                        "id": "sassy-ferret",
+                        "keepers": null,
+                        "length": 2,
+                        "prefix": null,
+                        "separator": "-"
+                    },
+                    "sensitive_attributes": [],
+                    "identity_schema_version": 0
+                }
+            ]
+        }
+    ]
+}
\ No newline at end of file
diff --git a/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/.terraform.lock.hcl b/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/.terraform.lock.hcl
new file mode 100644
index 0000000000..2140e1a144
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/.terraform.lock.hcl
@@ -0,0 +1,6 @@
+# This file is maintained automatically by "terraform init".
+# Manual edits may be lost in future updates.
+
+provider "registry.terraform.io/hashicorp/random" {
+  version = "1.0.0"
+}
diff --git a/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/main.tf b/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/main.tf
new file mode 100644
index 0000000000..d5857489d6
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/main.tf
@@ -0,0 +1,14 @@
+terraform {
+  required_providers {
+    random = {
+      source  = "hashicorp/random"
+      version = "<9.0.0"
+    }
+  }
+
+  backend "local" {
+    path = "./state-using-random-provider.tfstate"
+  }
+}
+
+resource "random_pet" "maurice" {}
diff --git a/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/state-using-random-provider.tfstate b/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/state-using-random-provider.tfstate
new file mode 100644
index 0000000000..76b543b06a
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/config-state-file-and-lockfile/state-using-random-provider.tfstate
@@ -0,0 +1,28 @@
+{
+    "version": 4,
+    "terraform_version": "1.14.0",
+    "serial": 1,
+    "lineage": "foobar",
+    "resources": [
+        {
+            "mode": "managed",
+            "type": "random_pet",
+            "name": "maurice",
+            "provider": "provider[\"registry.terraform.io/hashicorp/random\"]",
+            "instances": [
+                {
+                    "schema_version": 0,
+                    "attributes": {
+                        "id": "sassy-ferret",
+                        "keepers": null,
+                        "length": 2,
+                        "prefix": null,
+                        "separator": "-"
+                    },
+                    "sensitive_attributes": [],
+                    "identity_schema_version": 0
+                }
+            ]
+        }
+    ]
+}
\ No newline at end of file
diff --git a/internal/command/testdata/init-provider-download/state-file-only/main.tf b/internal/command/testdata/init-provider-download/state-file-only/main.tf
new file mode 100644
index 0000000000..d568a66bf2
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/state-file-only/main.tf
@@ -0,0 +1,5 @@
+terraform {
+  backend "local" {
+    path = "./state-using-random-provider.tfstate"
+  }
+}
diff --git a/internal/command/testdata/init-provider-download/state-file-only/state-using-random-provider.tfstate b/internal/command/testdata/init-provider-download/state-file-only/state-using-random-provider.tfstate
new file mode 100644
index 0000000000..76b543b06a
--- /dev/null
+++ b/internal/command/testdata/init-provider-download/state-file-only/state-using-random-provider.tfstate
@@ -0,0 +1,28 @@
+{
+    "version": 4,
+    "terraform_version": "1.14.0",
+    "serial": 1,
+    "lineage": "foobar",
+    "resources": [
+        {
+            "mode": "managed",
+            "type": "random_pet",
+            "name": "maurice",
+            "provider": "provider[\"registry.terraform.io/hashicorp/random\"]",
+            "instances": [
+                {
+                    "schema_version": 0,
+                    "attributes": {
+                        "id": "sassy-ferret",
+                        "keepers": null,
+                        "length": 2,
+                        "prefix": null,
+                        "separator": "-"
+                    },
+                    "sensitive_attributes": [],
+                    "identity_schema_version": 0
+                }
+            ]
+        }
+    ]
+}
\ No newline at end of file
diff --git a/internal/command/testdata/init-with-state-store/main.tf b/internal/command/testdata/init-with-state-store/main.tf
index a4042b401b..9939e9dece 100644
--- a/internal/command/testdata/init-with-state-store/main.tf
+++ b/internal/command/testdata/init-with-state-store/main.tf
@@ -1,5 +1,11 @@
 terraform {
-  state_store "foo_foo" {
 
+  required_providers {
+    foo = {
+      source = "my-org/foo"
+    }
+  }
+  state_store "foo_foo" {
+    provider "foo" {}
   }
 }
diff --git a/internal/command/views/init.go b/internal/command/views/init.go
index d228cb81d5..35d79e1ba8 100644
--- a/internal/command/views/init.go
+++ b/internal/command/views/init.go
@@ -186,6 +186,14 @@ var MessageRegistry map[InitMessageCode]InitMessage = map[InitMessageCode]InitMe
 		HumanValue: "\n[reset][bold]Initializing provider plugins...",
 		JSONValue:  "Initializing provider plugins...",
 	},
+	"initializing_provider_plugin_from_config_message": {
+		HumanValue: "\n[reset][bold]Initializing provider plugins found in the configuration...",
+		JSONValue:  "Initializing provider plugins found in the configuration...",
+	},
+	"initializing_provider_plugin_from_state_message": {
+		HumanValue: "\n[reset][bold]Initializing provider plugins found in the state...",
+		JSONValue:  "Initializing provider plugins found in the state...",
+	},
 	"initializing_state_store_message": {
 		HumanValue: "\n[reset][bold]Initializing the state store...",
 		JSONValue:  "Initializing the state store...",
@@ -210,6 +218,10 @@ var MessageRegistry map[InitMessageCode]InitMessage = map[InitMessageCode]InitMe
 		HumanValue: "- Reusing previous version of %s from the dependency lock file",
 		JSONValue:  "%s: Reusing previous version from the dependency lock file",
 	},
+	"reusing_version_during_state_provider_init": {
+		HumanValue: "- Reusing previous version of %s",
+		JSONValue:  "%s: Reusing previous version of %s",
+	},
 	"finding_matching_version_message": {
 		HumanValue: "- Finding %s versions matching %q...",
 		JSONValue:  "Finding matching versions for provider: %s, version_constraint: %q",
@@ -271,6 +283,14 @@ const (
 	DependenciesLockChangesInfo       InitMessageCode = "dependencies_lock_changes_info"
 
 	//// Message codes below are ONLY used INTERNALLY (for now)
+
+	// InitializingProviderPluginFromConfigMessage indicates the beginning of installing of providers described in configuration
+	InitializingProviderPluginFromConfigMessage InitMessageCode = "initializing_provider_plugin_from_config_message"
+	// InitializingProviderPluginFromStateMessage indicates the beginning of installing of providers described in state
+	InitializingProviderPluginFromStateMessage InitMessageCode = "initializing_provider_plugin_from_state_message"
+	// DependenciesLockPendingChangesInfo indicates when a provider installation step will reuse a provider from a previous installation step in the current operation
+	ReusingVersionIdentifiedFromConfig InitMessageCode = "reusing_version_during_state_provider_init"
+
 	// InitConfigError indicates problems encountered during initialisation
 	InitConfigError InitMessageCode = "init_config_error"
 	// FindingMatchingVersionMessage indicates that Terraform is looking for a provider version that matches the constraint during installation
diff --git a/internal/configs/parser_config.go b/internal/configs/parser_config.go
index fa72ba7f9c..4047fe1db8 100644
--- a/internal/configs/parser_config.go
+++ b/internal/configs/parser_config.go
@@ -104,14 +104,6 @@ func parseConfigFile(body hcl.Body, diags hcl.Diagnostics, override, allowExperi
 		switch block.Type {
 
 		case "terraform":
-			// TODO: Update once pluggable state store is out of experimental phase
-			if allowExperiments {
-				terraformBlockSchema.Blocks = append(terraformBlockSchema.Blocks,
-					hcl.BlockHeaderSchema{
-						Type:       "state_store",
-						LabelNames: []string{"type"},
-					})
-			}
 			content, contentDiags := block.Body.Content(terraformBlockSchema)
 			diags = append(diags, contentDiags...)
 
@@ -394,9 +386,10 @@ var terraformBlockSchema = &hcl.BodySchema{
 		{
 			Type: "required_providers",
 		},
-		// NOTE: An entry for state_store is not present here
-		// because we conditionally add it in the calling code
-		// depending on whether experiments are enabled or not.
+		{
+			Type:       "state_store",
+			LabelNames: []string{"type"},
+		},
 		{
 			Type:       "provider_meta",
 			LabelNames: []string{"provider"},