diff --git a/go.mod b/go.mod
index 3407e47df1d7..e66e875b56ba 100644
--- a/go.mod
+++ b/go.mod
@@ -40,7 +40,7 @@ require (
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-plugin v1.4.3
 	github.com/hashicorp/go-retryablehttp v0.7.1
-	github.com/hashicorp/go-tfe v1.7.0
+	github.com/hashicorp/go-tfe v1.9.0
 	github.com/hashicorp/go-uuid v1.0.3
 	github.com/hashicorp/go-version v1.6.0
 	github.com/hashicorp/hcl v0.0.0-20170504190234-a4b07c25de5f
@@ -144,7 +144,7 @@ require (
 	github.com/hashicorp/go-msgpack v0.5.4 // indirect
 	github.com/hashicorp/go-rootcerts v1.0.2 // indirect
 	github.com/hashicorp/go-safetemp v1.0.0 // indirect
-	github.com/hashicorp/go-slug v0.9.1 // indirect
+	github.com/hashicorp/go-slug v0.10.0 // indirect
 	github.com/hashicorp/golang-lru v0.5.1 // indirect
 	github.com/hashicorp/jsonapi v0.0.0-20210826224640-ee7dae0fb22d // indirect
 	github.com/hashicorp/serf v0.9.5 // indirect
diff --git a/go.sum b/go.sum
index 204fd15ee384..3a2c9803f62e 100644
--- a/go.sum
+++ b/go.sum
@@ -370,13 +370,13 @@ github.com/hashicorp/go-rootcerts v1.0.2 h1:jzhAVGtqPKbwpyCPELlgNWhE1znq+qwJtW5O
 github.com/hashicorp/go-rootcerts v1.0.2/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8=
 github.com/hashicorp/go-safetemp v1.0.0 h1:2HR189eFNrjHQyENnQMMpCiBAsRxzbTMIgBhEyExpmo=
 github.com/hashicorp/go-safetemp v1.0.0/go.mod h1:oaerMy3BhqiTbVye6QuFhFtIceqFoDHxNAB65b+Rj1I=
-github.com/hashicorp/go-slug v0.9.1 h1:gYNVJ3t0jAWx8AT2eYZci3Xd7NBHyjayW9AR1DU4ki0=
-github.com/hashicorp/go-slug v0.9.1/go.mod h1:Ib+IWBYfEfJGI1ZyXMGNbu2BU+aa3Dzu41RKLH301v4=
+github.com/hashicorp/go-slug v0.10.0 h1:mh4DDkBJTh9BuEjY/cv8PTo7k9OjT4PcW8PgZnJ4jTY=
+github.com/hashicorp/go-slug v0.10.0/go.mod h1:Ib+IWBYfEfJGI1ZyXMGNbu2BU+aa3Dzu41RKLH301v4=
 github.com/hashicorp/go-sockaddr v1.0.0 h1:GeH6tui99pF4NJgfnhp+L6+FfobzVW3Ah46sLo0ICXs=
 github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
 github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
-github.com/hashicorp/go-tfe v1.7.0 h1:GELRhS5dizF6giwjZBqUC/xPaSuNYB+hWRtUnf6i8K8=
-github.com/hashicorp/go-tfe v1.7.0/go.mod h1:E8a90lC4kjU5Lc2c0D+SnWhUuyuoCIVm4Ewzv3jCD3A=
+github.com/hashicorp/go-tfe v1.9.0 h1:jkmyo7WKNA7gZDegG5imndoC4sojWXhqMufO+KcHqrU=
+github.com/hashicorp/go-tfe v1.9.0/go.mod h1:uSWi2sPw7tLrqNIiASid9j3SprbbkPSJ/2s3X0mMemg=
 github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
diff --git a/internal/backend/local/backend.go b/internal/backend/local/backend.go
index dd6c9cc56f10..de1e7e5a422e 100644
--- a/internal/backend/local/backend.go
+++ b/internal/backend/local/backend.go
@@ -344,7 +344,7 @@ func (b *Local) opWait(
 
 		// try to force a PersistState just in case the process is terminated
 		// before we can complete.
-		if err := opStateMgr.PersistState(); err != nil {
+		if err := opStateMgr.PersistState(nil); err != nil {
 			// We can't error out from here, but warn the user if there was an error.
 			// If this isn't transient, we will catch it again below, and
 			// attempt to save the state another way.
diff --git a/internal/backend/local/backend_apply.go b/internal/backend/local/backend_apply.go
index a6c6ea9f0b4b..5bec4b442a44 100644
--- a/internal/backend/local/backend_apply.go
+++ b/internal/backend/local/backend_apply.go
@@ -53,7 +53,7 @@ func (b *Local) opApply(
 		op.ReportResult(runningOp, diags)
 		return
 	}
-	// the state was locked during succesfull context creation; unlock the state
+	// the state was locked during successful context creation; unlock the state
 	// when the operation completes
 	defer func() {
 		diags := op.StateLocker.Unlock()
@@ -68,6 +68,13 @@ func (b *Local) opApply(
 	// operation.
 	runningOp.State = lr.InputState
 
+	schemas, moreDiags := lr.Core.Schemas(lr.Config, lr.InputState)
+	diags = diags.Append(moreDiags)
+	if moreDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
 	var plan *plans.Plan
 	// If we weren't given a plan, then we refresh/plan
 	if op.PlanFile == nil {
@@ -80,13 +87,6 @@ func (b *Local) opApply(
 			return
 		}
 
-		schemas, moreDiags := lr.Core.Schemas(lr.Config, lr.InputState)
-		diags = diags.Append(moreDiags)
-		if moreDiags.HasErrors() {
-			op.ReportResult(runningOp, diags)
-			return
-		}
-
 		trivialPlan := !plan.CanApply()
 		hasUI := op.UIOut != nil && op.UIIn != nil
 		mustConfirm := hasUI && !op.AutoApprove && !trivialPlan
@@ -198,7 +198,7 @@ func (b *Local) opApply(
 
 	// Store the final state
 	runningOp.State = applyState
-	err := statemgr.WriteAndPersist(opState, applyState)
+	err := statemgr.WriteAndPersist(opState, applyState, schemas)
 	if err != nil {
 		// Export the state file from the state manager and assign the new
 		// state. This is needed to preserve the existing serial and lineage.
diff --git a/internal/backend/local/backend_local_test.go b/internal/backend/local/backend_local_test.go
index 05573f0f9d97..136dbaccf275 100644
--- a/internal/backend/local/backend_local_test.go
+++ b/internal/backend/local/backend_local_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
 	"github.com/hashicorp/terraform/internal/terminal"
+	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 )
 
@@ -233,6 +234,6 @@ func (s *stateStorageThatFailsRefresh) RefreshState() error {
 	return fmt.Errorf("intentionally failing for testing purposes")
 }
 
-func (s *stateStorageThatFailsRefresh) PersistState() error {
+func (s *stateStorageThatFailsRefresh) PersistState(schemas *terraform.Schemas) error {
 	return fmt.Errorf("unimplemented")
 }
diff --git a/internal/backend/local/backend_refresh.go b/internal/backend/local/backend_refresh.go
index 8ce3b6aff1a8..244e8e89bb6c 100644
--- a/internal/backend/local/backend_refresh.go
+++ b/internal/backend/local/backend_refresh.go
@@ -52,7 +52,7 @@ func (b *Local) opRefresh(
 		return
 	}
 
-	// the state was locked during succesfull context creation; unlock the state
+	// the state was locked during successful context creation; unlock the state
 	// when the operation completes
 	defer func() {
 		diags := op.StateLocker.Unlock()
@@ -73,6 +73,14 @@ func (b *Local) opRefresh(
 		))
 	}
 
+	// get schemas before writing state
+	schemas, moreDiags := lr.Core.Schemas(lr.Config, lr.InputState)
+	diags = diags.Append(moreDiags)
+	if moreDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
 	// Perform the refresh in a goroutine so we can be interrupted
 	var newState *states.State
 	var refreshDiags tfdiags.Diagnostics
@@ -96,7 +104,7 @@ func (b *Local) opRefresh(
 		return
 	}
 
-	err := statemgr.WriteAndPersist(opState, newState)
+	err := statemgr.WriteAndPersist(opState, newState, schemas)
 	if err != nil {
 		diags = diags.Append(fmt.Errorf("failed to write state: %w", err))
 		op.ReportResult(runningOp, diags)
diff --git a/internal/backend/remote-state/azure/backend_state.go b/internal/backend/remote-state/azure/backend_state.go
index 6a1a9c02f01d..82d2505c65a4 100644
--- a/internal/backend/remote-state/azure/backend_state.go
+++ b/internal/backend/remote-state/azure/backend_state.go
@@ -131,7 +131,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 				err = lockUnlock(err)
 				return nil, err
 			}
-			if err := stateMgr.PersistState(); err != nil {
+			if err := stateMgr.PersistState(nil); err != nil {
 				err = lockUnlock(err)
 				return nil, err
 			}
diff --git a/internal/backend/remote-state/consul/backend_state.go b/internal/backend/remote-state/consul/backend_state.go
index be1841eb8df3..5bd74b250878 100644
--- a/internal/backend/remote-state/consul/backend_state.go
+++ b/internal/backend/remote-state/consul/backend_state.go
@@ -120,7 +120,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 			err = lockUnlock(err)
 			return nil, err
 		}
-		if err := stateMgr.PersistState(); err != nil {
+		if err := stateMgr.PersistState(nil); err != nil {
 			err = lockUnlock(err)
 			return nil, err
 		}
diff --git a/internal/backend/remote-state/cos/backend_state.go b/internal/backend/remote-state/cos/backend_state.go
index ab92cfb7c0e8..4e47ae0e6811 100644
--- a/internal/backend/remote-state/cos/backend_state.go
+++ b/internal/backend/remote-state/cos/backend_state.go
@@ -126,7 +126,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 				err = lockUnlock(err)
 				return nil, err
 			}
-			if err := stateMgr.PersistState(); err != nil {
+			if err := stateMgr.PersistState(nil); err != nil {
 				err = lockUnlock(err)
 				return nil, err
 			}
diff --git a/internal/backend/remote-state/gcs/backend_state.go b/internal/backend/remote-state/gcs/backend_state.go
index ee764efb4c49..21b71834735c 100644
--- a/internal/backend/remote-state/gcs/backend_state.go
+++ b/internal/backend/remote-state/gcs/backend_state.go
@@ -131,7 +131,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 		if err := st.WriteState(states.NewState()); err != nil {
 			return nil, unlock(err)
 		}
-		if err := st.PersistState(); err != nil {
+		if err := st.PersistState(nil); err != nil {
 			return nil, unlock(err)
 		}
 
diff --git a/internal/backend/remote-state/inmem/backend.go b/internal/backend/remote-state/inmem/backend.go
index 7f8f56ef2034..4e0113cbc721 100644
--- a/internal/backend/remote-state/inmem/backend.go
+++ b/internal/backend/remote-state/inmem/backend.go
@@ -141,7 +141,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 			if err := s.WriteState(statespkg.NewState()); err != nil {
 				return nil, err
 			}
-			if err := s.PersistState(); err != nil {
+			if err := s.PersistState(nil); err != nil {
 				return nil, err
 			}
 		}
diff --git a/internal/backend/remote-state/inmem/backend_test.go b/internal/backend/remote-state/inmem/backend_test.go
index 395199890a78..b7e9a555a906 100644
--- a/internal/backend/remote-state/inmem/backend_test.go
+++ b/internal/backend/remote-state/inmem/backend_test.go
@@ -82,7 +82,7 @@ func TestRemoteState(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatal(err)
 	}
 
diff --git a/internal/backend/remote-state/kubernetes/backend_state.go b/internal/backend/remote-state/kubernetes/backend_state.go
index 56aa089ff81c..6e8ce449d1c4 100644
--- a/internal/backend/remote-state/kubernetes/backend_state.go
+++ b/internal/backend/remote-state/kubernetes/backend_state.go
@@ -123,7 +123,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 		if err := stateMgr.WriteState(states.NewState()); err != nil {
 			return nil, unlock(err)
 		}
-		if err := stateMgr.PersistState(); err != nil {
+		if err := stateMgr.PersistState(nil); err != nil {
 			return nil, unlock(err)
 		}
 
diff --git a/internal/backend/remote-state/manta/backend_state.go b/internal/backend/remote-state/manta/backend_state.go
index 925d82083d52..b30b250dc73a 100644
--- a/internal/backend/remote-state/manta/backend_state.go
+++ b/internal/backend/remote-state/manta/backend_state.go
@@ -108,7 +108,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 				err = lockUnlock(err)
 				return nil, err
 			}
-			if err := stateMgr.PersistState(); err != nil {
+			if err := stateMgr.PersistState(nil); err != nil {
 				err = lockUnlock(err)
 				return nil, err
 			}
diff --git a/internal/backend/remote-state/oss/backend_state.go b/internal/backend/remote-state/oss/backend_state.go
index 77a2775f8ac4..672a2e1aa25a 100644
--- a/internal/backend/remote-state/oss/backend_state.go
+++ b/internal/backend/remote-state/oss/backend_state.go
@@ -160,7 +160,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 				err = lockUnlock(err)
 				return nil, err
 			}
-			if err := stateMgr.PersistState(); err != nil {
+			if err := stateMgr.PersistState(nil); err != nil {
 				err = lockUnlock(err)
 				return nil, err
 			}
diff --git a/internal/backend/remote-state/pg/backend_state.go b/internal/backend/remote-state/pg/backend_state.go
index 2700c5196927..f3eb650092eb 100644
--- a/internal/backend/remote-state/pg/backend_state.go
+++ b/internal/backend/remote-state/pg/backend_state.go
@@ -99,7 +99,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 				err = lockUnlock(err)
 				return nil, err
 			}
-			if err := stateMgr.PersistState(); err != nil {
+			if err := stateMgr.PersistState(nil); err != nil {
 				err = lockUnlock(err)
 				return nil, err
 			}
diff --git a/internal/backend/remote-state/pg/backend_test.go b/internal/backend/remote-state/pg/backend_test.go
index da058483d845..064c001f5b3b 100644
--- a/internal/backend/remote-state/pg/backend_test.go
+++ b/internal/backend/remote-state/pg/backend_test.go
@@ -330,7 +330,7 @@ func TestBackendConcurrentLock(t *testing.T) {
 		t.Fatalf("failed to lock first state: %v", err)
 	}
 
-	if err = s1.PersistState(); err != nil {
+	if err = s1.PersistState(nil); err != nil {
 		t.Fatalf("failed to persist state: %v", err)
 	}
 
@@ -343,7 +343,7 @@ func TestBackendConcurrentLock(t *testing.T) {
 		t.Fatalf("failed to lock second state: %v", err)
 	}
 
-	if err = s2.PersistState(); err != nil {
+	if err = s2.PersistState(nil); err != nil {
 		t.Fatalf("failed to persist state: %v", err)
 	}
 
diff --git a/internal/backend/remote-state/s3/backend_state.go b/internal/backend/remote-state/s3/backend_state.go
index 0134c861d015..d13cc32d4f83 100644
--- a/internal/backend/remote-state/s3/backend_state.go
+++ b/internal/backend/remote-state/s3/backend_state.go
@@ -184,7 +184,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 				err = lockUnlock(err)
 				return nil, err
 			}
-			if err := stateMgr.PersistState(); err != nil {
+			if err := stateMgr.PersistState(nil); err != nil {
 				err = lockUnlock(err)
 				return nil, err
 			}
diff --git a/internal/backend/remote-state/s3/backend_test.go b/internal/backend/remote-state/s3/backend_test.go
index a44f154c08c4..1fd49c461ab5 100644
--- a/internal/backend/remote-state/s3/backend_test.go
+++ b/internal/backend/remote-state/s3/backend_test.go
@@ -478,7 +478,7 @@ func TestBackendExtraPaths(t *testing.T) {
 	// Write the first state
 	stateMgr := &remote.State{Client: client}
 	stateMgr.WriteState(s1)
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(nil); err != nil {
 		t.Fatal(err)
 	}
 
@@ -488,7 +488,7 @@ func TestBackendExtraPaths(t *testing.T) {
 	client.path = b.path("s2")
 	stateMgr2 := &remote.State{Client: client}
 	stateMgr2.WriteState(s2)
-	if err := stateMgr2.PersistState(); err != nil {
+	if err := stateMgr2.PersistState(nil); err != nil {
 		t.Fatal(err)
 	}
 
@@ -501,7 +501,7 @@ func TestBackendExtraPaths(t *testing.T) {
 	// put a state in an env directory name
 	client.path = b.workspaceKeyPrefix + "/error"
 	stateMgr.WriteState(states.NewState())
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(nil); err != nil {
 		t.Fatal(err)
 	}
 	if err := checkStateList(b, []string{"default", "s1", "s2"}); err != nil {
@@ -511,7 +511,7 @@ func TestBackendExtraPaths(t *testing.T) {
 	// add state with the wrong key for an existing env
 	client.path = b.workspaceKeyPrefix + "/s2/notTestState"
 	stateMgr.WriteState(states.NewState())
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(nil); err != nil {
 		t.Fatal(err)
 	}
 	if err := checkStateList(b, []string{"default", "s1", "s2"}); err != nil {
@@ -550,7 +550,7 @@ func TestBackendExtraPaths(t *testing.T) {
 	// add a state with a key that matches an existing environment dir name
 	client.path = b.workspaceKeyPrefix + "/s2/"
 	stateMgr.WriteState(states.NewState())
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(nil); err != nil {
 		t.Fatal(err)
 	}
 
diff --git a/internal/backend/remote-state/swift/backend_state.go b/internal/backend/remote-state/swift/backend_state.go
index b853b64c9638..719585d855f7 100644
--- a/internal/backend/remote-state/swift/backend_state.go
+++ b/internal/backend/remote-state/swift/backend_state.go
@@ -172,7 +172,7 @@ func (b *Backend) StateMgr(name string) (statemgr.Full, error) {
 				err = lockUnlock(err)
 				return nil, err
 			}
-			if err := stateMgr.PersistState(); err != nil {
+			if err := stateMgr.PersistState(nil); err != nil {
 				err = lockUnlock(err)
 				return nil, err
 			}
diff --git a/internal/backend/testing.go b/internal/backend/testing.go
index 844f95668dd5..a8c04e0ac7ec 100644
--- a/internal/backend/testing.go
+++ b/internal/backend/testing.go
@@ -132,7 +132,7 @@ func TestBackendStates(t *testing.T, b Backend) {
 		if err := foo.WriteState(fooState); err != nil {
 			t.Fatal("error writing foo state:", err)
 		}
-		if err := foo.PersistState(); err != nil {
+		if err := foo.PersistState(nil); err != nil {
 			t.Fatal("error persisting foo state:", err)
 		}
 
@@ -160,7 +160,7 @@ func TestBackendStates(t *testing.T, b Backend) {
 		if err := bar.WriteState(barState); err != nil {
 			t.Fatalf("bad: %s", err)
 		}
-		if err := bar.PersistState(); err != nil {
+		if err := bar.PersistState(nil); err != nil {
 			t.Fatalf("bad: %s", err)
 		}
 
diff --git a/internal/cloud/backend.go b/internal/cloud/backend.go
index 669cc372270d..ded38f75f29f 100644
--- a/internal/cloud/backend.go
+++ b/internal/cloud/backend.go
@@ -526,15 +526,10 @@ func (b *Cloud) DeleteWorkspace(name string) error {
 	}
 
 	// Configure the remote workspace name.
-	client := &remoteClient{
-		client:       b.client,
-		organization: b.organization,
-		workspace: &tfe.Workspace{
-			Name: name,
-		},
-	}
-
-	return client.Delete()
+	State := &State{tfeClient: b.client, organization: b.organization, workspace: &tfe.Workspace{
+		Name: name,
+	}}
+	return State.Delete()
 }
 
 // StateMgr implements backend.Enhanced.
@@ -619,16 +614,7 @@ func (b *Cloud) StateMgr(name string) (statemgr.Full, error) {
 		}
 	}
 
-	client := &remoteClient{
-		client:       b.client,
-		organization: b.organization,
-		workspace:    workspace,
-
-		// This is optionally set during Terraform Enterprise runs.
-		runID: os.Getenv("TFE_RUN_ID"),
-	}
-
-	return NewState(client), nil
+	return &State{tfeClient: b.client, organization: b.organization, workspace: workspace}, nil
 }
 
 // Operation implements backend.Enhanced.
diff --git a/internal/cloud/backend_state.go b/internal/cloud/backend_state.go
deleted file mode 100644
index f8cb9f24551d..000000000000
--- a/internal/cloud/backend_state.go
+++ /dev/null
@@ -1,196 +0,0 @@
-package cloud
-
-import (
-	"bytes"
-	"context"
-	"crypto/md5"
-	"encoding/base64"
-	"encoding/json"
-	"errors"
-	"fmt"
-
-	tfe "github.com/hashicorp/go-tfe"
-
-	"github.com/hashicorp/terraform/internal/command/jsonstate"
-	"github.com/hashicorp/terraform/internal/states/remote"
-	"github.com/hashicorp/terraform/internal/states/statefile"
-	"github.com/hashicorp/terraform/internal/states/statemgr"
-)
-
-type remoteClient struct {
-	client         *tfe.Client
-	lockInfo       *statemgr.LockInfo
-	organization   string
-	runID          string
-	stateUploadErr bool
-	workspace      *tfe.Workspace
-	forcePush      bool
-}
-
-// Get the remote state.
-func (r *remoteClient) Get() (*remote.Payload, error) {
-	ctx := context.Background()
-
-	sv, err := r.client.StateVersions.ReadCurrent(ctx, r.workspace.ID)
-	if err != nil {
-		if err == tfe.ErrResourceNotFound {
-			// If no state exists, then return nil.
-			return nil, nil
-		}
-		return nil, fmt.Errorf("failed to retrieve state: %w", err)
-	}
-
-	state, err := r.client.StateVersions.Download(ctx, sv.DownloadURL)
-	if err != nil {
-		return nil, fmt.Errorf("failed to download state: %w", err)
-	}
-
-	// If the state is empty, then return nil.
-	if len(state) == 0 {
-		return nil, nil
-	}
-
-	// Get the MD5 checksum of the state.
-	sum := md5.Sum(state)
-
-	return &remote.Payload{
-		Data: state,
-		MD5:  sum[:],
-	}, nil
-}
-
-// Put the remote state.
-func (r *remoteClient) Put(state []byte) error {
-	ctx := context.Background()
-
-	// Read the raw state into a Terraform state.
-	stateFile, err := statefile.Read(bytes.NewReader(state))
-	if err != nil {
-		return fmt.Errorf("failed to read state: %w", err)
-	}
-
-	ov, err := jsonstate.MarshalOutputs(stateFile.State.RootModule().OutputValues)
-	if err != nil {
-		return fmt.Errorf("failed to translate outputs: %w", err)
-	}
-	o, err := json.Marshal(ov)
-	if err != nil {
-		return fmt.Errorf("failed to marshal outputs to json: %w", err)
-	}
-
-	options := tfe.StateVersionCreateOptions{
-		Lineage:          tfe.String(stateFile.Lineage),
-		Serial:           tfe.Int64(int64(stateFile.Serial)),
-		MD5:              tfe.String(fmt.Sprintf("%x", md5.Sum(state))),
-		State:            tfe.String(base64.StdEncoding.EncodeToString(state)),
-		Force:            tfe.Bool(r.forcePush),
-		JSONStateOutputs: tfe.String(base64.StdEncoding.EncodeToString(o)),
-	}
-
-	// If we have a run ID, make sure to add it to the options
-	// so the state will be properly associated with the run.
-	if r.runID != "" {
-		options.Run = &tfe.Run{ID: r.runID}
-	}
-
-	// Create the new state.
-	_, err = r.client.StateVersions.Create(ctx, r.workspace.ID, options)
-	if err != nil {
-		r.stateUploadErr = true
-		return fmt.Errorf("failed to upload state: %w", err)
-	}
-
-	return nil
-}
-
-// Delete the remote state.
-func (r *remoteClient) Delete() error {
-	err := r.client.Workspaces.Delete(context.Background(), r.organization, r.workspace.Name)
-	if err != nil && err != tfe.ErrResourceNotFound {
-		return fmt.Errorf("failed to delete workspace %s: %w", r.workspace.Name, err)
-	}
-
-	return nil
-}
-
-// EnableForcePush to allow the remote client to overwrite state
-// by implementing remote.ClientForcePusher
-func (r *remoteClient) EnableForcePush() {
-	r.forcePush = true
-}
-
-// Lock the remote state.
-func (r *remoteClient) Lock(info *statemgr.LockInfo) (string, error) {
-	ctx := context.Background()
-
-	lockErr := &statemgr.LockError{Info: r.lockInfo}
-
-	// Lock the workspace.
-	_, err := r.client.Workspaces.Lock(ctx, r.workspace.ID, tfe.WorkspaceLockOptions{
-		Reason: tfe.String("Locked by Terraform"),
-	})
-	if err != nil {
-		if err == tfe.ErrWorkspaceLocked {
-			lockErr.Info = info
-			err = fmt.Errorf("%s (lock ID: \"%s/%s\")", err, r.organization, r.workspace.Name)
-		}
-		lockErr.Err = err
-		return "", lockErr
-	}
-
-	r.lockInfo = info
-
-	return r.lockInfo.ID, nil
-}
-
-// Unlock the remote state.
-func (r *remoteClient) Unlock(id string) error {
-	ctx := context.Background()
-
-	// We first check if there was an error while uploading the latest
-	// state. If so, we will not unlock the workspace to prevent any
-	// changes from being applied until the correct state is uploaded.
-	if r.stateUploadErr {
-		return nil
-	}
-
-	lockErr := &statemgr.LockError{Info: r.lockInfo}
-
-	// With lock info this should be treated as a normal unlock.
-	if r.lockInfo != nil {
-		// Verify the expected lock ID.
-		if r.lockInfo.ID != id {
-			lockErr.Err = errors.New("lock ID does not match existing lock")
-			return lockErr
-		}
-
-		// Unlock the workspace.
-		_, err := r.client.Workspaces.Unlock(ctx, r.workspace.ID)
-		if err != nil {
-			lockErr.Err = err
-			return lockErr
-		}
-
-		return nil
-	}
-
-	// Verify the optional force-unlock lock ID.
-	if r.organization+"/"+r.workspace.Name != id {
-		lockErr.Err = fmt.Errorf(
-			"lock ID %q does not match existing lock ID \"%s/%s\"",
-			id,
-			r.organization,
-			r.workspace.Name,
-		)
-		return lockErr
-	}
-
-	// Force unlock the workspace.
-	_, err := r.client.Workspaces.ForceUnlock(ctx, r.workspace.ID)
-	if err != nil {
-		lockErr.Err = err
-		return lockErr
-	}
-
-	return nil
-}
diff --git a/internal/cloud/backend_state_test.go b/internal/cloud/backend_state_test.go
deleted file mode 100644
index 3b9833c38e5e..000000000000
--- a/internal/cloud/backend_state_test.go
+++ /dev/null
@@ -1,103 +0,0 @@
-package cloud
-
-import (
-	"bytes"
-	"os"
-	"testing"
-
-	tfe "github.com/hashicorp/go-tfe"
-
-	"github.com/hashicorp/terraform/internal/states"
-	"github.com/hashicorp/terraform/internal/states/remote"
-	"github.com/hashicorp/terraform/internal/states/statefile"
-)
-
-func TestRemoteClient_impl(t *testing.T) {
-	var _ remote.Client = new(remoteClient)
-}
-
-func TestRemoteClient(t *testing.T) {
-	client := testRemoteClient(t)
-	remote.TestClient(t, client)
-}
-
-func TestRemoteClient_stateVersionCreated(t *testing.T) {
-	b, bCleanup := testBackendWithName(t)
-	defer bCleanup()
-
-	raw, err := b.StateMgr(testBackendSingleWorkspaceName)
-	if err != nil {
-		t.Fatalf("error: %v", err)
-	}
-
-	client := raw.(*State).Client
-
-	err = client.Put(([]byte)(`
-{
-	"version": 4,
-	"terraform_version": "1.3.0",
-	"serial": 1,
-	"lineage": "backend-change",
-	"outputs": {
-			"foo": {
-					"type": "string",
-					"value": "bar"
-			}
-	}
-}`))
-	if err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-
-	stateVersionsAPI := b.client.StateVersions.(*MockStateVersions)
-	if got, want := len(stateVersionsAPI.stateVersions), 1; got != want {
-		t.Fatalf("wrong number of state versions in the mock client %d; want %d", got, want)
-	}
-
-	var stateVersion *tfe.StateVersion
-	for _, sv := range stateVersionsAPI.stateVersions {
-		stateVersion = sv
-	}
-
-	if stateVersionsAPI.outputStates[stateVersion.ID] == nil || len(stateVersionsAPI.outputStates[stateVersion.ID]) == 0 {
-		t.Fatal("no state version outputs in the mock client")
-	}
-}
-
-func TestRemoteClient_TestRemoteLocks(t *testing.T) {
-	b, bCleanup := testBackendWithName(t)
-	defer bCleanup()
-
-	s1, err := b.StateMgr(testBackendSingleWorkspaceName)
-	if err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-
-	s2, err := b.StateMgr(testBackendSingleWorkspaceName)
-	if err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-
-	remote.TestRemoteLocks(t, s1.(*State).Client, s2.(*State).Client)
-}
-
-func TestRemoteClient_withRunID(t *testing.T) {
-	// Set the TFE_RUN_ID environment variable before creating the client!
-	if err := os.Setenv("TFE_RUN_ID", GenerateID("run-")); err != nil {
-		t.Fatalf("error setting env var TFE_RUN_ID: %v", err)
-	}
-
-	// Create a new test client.
-	client := testRemoteClient(t)
-
-	// Create a new empty state.
-	sf := statefile.New(states.NewState(), "", 0)
-	var buf bytes.Buffer
-	statefile.Write(sf, &buf)
-
-	// Store the new state to verify (this will be done
-	// by the mock that is used) that the run ID is set.
-	if err := client.Put(buf.Bytes()); err != nil {
-		t.Fatalf("expected no error, got %v", err)
-	}
-}
diff --git a/internal/cloud/state.go b/internal/cloud/state.go
index 73bea8ba5d9c..04d9f7773439 100644
--- a/internal/cloud/state.go
+++ b/internal/cloud/state.go
@@ -1,29 +1,57 @@
 package cloud
 
 import (
+	"bytes"
 	"context"
+	"crypto/md5"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"log"
+	"os"
 	"strings"
+	"sync"
 
-	"github.com/hashicorp/go-tfe"
 	"github.com/zclconf/go-cty/cty"
 	"github.com/zclconf/go-cty/cty/gocty"
 
+	tfe "github.com/hashicorp/go-tfe"
+	uuid "github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/terraform/internal/command/jsonstate"
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/remote"
+	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
+	"github.com/hashicorp/terraform/internal/terraform"
 )
 
-// State is similar to remote State and delegates to it, except in the case of output values,
-// which use a separate methodology that ensures the caller is authorized to read cloud
-// workspace outputs.
+// State implements the State interfaces in the state package to handle
+// reading and writing the remote state to TFC. This State on its own does no
+// local caching so every persist will go to the remote storage and local
+// writes will go to memory.
 type State struct {
-	Client *remoteClient
+	mu sync.Mutex
 
-	delegate remote.State
+	// We track two pieces of meta data in addition to the state itself:
+	//
+	// lineage - the state's unique ID
+	// serial  - the monotonic counter of "versions" of the state
+	//
+	// Both of these (along with state) have a sister field
+	// that represents the values read in from an existing source.
+	// All three of these values are used to determine if the new
+	// state has changed from an existing state we read in.
+	lineage, readLineage string
+	serial, readSerial   uint64
+	state, readState     *states.State
+	disableLocks         bool
+	tfeClient            *tfe.Client
+	organization         string
+	workspace            *tfe.Workspace
+	stateUploadErr       bool
+	forcePush            bool
+	lockInfo             *statemgr.LockInfo
 }
 
 var ErrStateVersionUnauthorizedUpgradeState = errors.New(strings.TrimSpace(`
@@ -33,69 +61,370 @@ of authorization and therefore this error can usually be fixed by upgrading the
 remote state version.
 `))
 
-// Proof that cloud State is a statemgr.Persistent interface
-var _ statemgr.Persistent = (*State)(nil)
+var _ statemgr.Full = (*State)(nil)
+var _ statemgr.Migrator = (*State)(nil)
 
-func NewState(client *remoteClient) *State {
-	return &State{
-		Client:   client,
-		delegate: remote.State{Client: client},
+// statemgr.Reader impl.
+func (s *State) State() *states.State {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.state.DeepCopy()
+}
+
+// StateForMigration is part of our implementation of statemgr.Migrator.
+func (s *State) StateForMigration() *statefile.File {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return statefile.New(s.state.DeepCopy(), s.lineage, s.serial)
+}
+
+// WriteStateForMigration is part of our implementation of statemgr.Migrator.
+func (s *State) WriteStateForMigration(f *statefile.File, force bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if !force {
+		checkFile := statefile.New(s.state, s.lineage, s.serial)
+		if err := statemgr.CheckValidImport(f, checkFile); err != nil {
+			return err
+		}
+	}
+
+	// The remote backend needs to pass the `force` flag through to its client.
+	// For backends that support such operations, inform the client
+	// that a force push has been requested
+	if force {
+		s.EnableForcePush()
 	}
+
+	// We create a deep copy of the state here, because the caller also has
+	// a reference to the given object and can potentially go on to mutate
+	// it after we return, but we want the snapshot at this point in time.
+	s.state = f.State.DeepCopy()
+	s.lineage = f.Lineage
+	s.serial = f.Serial
+
+	return nil
 }
 
-// State delegates calls to read State to the remote State
-func (s *State) State() *states.State {
-	return s.delegate.State()
+// DisableLocks turns the Lock and Unlock methods into no-ops. This is intended
+// to be called during initialization of a state manager and should not be
+// called after any of the statemgr.Full interface methods have been called.
+func (s *State) DisableLocks() {
+	s.disableLocks = true
 }
 
-// Lock delegates calls to lock state to the remote State
-func (s *State) Lock(info *statemgr.LockInfo) (string, error) {
-	return s.delegate.Lock(info)
+// StateSnapshotMeta returns the metadata from the most recently persisted
+// or refreshed persistent state snapshot.
+//
+// This is an implementation of statemgr.PersistentMeta.
+func (s *State) StateSnapshotMeta() statemgr.SnapshotMeta {
+	return statemgr.SnapshotMeta{
+		Lineage: s.lineage,
+		Serial:  s.serial,
+	}
 }
 
-// Unlock delegates calls to unlock state to the remote State
-func (s *State) Unlock(id string) error {
-	return s.delegate.Unlock(id)
+// statemgr.Writer impl.
+func (s *State) WriteState(state *states.State) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// We create a deep copy of the state here, because the caller also has
+	// a reference to the given object and can potentially go on to mutate
+	// it after we return, but we want the snapshot at this point in time.
+	s.state = state.DeepCopy()
+
+	return nil
+}
+
+// PersistState uploads a snapshot of the latest state as a StateVersion to Terraform Cloud
+func (s *State) PersistState(schemas *terraform.Schemas) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.readState != nil {
+		lineageUnchanged := s.readLineage != "" && s.lineage == s.readLineage
+		serialUnchanged := s.readSerial != 0 && s.serial == s.readSerial
+		stateUnchanged := statefile.StatesMarshalEqual(s.state, s.readState)
+		if stateUnchanged && lineageUnchanged && serialUnchanged {
+			// If the state, lineage or serial haven't changed at all then we have nothing to do.
+			return nil
+		}
+		s.serial++
+	} else {
+		// We might be writing a new state altogether, but before we do that
+		// we'll check to make sure there isn't already a snapshot present
+		// that we ought to be updating.
+		err := s.refreshState()
+		if err != nil {
+			return fmt.Errorf("failed checking for existing remote state: %s", err)
+		}
+		if s.lineage == "" { // indicates that no state snapshot is present yet
+			lineage, err := uuid.GenerateUUID()
+			if err != nil {
+				return fmt.Errorf("failed to generate initial lineage: %v", err)
+			}
+			s.lineage = lineage
+			s.serial = 0
+		}
+	}
+
+	f := statefile.New(s.state, s.lineage, s.serial)
+
+	var buf bytes.Buffer
+	err := statefile.Write(f, &buf)
+	if err != nil {
+		return err
+	}
+
+	var jsonState []byte
+	if schemas != nil {
+		jsonState, err = jsonstate.Marshal(f, schemas)
+		if err != nil {
+			return err
+		}
+	}
+
+	stateFile, err := statefile.Read(bytes.NewReader(buf.Bytes()))
+	if err != nil {
+		return fmt.Errorf("failed to read state: %w", err)
+	}
+
+	ov, err := jsonstate.MarshalOutputs(stateFile.State.RootModule().OutputValues)
+	if err != nil {
+		return fmt.Errorf("failed to translate outputs: %w", err)
+	}
+	jsonStateOutputs, err := json.Marshal(ov)
+	if err != nil {
+		return fmt.Errorf("failed to marshal outputs to json: %w", err)
+	}
+
+	err = s.uploadState(s.lineage, s.serial, s.forcePush, buf.Bytes(), jsonState, jsonStateOutputs)
+	if err != nil {
+		s.stateUploadErr = true
+		return fmt.Errorf("error uploading state: %w", err)
+	}
+	// After we've successfully persisted, what we just wrote is our new
+	// reference state until someone calls RefreshState again.
+	// We've potentially overwritten (via force) the state, lineage
+	// and / or serial (and serial was incremented) so we copy over all
+	// three fields so everything matches the new state and a subsequent
+	// operation would correctly detect no changes to the lineage, serial or state.
+	s.readState = s.state.DeepCopy()
+	s.readLineage = s.lineage
+	s.readSerial = s.serial
+	return nil
+}
+
+func (s *State) uploadState(lineage string, serial uint64, isForcePush bool, state, jsonState, jsonStateOutputs []byte) error {
+	ctx := context.Background()
+
+	options := tfe.StateVersionCreateOptions{
+		Lineage:          tfe.String(lineage),
+		Serial:           tfe.Int64(int64(serial)),
+		MD5:              tfe.String(fmt.Sprintf("%x", md5.Sum(state))),
+		State:            tfe.String(base64.StdEncoding.EncodeToString(state)),
+		Force:            tfe.Bool(isForcePush),
+		JSONState:        tfe.String(base64.StdEncoding.EncodeToString(jsonState)),
+		JSONStateOutputs: tfe.String(base64.StdEncoding.EncodeToString(jsonStateOutputs)),
+	}
+
+	// If we have a run ID, make sure to add it to the options
+	// so the state will be properly associated with the run.
+	runID := os.Getenv("TFE_RUN_ID")
+	if runID != "" {
+		options.Run = &tfe.Run{ID: runID}
+	}
+	// Create the new state.
+	_, err := s.tfeClient.StateVersions.Create(ctx, s.workspace.ID, options)
+	return err
+}
+
+// Lock calls the Client's Lock method if it's implemented.
+func (s *State) Lock(info *statemgr.LockInfo) (string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.disableLocks {
+		return "", nil
+	}
+	ctx := context.Background()
+
+	lockErr := &statemgr.LockError{Info: s.lockInfo}
+
+	// Lock the workspace.
+	_, err := s.tfeClient.Workspaces.Lock(ctx, s.workspace.ID, tfe.WorkspaceLockOptions{
+		Reason: tfe.String("Locked by Terraform"),
+	})
+	if err != nil {
+		if err == tfe.ErrWorkspaceLocked {
+			lockErr.Info = info
+			err = fmt.Errorf("%s (lock ID: \"%s/%s\")", err, s.organization, s.workspace.Name)
+		}
+		lockErr.Err = err
+		return "", lockErr
+	}
+
+	s.lockInfo = info
+
+	return s.lockInfo.ID, nil
 }
 
-// RefreshState delegates calls to refresh State to the remote State
+// statemgr.Refresher impl.
 func (s *State) RefreshState() error {
-	return s.delegate.RefreshState()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.refreshState()
 }
 
-// RefreshState delegates calls to refresh State to the remote State
-func (s *State) PersistState() error {
-	return s.delegate.PersistState()
+// refreshState is the main implementation of RefreshState, but split out so
+// that we can make internal calls to it from methods that are already holding
+// the s.mu lock.
+func (s *State) refreshState() error {
+	payload, err := s.getStatePayload()
+	if err != nil {
+		return err
+	}
+
+	// no remote state is OK
+	if payload == nil {
+		s.readState = nil
+		s.lineage = ""
+		s.serial = 0
+		return nil
+	}
+
+	stateFile, err := statefile.Read(bytes.NewReader(payload.Data))
+	if err != nil {
+		return err
+	}
+
+	s.lineage = stateFile.Lineage
+	s.serial = stateFile.Serial
+	s.state = stateFile.State
+
+	// Properties from the remote must be separate so we can
+	// track changes as lineage, serial and/or state are mutated
+	s.readLineage = stateFile.Lineage
+	s.readSerial = stateFile.Serial
+	s.readState = s.state.DeepCopy()
+	return nil
 }
 
-// WriteState delegates calls to write State to the remote State
-func (s *State) WriteState(state *states.State) error {
-	return s.delegate.WriteState(state)
+func (s *State) getStatePayload() (*remote.Payload, error) {
+	ctx := context.Background()
+
+	sv, err := s.tfeClient.StateVersions.ReadCurrent(ctx, s.workspace.ID)
+	if err != nil {
+		if err == tfe.ErrResourceNotFound {
+			// If no state exists, then return nil.
+			return nil, nil
+		}
+		return nil, fmt.Errorf("error retrieving state: %v", err)
+	}
+
+	state, err := s.tfeClient.StateVersions.Download(ctx, sv.DownloadURL)
+	if err != nil {
+		return nil, fmt.Errorf("error downloading state: %v", err)
+	}
+
+	// If the state is empty, then return nil.
+	if len(state) == 0 {
+		return nil, nil
+	}
+
+	// Get the MD5 checksum of the state.
+	sum := md5.Sum(state)
+
+	return &remote.Payload{
+		Data: state,
+		MD5:  sum[:],
+	}, nil
 }
 
-func (s *State) fallbackReadOutputsFromFullState() (map[string]*states.OutputValue, error) {
-	log.Printf("[DEBUG] falling back to reading full state")
+// Unlock calls the Client's Unlock method if it's implemented.
+func (s *State) Unlock(id string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.disableLocks {
+		return nil
+	}
+
+	ctx := context.Background()
+
+	// We first check if there was an error while uploading the latest
+	// state. If so, we will not unlock the workspace to prevent any
+	// changes from being applied until the correct state is uploaded.
+	if s.stateUploadErr {
+		return nil
+	}
+
+	lockErr := &statemgr.LockError{Info: s.lockInfo}
+
+	// With lock info this should be treated as a normal unlock.
+	if s.lockInfo != nil {
+		// Verify the expected lock ID.
+		if s.lockInfo.ID != id {
+			lockErr.Err = fmt.Errorf("lock ID does not match existing lock")
+			return lockErr
+		}
+
+		// Unlock the workspace.
+		_, err := s.tfeClient.Workspaces.Unlock(ctx, s.workspace.ID)
+		if err != nil {
+			lockErr.Err = err
+			return lockErr
+		}
 
-	if err := s.RefreshState(); err != nil {
-		return nil, fmt.Errorf("failed to load state: %w", err)
+		return nil
 	}
 
-	state := s.State()
-	if state == nil {
-		// We know that there is supposed to be state (and this is not simply a new workspace
-		// without state) because the fallback is only invoked when outputs are present but
-		// detailed types are not available.
-		return nil, ErrStateVersionUnauthorizedUpgradeState
+	// Verify the optional force-unlock lock ID.
+	if s.organization+"/"+s.workspace.Name != id {
+		lockErr.Err = fmt.Errorf(
+			"lock ID %q does not match existing lock ID \"%s/%s\"",
+			id,
+			s.organization,
+			s.workspace.Name,
+		)
+		return lockErr
 	}
 
-	return state.RootModule().OutputValues, nil
+	// Force unlock the workspace.
+	_, err := s.tfeClient.Workspaces.ForceUnlock(ctx, s.workspace.ID)
+	if err != nil {
+		lockErr.Err = err
+		return lockErr
+	}
+
+	return nil
+}
+
+// Delete the remote state.
+func (s *State) Delete() error {
+	err := s.tfeClient.Workspaces.Delete(context.Background(), s.organization, s.workspace.Name)
+	if err != nil && err != tfe.ErrResourceNotFound {
+		return fmt.Errorf("error deleting workspace %s: %v", s.workspace.Name, err)
+	}
+
+	return nil
+}
+
+// EnableForcePush to allow the remote client to overwrite state
+// by implementing remote.ClientForcePusher
+func (s *State) EnableForcePush() {
+	s.forcePush = true
 }
 
 // GetRootOutputValues fetches output values from Terraform Cloud
 func (s *State) GetRootOutputValues() (map[string]*states.OutputValue, error) {
 	ctx := context.Background()
 
-	so, err := s.Client.client.StateVersionOutputs.ReadCurrent(ctx, s.Client.workspace.ID)
+	so, err := s.tfeClient.StateVersionOutputs.ReadCurrent(ctx, s.workspace.ID)
 
 	if err != nil {
 		return nil, fmt.Errorf("could not read state version outputs: %w", err)
@@ -109,13 +438,27 @@ func (s *State) GetRootOutputValues() (map[string]*states.OutputValue, error) {
 			// with a version of terraform < 1.3.0. In this case, we'll eject completely from this
 			// function and fall back to the old behavior of reading the entire state file, which
 			// requires a higher level of authorization.
-			return s.fallbackReadOutputsFromFullState()
+			log.Printf("[DEBUG] falling back to reading full state")
+
+			if err := s.RefreshState(); err != nil {
+				return nil, fmt.Errorf("failed to load state: %w", err)
+			}
+
+			state := s.State()
+			if state == nil {
+				// We know that there is supposed to be state (and this is not simply a new workspace
+				// without state) because the fallback is only invoked when outputs are present but
+				// detailed types are not available.
+				return nil, ErrStateVersionUnauthorizedUpgradeState
+			}
+
+			return state.RootModule().OutputValues, nil
 		}
 
 		if output.Sensitive {
 			// Since this is a sensitive value, the output must be requested explicitly in order to
 			// read its value, which is assumed to be present by callers
-			sensitiveOutput, err := s.Client.client.StateVersionOutputs.Read(ctx, output.ID)
+			sensitiveOutput, err := s.tfeClient.StateVersionOutputs.Read(ctx, output.ID)
 			if err != nil {
 				return nil, fmt.Errorf("could not read state version output %s: %w", output.ID, err)
 			}
diff --git a/internal/cloud/state_test.go b/internal/cloud/state_test.go
index 738ae721a44d..ee819d339fcf 100644
--- a/internal/cloud/state_test.go
+++ b/internal/cloud/state_test.go
@@ -1,10 +1,12 @@
 package cloud
 
 import (
+	"bytes"
+	"io/ioutil"
 	"testing"
 
-	"github.com/hashicorp/go-tfe"
-
+	tfe "github.com/hashicorp/go-tfe"
+	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
 )
 
@@ -27,14 +29,9 @@ func TestState_GetRootOutputValues(t *testing.T) {
 	b, bCleanup := testBackendWithOutputs(t)
 	defer bCleanup()
 
-	client := &remoteClient{
-		client: b.client,
-		workspace: &tfe.Workspace{
-			ID: "ws-abcd",
-		},
-	}
-
-	state := NewState(client)
+	state := &State{tfeClient: b.client, organization: b.organization, workspace: &tfe.Workspace{
+		ID: "ws-abcd",
+	}}
 	outputs, err := state.GetRootOutputValues()
 
 	if err != nil {
@@ -81,3 +78,118 @@ func TestState_GetRootOutputValues(t *testing.T) {
 		}
 	}
 }
+
+func TestState(t *testing.T) {
+	var buf bytes.Buffer
+	s := statemgr.TestFullInitialState()
+	sf := statefile.New(s, "stub-lineage", 2)
+	err := statefile.Write(sf, &buf)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	data := buf.Bytes()
+
+	state := testCloudState(t)
+
+	jsonState, err := ioutil.ReadFile("../command/testdata/show-json-state/sensitive-variables/output.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jsonStateOutputs := []byte(`
+{
+	"outputs": {
+			"foo": {
+					"type": "string",
+					"value": "bar"
+			}
+	}
+}`)
+
+	if err := state.uploadState(state.lineage, state.serial, state.forcePush, data, jsonState, jsonStateOutputs); err != nil {
+		t.Fatalf("put: %s", err)
+	}
+
+	payload, err := state.getStatePayload()
+	if err != nil {
+		t.Fatalf("get: %s", err)
+	}
+	if !bytes.Equal(payload.Data, data) {
+		t.Fatalf("expected full state %q\n\ngot: %q", string(payload.Data), string(data))
+	}
+
+	if err := state.Delete(); err != nil {
+		t.Fatalf("delete: %s", err)
+	}
+
+	p, err := state.getStatePayload()
+	if err != nil {
+		t.Fatalf("get: %s", err)
+	}
+	if p != nil {
+		t.Fatalf("expected empty state, got: %q", string(p.Data))
+	}
+}
+
+func TestCloudLocks(t *testing.T) {
+	back, bCleanup := testBackendWithName(t)
+	defer bCleanup()
+
+	a, err := back.StateMgr(testBackendSingleWorkspaceName)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	b, err := back.StateMgr(testBackendSingleWorkspaceName)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	lockerA, ok := a.(statemgr.Locker)
+	if !ok {
+		t.Fatal("client A not a statemgr.Locker")
+	}
+
+	lockerB, ok := b.(statemgr.Locker)
+	if !ok {
+		t.Fatal("client B not a statemgr.Locker")
+	}
+
+	infoA := statemgr.NewLockInfo()
+	infoA.Operation = "test"
+	infoA.Who = "clientA"
+
+	infoB := statemgr.NewLockInfo()
+	infoB.Operation = "test"
+	infoB.Who = "clientB"
+
+	lockIDA, err := lockerA.Lock(infoA)
+	if err != nil {
+		t.Fatal("unable to get initial lock:", err)
+	}
+
+	_, err = lockerB.Lock(infoB)
+	if err == nil {
+		lockerA.Unlock(lockIDA)
+		t.Fatal("client B obtained lock while held by client A")
+	}
+	if _, ok := err.(*statemgr.LockError); !ok {
+		t.Errorf("expected a LockError, but was %t: %s", err, err)
+	}
+
+	if err := lockerA.Unlock(lockIDA); err != nil {
+		t.Fatal("error unlocking client A", err)
+	}
+
+	lockIDB, err := lockerB.Lock(infoB)
+	if err != nil {
+		t.Fatal("unable to obtain lock from client B")
+	}
+
+	if lockIDB == lockIDA {
+		t.Fatalf("duplicate lock IDs: %q", lockIDB)
+	}
+
+	if err = lockerB.Unlock(lockIDB); err != nil {
+		t.Fatal("error unlocking client B:", err)
+	}
+}
diff --git a/internal/cloud/testing.go b/internal/cloud/testing.go
index cfb49cf9b352..6cfe898337d2 100644
--- a/internal/cloud/testing.go
+++ b/internal/cloud/testing.go
@@ -23,7 +23,6 @@ import (
 	"github.com/hashicorp/terraform/internal/configs/configschema"
 	"github.com/hashicorp/terraform/internal/httpclient"
 	"github.com/hashicorp/terraform/internal/providers"
-	"github.com/hashicorp/terraform/internal/states/remote"
 	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/hashicorp/terraform/version"
@@ -110,7 +109,7 @@ func testBackendNoOperations(t *testing.T) (*Cloud, func()) {
 	return testBackend(t, obj)
 }
 
-func testRemoteClient(t *testing.T) remote.Client {
+func testCloudState(t *testing.T) *State {
 	b, bCleanup := testBackendWithName(t)
 	defer bCleanup()
 
@@ -119,7 +118,7 @@ func testRemoteClient(t *testing.T) remote.Client {
 		t.Fatalf("error: %v", err)
 	}
 
-	return raw.(*State).Client
+	return raw.(*State)
 }
 
 func testBackendWithOutputs(t *testing.T) (*Cloud, func()) {
diff --git a/internal/command/command_test.go b/internal/command/command_test.go
index 09b2bb793016..fa498b438270 100644
--- a/internal/command/command_test.go
+++ b/internal/command/command_test.go
@@ -7,7 +7,6 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"github.com/google/go-cmp/cmp"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -20,6 +19,8 @@ import (
 	"syscall"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
+
 	svchost "github.com/hashicorp/terraform-svchost"
 	"github.com/hashicorp/terraform-svchost/disco"
 	"github.com/hashicorp/terraform/internal/addrs"
@@ -137,6 +138,9 @@ func metaOverridesForProvider(p providers.Interface) *testingOverrides {
 		Providers: map[addrs.Provider]providers.Factory{
 			addrs.NewDefaultProvider("test"):                                           providers.FactoryFixed(p),
 			addrs.NewProvider(addrs.DefaultProviderRegistryHost, "hashicorp2", "test"): providers.FactoryFixed(p),
+			addrs.NewLegacyProvider("null"):                                            providers.FactoryFixed(p),
+			addrs.NewLegacyProvider("azurerm"):                                         providers.FactoryFixed(p),
+			addrs.NewProvider(addrs.DefaultProviderRegistryHost, "acmecorp", "aws"):    providers.FactoryFixed(p),
 		},
 	}
 }
diff --git a/internal/command/helper.go b/internal/command/helper.go
new file mode 100644
index 000000000000..acb6c26af0df
--- /dev/null
+++ b/internal/command/helper.go
@@ -0,0 +1,28 @@
+package command
+
+import (
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/internal/cloud"
+)
+
+const failedToLoadSchemasMessage = `
+Warning: Failed to update data for external integrations
+
+Terraform was unable to generate a description of the updated
+state for use with external integrations in Terraform Cloud.
+Any integrations configured for this workspace which depend on
+information from the state may not work correctly when using the
+result of this action.
+
+This problem occurs when Terraform cannot read the schema for
+one or more of the providers used in the state. The next successful
+apply will correct the problem by re-generating the JSON description
+of the state:
+    terraform apply
+`
+
+func isCloudMode(b backend.Enhanced) bool {
+	_, ok := b.(*cloud.Cloud)
+
+	return ok
+}
diff --git a/internal/command/import.go b/internal/command/import.go
index 51d3895e73e9..28984d51f65e 100644
--- a/internal/command/import.go
+++ b/internal/command/import.go
@@ -248,13 +248,21 @@ func (c *ImportCommand) Run(args []string) int {
 		return 1
 	}
 
+	// Get schemas, if possible, before writing state
+	var schemas *terraform.Schemas
+	if isCloudMode(b) {
+		var schemaDiags tfdiags.Diagnostics
+		schemas, schemaDiags = c.MaybeGetSchemas(newState, nil)
+		diags = diags.Append(schemaDiags)
+	}
+
 	// Persist the final state
 	log.Printf("[INFO] Writing state output to: %s", c.Meta.StateOutPath())
 	if err := state.WriteState(newState); err != nil {
 		c.Ui.Error(fmt.Sprintf("Error writing state file: %s", err))
 		return 1
 	}
-	if err := state.PersistState(); err != nil {
+	if err := state.PersistState(schemas); err != nil {
 		c.Ui.Error(fmt.Sprintf("Error writing state file: %s", err))
 		return 1
 	}
diff --git a/internal/command/meta.go b/internal/command/meta.go
index 594292f1b95c..de5a92118888 100644
--- a/internal/command/meta.go
+++ b/internal/command/meta.go
@@ -27,11 +27,13 @@ import (
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/command/webbrowser"
 	"github.com/hashicorp/terraform/internal/command/workdir"
+	"github.com/hashicorp/terraform/internal/configs"
 	"github.com/hashicorp/terraform/internal/configs/configload"
 	"github.com/hashicorp/terraform/internal/getproviders"
 	legacy "github.com/hashicorp/terraform/internal/legacy/terraform"
 	"github.com/hashicorp/terraform/internal/providers"
 	"github.com/hashicorp/terraform/internal/provisioners"
+	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/terminal"
 	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
@@ -779,3 +781,48 @@ func (m *Meta) checkRequiredVersion() tfdiags.Diagnostics {
 
 	return nil
 }
+
+// MaybeGetSchemas attempts to load and return the schemas
+// If there is not enough information to return the schemas,
+// it could potentially return nil without errors. It is the
+// responsibility of the caller to handle the lack of schema
+// information accordingly
+func (c *Meta) MaybeGetSchemas(state *states.State, config *configs.Config) (*terraform.Schemas, tfdiags.Diagnostics) {
+	var diags tfdiags.Diagnostics
+
+	path, err := os.Getwd()
+	if err != nil {
+		diags.Append(tfdiags.SimpleWarning(failedToLoadSchemasMessage))
+		return nil, diags
+	}
+
+	if config == nil {
+		config, diags = c.loadConfig(path)
+		if diags.HasErrors() {
+			diags.Append(tfdiags.SimpleWarning(failedToLoadSchemasMessage))
+			return nil, diags
+		}
+	}
+
+	if config != nil || state != nil {
+		opts, err := c.contextOpts()
+		if err != nil {
+			diags = diags.Append(err)
+			return nil, diags
+		}
+		tfCtx, ctxDiags := terraform.NewContext(opts)
+		diags = diags.Append(ctxDiags)
+		if ctxDiags.HasErrors() {
+			return nil, diags
+		}
+		var schemaDiags tfdiags.Diagnostics
+		schemas, schemaDiags := tfCtx.Schemas(config, state)
+		diags = diags.Append(schemaDiags)
+		if schemaDiags.HasErrors() {
+			return nil, diags
+		}
+		return schemas, diags
+
+	}
+	return nil, diags
+}
diff --git a/internal/command/meta_backend.go b/internal/command/meta_backend.go
index 43e49152cf71..d7fa62e00121 100644
--- a/internal/command/meta_backend.go
+++ b/internal/command/meta_backend.go
@@ -998,7 +998,7 @@ func (m *Meta) backend_C_r_s(c *configs.Backend, cHash int, sMgr *clistate.Local
 					diags = diags.Append(fmt.Errorf(errBackendMigrateLocalDelete, err))
 					return nil, diags
 				}
-				if err := localState.PersistState(); err != nil {
+				if err := localState.PersistState(nil); err != nil {
 					diags = diags.Append(fmt.Errorf(errBackendMigrateLocalDelete, err))
 					return nil, diags
 				}
diff --git a/internal/command/meta_backend_migrate.go b/internal/command/meta_backend_migrate.go
index 8cd011eff077..46161cd623a8 100644
--- a/internal/command/meta_backend_migrate.go
+++ b/internal/command/meta_backend_migrate.go
@@ -438,7 +438,11 @@ func (m *Meta) backendMigrateState_s_s(opts *backendMigrateOpts) error {
 		return fmt.Errorf(strings.TrimSpace(errBackendStateCopy),
 			opts.SourceType, opts.DestinationType, err)
 	}
-	if err := destinationState.PersistState(); err != nil {
+	// The backend is currently handled before providers are installed during init,
+	// so requiring schemas here could lead to a catch-22 where it requires some manual
+	// intervention to proceed far enough for provider installation. To avoid this,
+	// when migrating to TFC backend, the initial JSON varient of state won't be generated and stored.
+	if err := destinationState.PersistState(nil); err != nil {
 		return fmt.Errorf(strings.TrimSpace(errBackendStateCopy),
 			opts.SourceType, opts.DestinationType, err)
 	}
@@ -960,7 +964,7 @@ This will attempt to copy (with permission) all workspaces again.
 `
 
 const errBackendStateCopy = `
-Error copying state from the previous %q backend to the newly configured 
+Error copying state from the previous %q backend to the newly configured
 %q backend:
     %s
 
diff --git a/internal/command/meta_backend_test.go b/internal/command/meta_backend_test.go
index 111e136cee5b..2fa61b618f7f 100644
--- a/internal/command/meta_backend_test.go
+++ b/internal/command/meta_backend_test.go
@@ -45,7 +45,7 @@ func TestMetaBackend_emptyDir(t *testing.T) {
 		t.Fatalf("unexpected error: %s", err)
 	}
 	s.WriteState(testState())
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -134,7 +134,7 @@ func TestMetaBackend_emptyWithDefaultState(t *testing.T) {
 	next := testState()
 	next.RootModule().SetOutputValue("foo", cty.StringVal("bar"), false)
 	s.WriteState(next)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -205,7 +205,7 @@ func TestMetaBackend_emptyWithExplicitState(t *testing.T) {
 	next := testState()
 	markStateForMatching(next, "bar") // just any change so it shows as different than before
 	s.WriteState(next)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -265,7 +265,7 @@ func TestMetaBackend_configureNew(t *testing.T) {
 	mark := markStateForMatching(state, "changing")
 
 	s.WriteState(state)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -339,7 +339,7 @@ func TestMetaBackend_configureNewWithState(t *testing.T) {
 	state = states.NewState()
 	mark := markStateForMatching(state, "changing")
 
-	if err := statemgr.WriteAndPersist(s, state); err != nil {
+	if err := statemgr.WriteAndPersist(s, state, nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -505,7 +505,7 @@ func TestMetaBackend_configureNewWithStateExisting(t *testing.T) {
 	mark := markStateForMatching(state, "changing")
 
 	s.WriteState(state)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -576,7 +576,7 @@ func TestMetaBackend_configureNewWithStateExistingNoMigrate(t *testing.T) {
 	state = states.NewState()
 	mark := markStateForMatching(state, "changing")
 	s.WriteState(state)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -695,7 +695,7 @@ func TestMetaBackend_configuredChange(t *testing.T) {
 	mark := markStateForMatching(state, "changing")
 
 	s.WriteState(state)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -1448,7 +1448,7 @@ func TestMetaBackend_configuredUnset(t *testing.T) {
 
 	// Write some state
 	s.WriteState(testState())
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -1506,7 +1506,7 @@ func TestMetaBackend_configuredUnsetCopy(t *testing.T) {
 
 	// Write some state
 	s.WriteState(testState())
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -1585,7 +1585,7 @@ func TestMetaBackend_planLocal(t *testing.T) {
 	mark := markStateForMatching(state, "changing")
 
 	s.WriteState(state)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -1686,7 +1686,7 @@ func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	mark := markStateForMatching(state, "changing")
 
 	s.WriteState(state)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
@@ -1773,7 +1773,7 @@ func TestMetaBackend_planLocalMatch(t *testing.T) {
 	mark := markStateForMatching(state, "changing")
 
 	s.WriteState(state)
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 
diff --git a/internal/command/show.go b/internal/command/show.go
index 123b86536eab..3e58a979a870 100644
--- a/internal/command/show.go
+++ b/internal/command/show.go
@@ -110,20 +110,8 @@ func (c *ShowCommand) show(path string) (*plans.Plan, *statefile.File, *configs.
 
 	// Get schemas, if possible
 	if config != nil || stateFile != nil {
-		opts, err := c.contextOpts()
-		if err != nil {
-			diags = diags.Append(err)
-			return plan, stateFile, config, schemas, diags
-		}
-		tfCtx, ctxDiags := terraform.NewContext(opts)
-		diags = diags.Append(ctxDiags)
-		if ctxDiags.HasErrors() {
-			return plan, stateFile, config, schemas, diags
-		}
-		var schemaDiags tfdiags.Diagnostics
-		schemas, schemaDiags = tfCtx.Schemas(config, stateFile.State)
-		diags = diags.Append(schemaDiags)
-		if schemaDiags.HasErrors() {
+		schemas, diags = c.MaybeGetSchemas(stateFile.State, config)
+		if diags.HasErrors() {
 			return plan, stateFile, config, schemas, diags
 		}
 	}
diff --git a/internal/command/state_mv.go b/internal/command/state_mv.go
index 949f6c4b459d..feb650ac886f 100644
--- a/internal/command/state_mv.go
+++ b/internal/command/state_mv.go
@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/terraform/internal/command/clistate"
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/mitchellh/cli"
 )
@@ -385,12 +386,27 @@ func (c *StateMvCommand) Run(args []string) int {
 		return 0 // This is as far as we go in dry-run mode
 	}
 
+	b, backendDiags := c.Backend(nil)
+	diags = diags.Append(backendDiags)
+	if backendDiags.HasErrors() {
+		c.showDiagnostics(diags)
+		return 1
+	}
+
+	// Get schemas, if possible, before writing state
+	var schemas *terraform.Schemas
+	if isCloudMode(b) {
+		var schemaDiags tfdiags.Diagnostics
+		schemas, schemaDiags = c.MaybeGetSchemas(stateTo, nil)
+		diags = diags.Append(schemaDiags)
+	}
+
 	// Write the new state
 	if err := stateToMgr.WriteState(stateTo); err != nil {
 		c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 		return 1
 	}
-	if err := stateToMgr.PersistState(); err != nil {
+	if err := stateToMgr.PersistState(schemas); err != nil {
 		c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 		return 1
 	}
@@ -401,7 +417,7 @@ func (c *StateMvCommand) Run(args []string) int {
 			c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 			return 1
 		}
-		if err := stateFromMgr.PersistState(); err != nil {
+		if err := stateFromMgr.PersistState(schemas); err != nil {
 			c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 			return 1
 		}
diff --git a/internal/command/state_push.go b/internal/command/state_push.go
index 0b863740c59f..c738dc70e36a 100644
--- a/internal/command/state_push.go
+++ b/internal/command/state_push.go
@@ -11,6 +11,8 @@ import (
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
+	"github.com/hashicorp/terraform/internal/terraform"
+	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/mitchellh/cli"
 )
 
@@ -126,15 +128,24 @@ func (c *StatePushCommand) Run(args []string) int {
 		c.Ui.Error(fmt.Sprintf("Failed to write state: %s", err))
 		return 1
 	}
+
+	// Get schemas, if possible, before writing state
+	var schemas *terraform.Schemas
+	var diags tfdiags.Diagnostics
+	if isCloudMode(b) {
+		schemas, diags = c.MaybeGetSchemas(srcStateFile.State, nil)
+	}
+
 	if err := stateMgr.WriteState(srcStateFile.State); err != nil {
 		c.Ui.Error(fmt.Sprintf("Failed to write state: %s", err))
 		return 1
 	}
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(schemas); err != nil {
 		c.Ui.Error(fmt.Sprintf("Failed to persist state: %s", err))
 		return 1
 	}
 
+	c.showDiagnostics(diags)
 	return 0
 }
 
diff --git a/internal/command/state_push_test.go b/internal/command/state_push_test.go
index e30010bb9eac..f79efa9b3b5e 100644
--- a/internal/command/state_push_test.go
+++ b/internal/command/state_push_test.go
@@ -267,7 +267,7 @@ func TestStatePush_forceRemoteState(t *testing.T) {
 	if err := sMgr.WriteState(states.NewState()); err != nil {
 		t.Fatal(err)
 	}
-	if err := sMgr.PersistState(); err != nil {
+	if err := sMgr.PersistState(nil); err != nil {
 		t.Fatal(err)
 	}
 
diff --git a/internal/command/state_replace_provider.go b/internal/command/state_replace_provider.go
index ec5347a7697a..42fdc6255907 100644
--- a/internal/command/state_replace_provider.go
+++ b/internal/command/state_replace_provider.go
@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/terraform/internal/command/clistate"
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/mitchellh/cli"
 )
@@ -160,16 +161,32 @@ func (c *StateReplaceProviderCommand) Run(args []string) int {
 		resource.ProviderConfig.Provider = to
 	}
 
+	b, backendDiags := c.Backend(nil)
+	diags = diags.Append(backendDiags)
+	if backendDiags.HasErrors() {
+		c.showDiagnostics(diags)
+		return 1
+	}
+
+	// Get schemas, if possible, before writing state
+	var schemas *terraform.Schemas
+	if isCloudMode(b) {
+		var schemaDiags tfdiags.Diagnostics
+		schemas, schemaDiags = c.MaybeGetSchemas(state, nil)
+		diags = diags.Append(schemaDiags)
+	}
+
 	// Write the updated state
 	if err := stateMgr.WriteState(state); err != nil {
 		c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 		return 1
 	}
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(schemas); err != nil {
 		c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 		return 1
 	}
 
+	c.showDiagnostics(diags)
 	c.Ui.Output(fmt.Sprintf("\nSuccessfully replaced provider for %d resources.", len(willReplace)))
 	return 0
 }
diff --git a/internal/command/state_rm.go b/internal/command/state_rm.go
index f126c5f5a561..77d4b1823b2a 100644
--- a/internal/command/state_rm.go
+++ b/internal/command/state_rm.go
@@ -8,6 +8,7 @@ import (
 	"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/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/mitchellh/cli"
 )
@@ -110,11 +111,26 @@ func (c *StateRmCommand) Run(args []string) int {
 		return 0 // This is as far as we go in dry-run mode
 	}
 
+	b, backendDiags := c.Backend(nil)
+	diags = diags.Append(backendDiags)
+	if backendDiags.HasErrors() {
+		c.showDiagnostics(diags)
+		return 1
+	}
+
+	// Get schemas, if possible, before writing state
+	var schemas *terraform.Schemas
+	if isCloudMode(b) {
+		var schemaDiags tfdiags.Diagnostics
+		schemas, schemaDiags = c.MaybeGetSchemas(state, nil)
+		diags = diags.Append(schemaDiags)
+	}
+
 	if err := stateMgr.WriteState(state); err != nil {
 		c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 		return 1
 	}
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(schemas); err != nil {
 		c.Ui.Error(fmt.Sprintf(errStateRmPersist, err))
 		return 1
 	}
diff --git a/internal/command/taint.go b/internal/command/taint.go
index 0c5a499f2e32..e4da31d9b2ef 100644
--- a/internal/command/taint.go
+++ b/internal/command/taint.go
@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/terraform/internal/command/clistate"
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 )
 
@@ -125,6 +126,14 @@ func (c *TaintCommand) Run(args []string) int {
 		return 1
 	}
 
+	// Get schemas, if possible, before writing state
+	var schemas *terraform.Schemas
+	if isCloudMode(b) {
+		var schemaDiags tfdiags.Diagnostics
+		schemas, schemaDiags = c.MaybeGetSchemas(state, nil)
+		diags = diags.Append(schemaDiags)
+	}
+
 	ss := state.SyncWrapper()
 
 	// Get the resource and instance we're going to taint
@@ -171,11 +180,12 @@ func (c *TaintCommand) Run(args []string) int {
 		c.Ui.Error(fmt.Sprintf("Error writing state file: %s", err))
 		return 1
 	}
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(schemas); err != nil {
 		c.Ui.Error(fmt.Sprintf("Error writing state file: %s", err))
 		return 1
 	}
 
+	c.showDiagnostics(diags)
 	c.Ui.Output(fmt.Sprintf("Resource instance %s has been marked as tainted.", addr))
 	return 0
 }
diff --git a/internal/command/untaint.go b/internal/command/untaint.go
index ba290a8a47a9..d02a794cfeab 100644
--- a/internal/command/untaint.go
+++ b/internal/command/untaint.go
@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/terraform/internal/command/clistate"
 	"github.com/hashicorp/terraform/internal/command/views"
 	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 )
 
@@ -163,6 +164,15 @@ func (c *UntaintCommand) Run(args []string) int {
 		c.showDiagnostics(diags)
 		return 1
 	}
+
+	// Get schemas, if possible, before writing state
+	var schemas *terraform.Schemas
+	if isCloudMode(b) {
+		var schemaDiags tfdiags.Diagnostics
+		schemas, schemaDiags = c.MaybeGetSchemas(state, nil)
+		diags = diags.Append(schemaDiags)
+	}
+
 	obj.Status = states.ObjectReady
 	ss.SetResourceInstanceCurrent(addr, obj, rs.ProviderConfig)
 
@@ -170,11 +180,12 @@ func (c *UntaintCommand) Run(args []string) int {
 		c.Ui.Error(fmt.Sprintf("Error writing state file: %s", err))
 		return 1
 	}
-	if err := stateMgr.PersistState(); err != nil {
+	if err := stateMgr.PersistState(schemas); err != nil {
 		c.Ui.Error(fmt.Sprintf("Error writing state file: %s", err))
 		return 1
 	}
 
+	c.showDiagnostics(diags)
 	c.Ui.Output(fmt.Sprintf("Resource instance %s has been successfully untainted.", addr))
 	return 0
 }
diff --git a/internal/command/workspace_new.go b/internal/command/workspace_new.go
index cd28e6986744..38985622079f 100644
--- a/internal/command/workspace_new.go
+++ b/internal/command/workspace_new.go
@@ -156,7 +156,7 @@ func (c *WorkspaceNewCommand) Run(args []string) int {
 		c.Ui.Error(err.Error())
 		return 1
 	}
-	err = stateMgr.PersistState()
+	err = stateMgr.PersistState(nil)
 	if err != nil {
 		c.Ui.Error(err.Error())
 		return 1
diff --git a/internal/states/remote/state.go b/internal/states/remote/state.go
index ae123f8e3647..412adc3eb4f4 100644
--- a/internal/states/remote/state.go
+++ b/internal/states/remote/state.go
@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/statefile"
 	"github.com/hashicorp/terraform/internal/states/statemgr"
+	"github.com/hashicorp/terraform/internal/terraform"
 )
 
 // State implements the State interfaces in the state package to handle
@@ -153,7 +154,7 @@ func (s *State) refreshState() error {
 }
 
 // statemgr.Persister impl.
-func (s *State) PersistState() error {
+func (s *State) PersistState(schemas *terraform.Schemas) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
diff --git a/internal/states/remote/state_test.go b/internal/states/remote/state_test.go
index 1089ba1aaeed..721a7a0af38c 100644
--- a/internal/states/remote/state_test.go
+++ b/internal/states/remote/state_test.go
@@ -37,7 +37,7 @@ func TestStateRace(t *testing.T) {
 		go func() {
 			defer wg.Done()
 			s.WriteState(current)
-			s.PersistState()
+			s.PersistState(nil)
 			s.RefreshState()
 		}()
 	}
@@ -252,7 +252,7 @@ func TestStatePersist(t *testing.T) {
 			if err := mgr.WriteState(s); err != nil {
 				t.Fatalf("failed to WriteState for %q: %s", tc.name, err)
 			}
-			if err := mgr.PersistState(); err != nil {
+			if err := mgr.PersistState(nil); err != nil {
 				t.Fatalf("failed to PersistState for %q: %s", tc.name, err)
 			}
 
@@ -447,7 +447,7 @@ func TestWriteStateForMigration(t *testing.T) {
 			// At this point we should just do a normal write and persist
 			// as would happen from the CLI
 			mgr.WriteState(mgr.State())
-			mgr.PersistState()
+			mgr.PersistState(nil)
 
 			if logIdx >= len(mockClient.log) {
 				t.Fatalf("request lock and index are out of sync on %q: idx=%d len=%d", tc.name, logIdx, len(mockClient.log))
@@ -611,7 +611,7 @@ func TestWriteStateForMigrationWithForcePushClient(t *testing.T) {
 			// At this point we should just do a normal write and persist
 			// as would happen from the CLI
 			mgr.WriteState(mgr.State())
-			mgr.PersistState()
+			mgr.PersistState(nil)
 
 			if logIdx >= len(mockClient.log) {
 				t.Fatalf("request lock and index are out of sync on %q: idx=%d len=%d", tc.name, logIdx, len(mockClient.log))
diff --git a/internal/states/statemgr/filesystem.go b/internal/states/statemgr/filesystem.go
index 1406f046563b..bdfc6832b5dd 100644
--- a/internal/states/statemgr/filesystem.go
+++ b/internal/states/statemgr/filesystem.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/statefile"
+	"github.com/hashicorp/terraform/internal/terraform"
 )
 
 // Filesystem is a full state manager that uses a file in the local filesystem
@@ -223,7 +224,7 @@ func (s *Filesystem) writeState(state *states.State, meta *SnapshotMeta) error {
 
 // PersistState is an implementation of Persister that does nothing because
 // this type's Writer implementation does its own persistence.
-func (s *Filesystem) PersistState() error {
+func (s *Filesystem) PersistState(schemas *terraform.Schemas) error {
 	return nil
 }
 
diff --git a/internal/states/statemgr/helper.go b/internal/states/statemgr/helper.go
index a019b2c431a6..6cae85702ea9 100644
--- a/internal/states/statemgr/helper.go
+++ b/internal/states/statemgr/helper.go
@@ -6,6 +6,7 @@ package statemgr
 import (
 	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/states/statefile"
+	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/version"
 )
 
@@ -44,10 +45,10 @@ func RefreshAndRead(mgr Storage) (*states.State, error) {
 // out quickly with a user-facing error. In situations where more control
 // is required, call WriteState and PersistState on the state manager directly
 // and handle their errors.
-func WriteAndPersist(mgr Storage, state *states.State) error {
+func WriteAndPersist(mgr Storage, state *states.State, schemas *terraform.Schemas) error {
 	err := mgr.WriteState(state)
 	if err != nil {
 		return err
 	}
-	return mgr.PersistState()
+	return mgr.PersistState(schemas)
 }
diff --git a/internal/states/statemgr/lock.go b/internal/states/statemgr/lock.go
index 79c149fe736e..863dc2f0dd18 100644
--- a/internal/states/statemgr/lock.go
+++ b/internal/states/statemgr/lock.go
@@ -1,6 +1,9 @@
 package statemgr
 
-import "github.com/hashicorp/terraform/internal/states"
+import (
+	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
+)
 
 // LockDisabled implements State and Locker but disables state locking.
 // If State doesn't support locking, this is a no-op. This is useful for
@@ -27,8 +30,8 @@ func (s *LockDisabled) RefreshState() error {
 	return s.Inner.RefreshState()
 }
 
-func (s *LockDisabled) PersistState() error {
-	return s.Inner.PersistState()
+func (s *LockDisabled) PersistState(schemas *terraform.Schemas) error {
+	return s.Inner.PersistState(schemas)
 }
 
 func (s *LockDisabled) Lock(info *LockInfo) (string, error) {
diff --git a/internal/states/statemgr/persistent.go b/internal/states/statemgr/persistent.go
index fde4a7f0f8ac..70d709f85f4e 100644
--- a/internal/states/statemgr/persistent.go
+++ b/internal/states/statemgr/persistent.go
@@ -4,6 +4,7 @@ import (
 	version "github.com/hashicorp/go-version"
 
 	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
 )
 
 // Persistent is a union of the Refresher and Persistent interfaces, for types
@@ -72,8 +73,12 @@ type Refresher interface {
 // is most commonly achieved by making use of atomic write capabilities on
 // the remote storage backend in conjunction with book-keeping with the
 // Serial and Lineage fields in the standard state file formats.
+//
+// Some implementations may optionally utilize config schema to persist
+// state. For example, when representing state in an external JSON
+// representation.
 type Persister interface {
-	PersistState() error
+	PersistState(*terraform.Schemas) error
 }
 
 // PersistentMeta is an optional extension to Persistent that allows inspecting
diff --git a/internal/states/statemgr/statemgr_fake.go b/internal/states/statemgr/statemgr_fake.go
index 8d88e4d24e7e..985e6c677517 100644
--- a/internal/states/statemgr/statemgr_fake.go
+++ b/internal/states/statemgr/statemgr_fake.go
@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/terraform"
 )
 
 // NewFullFake returns a full state manager that really only supports transient
@@ -61,7 +62,7 @@ func (m *fakeFull) RefreshState() error {
 	return m.t.WriteState(m.fakeP.State())
 }
 
-func (m *fakeFull) PersistState() error {
+func (m *fakeFull) PersistState(schemas *terraform.Schemas) error {
 	return m.fakeP.WriteState(m.t.State())
 }
 
@@ -127,7 +128,7 @@ func (m *fakeErrorFull) RefreshState() error {
 	return errors.New("fake state manager error")
 }
 
-func (m *fakeErrorFull) PersistState() error {
+func (m *fakeErrorFull) PersistState(schemas *terraform.Schemas) error {
 	return errors.New("fake state manager error")
 }
 
diff --git a/internal/states/statemgr/testing.go b/internal/states/statemgr/testing.go
index 171b21ad2ea0..eabf46dc0e40 100644
--- a/internal/states/statemgr/testing.go
+++ b/internal/states/statemgr/testing.go
@@ -56,7 +56,7 @@ func TestFull(t *testing.T, s Full) {
 	}
 
 	// Test persistence
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("err: %s", err)
 	}
 
@@ -81,7 +81,7 @@ func TestFull(t *testing.T, s Full) {
 	if err := s.WriteState(current); err != nil {
 		t.Fatalf("err: %s", err)
 	}
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("err: %s", err)
 	}
 
@@ -104,7 +104,7 @@ func TestFull(t *testing.T, s Full) {
 	if err := s.WriteState(current); err != nil {
 		t.Fatalf("err: %s", err)
 	}
-	if err := s.PersistState(); err != nil {
+	if err := s.PersistState(nil); err != nil {
 		t.Fatalf("err: %s", err)
 	}
 
diff --git a/internal/states/statemgr/transient.go b/internal/states/statemgr/transient.go
index 0ac9b1dedaaa..e47683e98bb3 100644
--- a/internal/states/statemgr/transient.go
+++ b/internal/states/statemgr/transient.go
@@ -57,7 +57,7 @@ type Reader interface {
 // since the caller may continue to modify the given state object after
 // WriteState returns.
 type Writer interface {
-	// Write state saves a transient snapshot of the given state.
+	// WriteState saves a transient snapshot of the given state.
 	//
 	// The caller must ensure that the given state object is not concurrently
 	// modified while a WriteState call is in progress. WriteState itself