Apply a confirmable run when given a saved cloud plan (#33270)

It displays a run header with link to web UI, like starting a new plan does, then confirms the run
and streams the apply logs. If you can't apply the run (it's from a different workspace, is in an
unconfirmable state, etc. etc.), it displays an error instead.

Notable points along the way:

* Implement `WrappedPlanFile` sum type, and update planfile consumers to use it instead of a plain `planfile.Reader`.

* Enable applying a saved cloud plan

* Update TFC mocks — add org name to workspace, and minimal support for includes on MockRuns.ReadWithOptions.

internal/backend/backend.go

@@ -270,7 +270,7 @@ type Operation struct {
 
 	// Plan is a plan that was passed as an argument. This is valid for
 	// plan and apply arguments but may not work for all backends.
-	PlanFile *planfile.Reader
+	PlanFile *planfile.WrappedPlanFile
 
 	// The options below are more self-explanatory and affect the runtime
 	// behavior of the operation.

internal/backend/local/backend_local.go

@@ -77,7 +77,12 @@ func (b *Local) localRun(op *backend.Operation) (*backend.LocalRun, *configload.
 
 	var ctxDiags tfdiags.Diagnostics
 	var configSnap *configload.Snapshot
-	if op.PlanFile != nil {
+	if op.PlanFile.IsCloud() {
+		diags = diags.Append(fmt.Errorf("error: using a saved cloud plan with the local backend is not supported"))
+		return nil, nil, nil, diags
+	}
+
+	if lp, ok := op.PlanFile.Local(); ok {
 		var stateMeta *statemgr.SnapshotMeta
 		// If the statemgr implements our optional PersistentMeta interface then we'll
 		// additionally verify that the state snapshot in the plan file has
@@ -87,7 +92,7 @@ func (b *Local) localRun(op *backend.Operation) (*backend.LocalRun, *configload.
 			stateMeta = &m
 		}
 		log.Printf("[TRACE] backend/local: populating backend.LocalRun from plan file")
-		ret, configSnap, ctxDiags = b.localRunForPlanFile(op, op.PlanFile, ret, &coreOpts, stateMeta)
+		ret, configSnap, ctxDiags = b.localRunForPlanFile(op, lp, ret, &coreOpts, stateMeta)
 		if ctxDiags.HasErrors() {
 			diags = diags.Append(ctxDiags)
 			return nil, nil, nil, diags

internal/backend/local/backend_local_test.go

@@ -86,6 +86,41 @@ func TestLocalRun_error(t *testing.T) {
 	assertBackendStateUnlocked(t, b)
 }
 
+func TestLocalRun_cloudPlan(t *testing.T) {
+	configDir := "./testdata/apply"
+	b := TestLocal(t)
+
+	_, configLoader, configCleanup := initwd.MustLoadConfigForTests(t, configDir)
+	defer configCleanup()
+
+	planPath := "../../cloud/cloudplan/testdata/plan-bookmark/bookmark.json"
+
+	planFile, err := planfile.OpenWrapped(planPath)
+	if err != nil {
+		t.Fatalf("unexpected error reading planfile: %s", err)
+	}
+
+	streams, _ := terminal.StreamsForTesting(t)
+	view := views.NewView(streams)
+	stateLocker := clistate.NewLocker(0, views.NewStateLocker(arguments.ViewHuman, view))
+
+	op := &backend.Operation{
+		ConfigDir:    configDir,
+		ConfigLoader: configLoader,
+		PlanFile:     planFile,
+		Workspace:    backend.DefaultStateName,
+		StateLocker:  stateLocker,
+	}
+
+	_, _, diags := b.LocalRun(op)
+	if !diags.HasErrors() {
+		t.Fatal("unexpected success")
+	}
+
+	// LocalRun() unlocks the state on failure
+	assertBackendStateUnlocked(t, b)
+}
+
 func TestLocalRun_stalePlan(t *testing.T) {
 	configDir := "./testdata/apply"
 	b := TestLocal(t)
@@ -146,7 +181,7 @@ func TestLocalRun_stalePlan(t *testing.T) {
 	if err := planfile.Create(planPath, planfileArgs); err != nil {
 		t.Fatalf("unexpected error writing planfile: %s", err)
 	}
-	planFile, err := planfile.Open(planPath)
+	planFile, err := planfile.OpenWrapped(planPath)
 	if err != nil {
 		t.Fatalf("unexpected error reading planfile: %s", err)
 	}

internal/backend/remote/backend_apply_test.go

@@ -264,7 +264,7 @@ func TestRemote_applyWithPlan(t *testing.T) {
 	op, configCleanup, done := testOperationApply(t, "./testdata/apply")
 	defer configCleanup()
 
-	op.PlanFile = &planfile.Reader{}
+	op.PlanFile = planfile.NewWrappedLocal(&planfile.Reader{})
 	op.Workspace = backend.DefaultStateName
 
 	run, err := b.Operation(context.Background(), op)

internal/backend/remote/backend_plan_test.go

@@ -239,7 +239,7 @@ func TestRemote_planWithPlan(t *testing.T) {
 	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
 	defer configCleanup()
 
-	op.PlanFile = &planfile.Reader{}
+	op.PlanFile = planfile.NewWrappedLocal(&planfile.Reader{})
 	op.Workspace = backend.DefaultStateName
 
 	run, err := b.Operation(context.Background(), op)

internal/cloud/backend_apply.go

@@ -7,8 +7,10 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"log"
+	"strings"
 
 	tfe "github.com/hashicorp/go-tfe"
 	"github.com/hashicorp/terraform/internal/backend"
@@ -54,12 +56,12 @@ func (b *Cloud) opApply(stopCtx, cancelCtx context.Context, op *backend.Operatio
 		))
 	}
 
-	if op.PlanFile != nil {
+	if op.PlanFile.IsLocal() {
 		diags = diags.Append(tfdiags.Sourceless(
 			tfdiags.Error,
-			"Applying a saved plan is currently not supported",
-			`Terraform Cloud currently requires configuration to be present and `+
-				`does not accept an existing saved plan as an argument at this time.`,
+			"Applying a saved local plan is not supported",
+			`Terraform Cloud can apply a saved cloud plan, or create a new plan when `+
+				`configuration is present. It cannot apply a saved local plan.`,
 		))
 	}
 
@@ -79,59 +81,107 @@ func (b *Cloud) opApply(stopCtx, cancelCtx context.Context, op *backend.Operatio
 		return nil, diags.Err()
 	}
 
-	// Run the plan phase.
-	r, err := b.plan(stopCtx, cancelCtx, op, w)
-	if err != nil {
-		return r, err
-	}
+	var r *tfe.Run
+	var err error
 
-	// This check is also performed in the plan method to determine if
-	// the policies should be checked, but we need to check the values
-	// here again to determine if we are done and should return.
-	if !r.HasChanges || r.Status == tfe.RunCanceled || r.Status == tfe.RunErrored {
-		return r, nil
-	}
-
-	// Retrieve the run to get its current status.
-	r, err = b.client.Runs.Read(stopCtx, r.ID)
-	if err != nil {
-		return r, generalError("Failed to retrieve run", err)
-	}
-
-	// Return if the run cannot be confirmed.
-	if !op.AutoApprove && !r.Actions.IsConfirmable {
-		return r, nil
-	}
-
-	mustConfirm := (op.UIIn != nil && op.UIOut != nil) && !op.AutoApprove
-
-	if mustConfirm && b.input {
-		opts := &terraform.InputOpts{Id: "approve"}
-
-		if op.PlanMode == plans.DestroyMode {
-			opts.Query = "\nDo you really want to destroy all resources in workspace \"" + op.Workspace + "\"?"
-			opts.Description = "Terraform will destroy all your managed infrastructure, as shown above.\n" +
-				"There is no undo. Only 'yes' will be accepted to confirm."
-		} else {
-			opts.Query = "\nDo you want to perform these actions in workspace \"" + op.Workspace + "\"?"
-			opts.Description = "Terraform will perform the actions described above.\n" +
-				"Only 'yes' will be accepted to approve."
+	if cp, ok := op.PlanFile.Cloud(); ok {
+		log.Printf("[TRACE] Loading saved cloud plan for apply")
+		// Check hostname first, for a more actionable error than a generic 404 later
+		if cp.Hostname != b.hostname {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Saved plan is for a different hostname",
+				fmt.Sprintf("The given saved plan refers to a run on %s, but the currently configured Terraform Cloud or Terraform Enterprise instance is %s.", cp.Hostname, b.hostname),
+			))
+			return r, diags.Err()
 		}
+		// Fetch the run referenced in the saved plan bookmark.
+		r, err = b.client.Runs.ReadWithOptions(stopCtx, cp.RunID, &tfe.RunReadOptions{
+			Include: []tfe.RunIncludeOpt{tfe.RunWorkspace},
+		})
 
-		err = b.confirm(stopCtx, op, opts, r, "yes")
-		if err != nil && err != errRunApproved {
+		if err != nil {
 			return r, err
 		}
-	} else if mustConfirm && !b.input {
-		return r, errApplyNeedsUIConfirmation
-	} else {
-		// If we don't need to ask for confirmation, insert a blank
-		// line to separate the ouputs.
+
+		if r.Workspace.ID != w.ID {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Saved plan is for a different workspace",
+				fmt.Sprintf("The given saved plan does not refer to a run in the current workspace (%s/%s), so it cannot currently be applied. For more details, view this run in a browser at:\n%s", w.Organization.Name, w.Name, runURL(b.hostname, r.Workspace.Organization.Name, r.Workspace.Name, r.ID)),
+			))
+			return r, diags.Err()
+		}
+
+		if !r.Actions.IsConfirmable {
+			url := runURL(b.hostname, b.organization, op.Workspace, r.ID)
+			return r, unusableSavedPlanError(r.Status, url)
+		}
+
+		// Since we're not calling plan(), we need to print a run header ourselves:
 		if b.CLI != nil {
-			b.CLI.Output("")
+			b.CLI.Output(b.Colorize().Color(strings.TrimSpace(applySavedHeader) + "\n"))
+			b.CLI.Output(b.Colorize().Color(strings.TrimSpace(fmt.Sprintf(
+				runHeader, b.hostname, b.organization, r.Workspace.Name, r.ID)) + "\n"))
+		}
+	} else {
+		log.Printf("[TRACE] Running new cloud plan for apply")
+		// Run the plan phase.
+		r, err = b.plan(stopCtx, cancelCtx, op, w)
+
+		if err != nil {
+			return r, err
+		}
+
+		// This check is also performed in the plan method to determine if
+		// the policies should be checked, but we need to check the values
+		// here again to determine if we are done and should return.
+		if !r.HasChanges || r.Status == tfe.RunCanceled || r.Status == tfe.RunErrored {
+			return r, nil
+		}
+
+		// Retrieve the run to get its current status.
+		r, err = b.client.Runs.Read(stopCtx, r.ID)
+		if err != nil {
+			return r, generalError("Failed to retrieve run", err)
+		}
+
+		// Return if the run cannot be confirmed.
+		if !op.AutoApprove && !r.Actions.IsConfirmable {
+			return r, nil
+		}
+
+		mustConfirm := (op.UIIn != nil && op.UIOut != nil) && !op.AutoApprove
+
+		if mustConfirm && b.input {
+			opts := &terraform.InputOpts{Id: "approve"}
+
+			if op.PlanMode == plans.DestroyMode {
+				opts.Query = "\nDo you really want to destroy all resources in workspace \"" + op.Workspace + "\"?"
+				opts.Description = "Terraform will destroy all your managed infrastructure, as shown above.\n" +
+					"There is no undo. Only 'yes' will be accepted to confirm."
+			} else {
+				opts.Query = "\nDo you want to perform these actions in workspace \"" + op.Workspace + "\"?"
+				opts.Description = "Terraform will perform the actions described above.\n" +
+					"Only 'yes' will be accepted to approve."
+			}
+
+			err = b.confirm(stopCtx, op, opts, r, "yes")
+			if err != nil && err != errRunApproved {
+				return r, err
+			}
+		} else if mustConfirm && !b.input {
+			return r, errApplyNeedsUIConfirmation
+		} else {
+			// If we don't need to ask for confirmation, insert a blank
+			// line to separate the ouputs.
+			if b.CLI != nil {
+				b.CLI.Output("")
+			}
 		}
 	}
 
+	// Do the apply!
 	if !op.AutoApprove && err != errRunApproved {
 		if err = b.client.Runs.Apply(stopCtx, r.ID, tfe.RunApplyOptions{}); err != nil {
 			return r, generalError("Failed to approve the apply command", err)
@@ -222,6 +272,52 @@ func (b *Cloud) renderApplyLogs(ctx context.Context, run *tfe.Run) error {
 	return nil
 }
 
+func runURL(hostname, orgName, wsName, runID string) string {
+	return fmt.Sprintf("https://%s/app/%s/%s/runs/%s", hostname, orgName, wsName, runID)
+}
+
+func unusableSavedPlanError(status tfe.RunStatus, url string) error {
+	var diags tfdiags.Diagnostics
+	var summary, reason string
+
+	switch status {
+	case tfe.RunApplied:
+		summary = "Saved plan is already applied"
+		reason = "The given plan file was already successfully applied, and cannot be applied again."
+	case tfe.RunApplying, tfe.RunApplyQueued, tfe.RunConfirmed:
+		summary = "Saved plan is already confirmed"
+		reason = "The given plan file is already being applied, and cannot be applied again."
+	case tfe.RunCanceled:
+		summary = "Saved plan is canceled"
+		reason = "The given plan file can no longer be applied because the run was canceled via the Terraform Cloud UI or API."
+	case tfe.RunDiscarded:
+		summary = "Saved plan is discarded"
+		reason = "The given plan file can no longer be applied; either another run was applied first, or a user discarded it via the Terraform Cloud UI or API."
+	case tfe.RunErrored:
+		summary = "Saved plan is errored"
+		reason = "The given plan file refers to a plan that had errors and did not complete successfully. It cannot be applied."
+	case tfe.RunPlannedAndFinished:
+		// Note: planned and finished can also indicate a plan-only run, but
+		// terraform plan can't create a saved plan for a plan-only run, so we
+		// know it's no-changes in this case.
+		summary = "Saved plan has no changes"
+		reason = "The given plan file contains no changes, so it cannot be applied."
+	case tfe.RunPolicyOverride:
+		summary = "Saved plan requires policy override"
+		reason = "The given plan file has soft policy failures, and cannot be applied until a user with appropriate permissions overrides the policy check."
+	default:
+		summary = "Saved plan cannot be applied"
+		reason = "Terraform Cloud cannot apply the given plan file. This may mean the plan and checks have not yet completed, or may indicate another problem."
+	}
+
+	diags = diags.Append(tfdiags.Sourceless(
+		tfdiags.Error,
+		summary,
+		fmt.Sprintf("%s For more details, view this run in a browser at:\n%s", reason, url),
+	))
+	return diags.Err()
+}
+
 const applyDefaultHeader = `
 [reset][yellow]Running apply in Terraform Cloud. Output will stream here. Pressing Ctrl-C
 will cancel the remote apply if it's still pending. If the apply started it
@@ -229,3 +325,10 @@ will stop streaming the logs, but will not stop the apply running remotely.[rese
 
 Preparing the remote apply...
 `
+
+const applySavedHeader = `
+[reset][yellow]Running apply in Terraform Cloud. Output will stream here. Pressing Ctrl-C
+will stop streaming the logs, but will not stop the apply running remotely.[reset]
+
+Preparing the remote apply...
+`

internal/cloud/backend_apply_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/hashicorp/terraform/internal/addrs"
 	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/internal/cloud/cloudplan"
 	"github.com/hashicorp/terraform/internal/command/arguments"
 	"github.com/hashicorp/terraform/internal/command/clistate"
 	"github.com/hashicorp/terraform/internal/command/jsonformat"
@@ -407,14 +408,15 @@ func TestCloud_applyWithParallelism(t *testing.T) {
 	}
 }
 
-func TestCloud_applyWithPlan(t *testing.T) {
+// Apply with local plan file should fail.
+func TestCloud_applyWithLocalPlan(t *testing.T) {
 	b, bCleanup := testBackendWithName(t)
 	defer bCleanup()
 
 	op, configCleanup, done := testOperationApply(t, "./testdata/apply")
 	defer configCleanup()
 
-	op.PlanFile = &planfile.Reader{}
+	op.PlanFile = planfile.NewWrappedLocal(&planfile.Reader{})
 	op.Workspace = testBackendSingleWorkspaceName
 
 	run, err := b.Operation(context.Background(), op)
@@ -432,11 +434,80 @@ func TestCloud_applyWithPlan(t *testing.T) {
 	}
 
 	errOutput := output.Stderr()
-	if !strings.Contains(errOutput, "saved plan is currently not supported") {
+	if !strings.Contains(errOutput, "saved local plan is not supported") {
 		t.Fatalf("expected a saved plan error, got: %v", errOutput)
 	}
 }
 
+// Apply with bookmark to an existing cloud plan that's in a confirmable state
+// should work.
+func TestCloud_applyWithCloudPlan(t *testing.T) {
+	b, bCleanup := testBackendWithName(t)
+	defer bCleanup()
+
+	op, configCleanup, done := testOperationApply(t, "./testdata/apply-json")
+	defer configCleanup()
+	defer done(t)
+
+	op.UIOut = b.CLI
+	op.Workspace = testBackendSingleWorkspaceName
+
+	mockSROWorkspace(t, b, op.Workspace)
+
+	// Perform the plan before trying to apply it
+	ws, err := b.client.Workspaces.Read(context.Background(), b.organization, b.WorkspaceMapping.Name)
+	if err != nil {
+		t.Fatalf("Couldn't read workspace: %s", err)
+	}
+
+	planRun, err := b.plan(context.Background(), context.Background(), op, ws)
+	if err != nil {
+		t.Fatalf("Couldn't perform plan: %s", err)
+	}
+
+	// Synthesize a cloud plan file with the plan's run ID
+	pf := &cloudplan.SavedPlanBookmark{
+		RemotePlanFormat: 1,
+		RunID:            planRun.ID,
+		Hostname:         b.hostname,
+	}
+	op.PlanFile = planfile.NewWrappedCloud(pf)
+
+	// Start spying on the apply output (now that the plan's done)
+	stream, close := terminal.StreamsForTesting(t)
+
+	b.renderer = &jsonformat.Renderer{
+		Streams:  stream,
+		Colorize: mockColorize(),
+	}
+
+	// Try apply
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("error starting operation: %v", err)
+	}
+
+	<-run.Done()
+	output := close(t)
+	if run.Result != backend.OperationSuccess {
+		t.Fatal("expected apply operation to succeed")
+	}
+	if run.PlanEmpty {
+		t.Fatalf("expected plan to not be empty")
+	}
+
+	gotOut := output.Stdout()
+	if !strings.Contains(gotOut, "1 added, 0 changed, 0 destroyed") {
+		t.Fatalf("expected apply summary in output: %s", gotOut)
+	}
+
+	stateMgr, _ := b.StateMgr(testBackendSingleWorkspaceName)
+	// An error suggests that the state was not unlocked after apply
+	if _, err := stateMgr.Lock(statemgr.NewLockInfo()); err != nil {
+		t.Fatalf("unexpected error locking state after apply: %s", err.Error())
+	}
+}
+
 func TestCloud_applyWithoutRefresh(t *testing.T) {
 	b, bCleanup := testBackendWithName(t)
 	defer bCleanup()

internal/cloud/backend_plan_test.go

@@ -337,7 +337,7 @@ func TestCloud_planWithPlan(t *testing.T) {
 	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
 	defer configCleanup()
 
-	op.PlanFile = &planfile.Reader{}
+	op.PlanFile = planfile.NewWrappedLocal(&planfile.Reader{})
 	op.Workspace = testBackendSingleWorkspaceName
 
 	run, err := b.Operation(context.Background(), op)

internal/cloud/testing.go

@@ -322,6 +322,7 @@ func testUnconfiguredBackend(t *testing.T) (*Cloud, func()) {
 	b.client.Runs = mc.Runs
 	b.client.RunEvents = mc.RunEvents
 	b.client.StateVersions = mc.StateVersions
+	b.client.StateVersionOutputs = mc.StateVersionOutputs
 	b.client.Variables = mc.Variables
 	b.client.Workspaces = mc.Workspaces
 

internal/cloud/tfe_client_mock.go

@@ -1085,7 +1085,7 @@ func (m *MockRuns) Read(ctx context.Context, runID string) (*tfe.Run, error) {
 	return m.ReadWithOptions(ctx, runID, nil)
 }
 
-func (m *MockRuns) ReadWithOptions(ctx context.Context, runID string, _ *tfe.RunReadOptions) (*tfe.Run, error) {
+func (m *MockRuns) ReadWithOptions(ctx context.Context, runID string, options *tfe.RunReadOptions) (*tfe.Run, error) {
 	m.Lock()
 	defer m.Unlock()
 
@@ -1136,8 +1136,22 @@ func (m *MockRuns) ReadWithOptions(ctx context.Context, runID string, _ *tfe.Run
 	if err != nil {
 		panic(err)
 	}
+	r = rc.(*tfe.Run)
 
-	return rc.(*tfe.Run), nil
+	// After copying, handle includes... or at least, any includes we're known to rely on.
+	if options != nil {
+		for _, n := range options.Include {
+			switch n {
+			case tfe.RunWorkspace:
+				ws, ok := m.client.Workspaces.workspaceIDs[r.Workspace.ID]
+				if ok {
+					r.Workspace = ws
+				}
+			}
+		}
+	}
+
+	return r, nil
 }
 
 func (m *MockRuns) Apply(ctx context.Context, runID string, options tfe.RunApplyOptions) error {
@@ -1534,6 +1548,9 @@ func (m *MockWorkspaces) Create(ctx context.Context, organization string, option
 			CanQueueRun:    true,
 			CanForceDelete: tfe.Bool(true),
 		},
+		Organization: &tfe.Organization{
+			Name: organization,
+		},
 	}
 	if options.AutoApply != nil {
 		w.AutoApply = *options.AutoApply

internal/command/apply.go

@@ -150,8 +150,8 @@ func (c *ApplyCommand) Run(rawArgs []string) int {
 	return 0
 }
 
-func (c *ApplyCommand) LoadPlanFile(path string) (*planfile.Reader, tfdiags.Diagnostics) {
-	var planFile *planfile.Reader
+func (c *ApplyCommand) LoadPlanFile(path string) (*planfile.WrappedPlanFile, tfdiags.Diagnostics) {
+	var planFile *planfile.WrappedPlanFile
 	var diags tfdiags.Diagnostics
 
 	// Try to load plan if path is specified
@@ -194,7 +194,7 @@ func (c *ApplyCommand) LoadPlanFile(path string) (*planfile.Reader, tfdiags.Diag
 	return planFile, diags
 }
 
-func (c *ApplyCommand) PrepareBackend(planFile *planfile.Reader, args *arguments.State, viewType arguments.ViewType) (backend.Enhanced, tfdiags.Diagnostics) {
+func (c *ApplyCommand) PrepareBackend(planFile *planfile.WrappedPlanFile, args *arguments.State, viewType arguments.ViewType) (backend.Enhanced, tfdiags.Diagnostics) {
 	var diags tfdiags.Diagnostics
 
 	// FIXME: we need to apply the state arguments to the meta object here
@@ -206,19 +206,8 @@ func (c *ApplyCommand) PrepareBackend(planFile *planfile.Reader, args *arguments
 	// Load the backend
 	var be backend.Enhanced
 	var beDiags tfdiags.Diagnostics
-	if planFile == nil {
-		backendConfig, configDiags := c.loadBackendConfig(".")
-		diags = diags.Append(configDiags)
-		if configDiags.HasErrors() {
-			return nil, diags
-		}
-
-		be, beDiags = c.Backend(&BackendOpts{
-			Config:   backendConfig,
-			ViewType: viewType,
-		})
-	} else {
-		plan, err := planFile.ReadPlan()
+	if lp, ok := planFile.Local(); ok {
+		plan, err := lp.ReadPlan()
 		if err != nil {
 			diags = diags.Append(tfdiags.Sourceless(
 				tfdiags.Error,
@@ -236,7 +225,19 @@ func (c *ApplyCommand) PrepareBackend(planFile *planfile.Reader, args *arguments
 			))
 			return nil, diags
 		}
-		be, beDiags = c.BackendForPlan(plan.Backend)
+		be, beDiags = c.BackendForLocalPlan(plan.Backend)
+	} else {
+		// Both new plans and saved cloud plans load their backend from config.
+		backendConfig, configDiags := c.loadBackendConfig(".")
+		diags = diags.Append(configDiags)
+		if configDiags.HasErrors() {
+			return nil, diags
+		}
+
+		be, beDiags = c.Backend(&BackendOpts{
+			Config:   backendConfig,
+			ViewType: viewType,
+		})
 	}
 
 	diags = diags.Append(beDiags)
@@ -250,7 +251,7 @@ func (c *ApplyCommand) OperationRequest(
 	be backend.Enhanced,
 	view views.Apply,
 	viewType arguments.ViewType,
-	planFile *planfile.Reader,
+	planFile *planfile.WrappedPlanFile,
 	args *arguments.Operation,
 	autoApprove bool,
 ) (*backend.Operation, tfdiags.Diagnostics) {

internal/command/graph.go

@@ -55,7 +55,7 @@ func (c *GraphCommand) Run(args []string) int {
 	}
 
 	// Try to load plan if path is specified
-	var planFile *planfile.Reader
+	var planFile *planfile.WrappedPlanFile
 	if planPath != "" {
 		planFile, err = c.PlanFile(planPath)
 		if err != nil {

internal/command/meta_backend.go

@@ -295,13 +295,13 @@ func (m *Meta) selectWorkspace(b backend.Backend) error {
 	return m.SetWorkspace(workspace)
 }
 
-// BackendForPlan is similar to Backend, but uses backend settings that were
+// BackendForLocalPlan is similar to Backend, but uses backend settings that were
 // stored in a plan.
 //
 // The current workspace name is also stored as part of the plan, and so this
 // method will check that it matches the currently-selected workspace name
 // and produce error diagnostics if not.
-func (m *Meta) BackendForPlan(settings plans.Backend) (backend.Enhanced, tfdiags.Diagnostics) {
+func (m *Meta) BackendForLocalPlan(settings plans.Backend) (backend.Enhanced, tfdiags.Diagnostics) {
 	var diags tfdiags.Diagnostics
 
 	f := backendInit.Backend(settings.Type)
@@ -310,7 +310,7 @@ func (m *Meta) BackendForPlan(settings plans.Backend) (backend.Enhanced, tfdiags
 		return nil, diags
 	}
 	b := f()
-	log.Printf("[TRACE] Meta.BackendForPlan: instantiated backend of type %T", b)
+	log.Printf("[TRACE] Meta.BackendForLocalPlan: instantiated backend of type %T", b)
 
 	schema := b.ConfigSchema()
 	configVal, err := settings.Config.Decode(schema.ImpliedType())
@@ -361,7 +361,7 @@ func (m *Meta) BackendForPlan(settings plans.Backend) (backend.Enhanced, tfdiags
 
 	// Otherwise, we'll wrap our state-only remote backend in the local backend
 	// to cause any operations to be run locally.
-	log.Printf("[TRACE] Meta.Backend: backend %T does not support operations, so wrapping it in a local backend", b)
+	log.Printf("[TRACE] Meta.BackendForLocalPlan: backend %T does not support operations, so wrapping it in a local backend", b)
 	cliOpts, err := m.backendCLIOpts()
 	if err != nil {
 		diags = diags.Append(err)

internal/command/meta_backend_test.go

@@ -1550,7 +1550,7 @@ func TestMetaBackend_planLocal(t *testing.T) {
 	m := testMetaBackend(t, nil)
 
 	// Get the backend
-	b, diags := m.BackendForPlan(backendConfig)
+	b, diags := m.BackendForLocalPlan(backendConfig)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}
@@ -1651,7 +1651,7 @@ func TestMetaBackend_planLocalStatePath(t *testing.T) {
 	m.stateOutPath = statePath
 
 	// Get the backend
-	b, diags := m.BackendForPlan(plannedBackend)
+	b, diags := m.BackendForLocalPlan(plannedBackend)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}
@@ -1740,7 +1740,7 @@ func TestMetaBackend_planLocalMatch(t *testing.T) {
 	m := testMetaBackend(t, nil)
 
 	// Get the backend
-	b, diags := m.BackendForPlan(backendConfig)
+	b, diags := m.BackendForLocalPlan(backendConfig)
 	if diags.HasErrors() {
 		t.Fatal(diags.Err())
 	}

internal/command/meta_new.go

@@ -27,14 +27,15 @@ func (m *Meta) Input() bool {
 	return true
 }
 
-// PlanFile returns a reader for the plan file at the given path.
+// PlanFile loads the plan file at the given path, which might be either a local
+// or cloud plan.
 //
 // If the return value and error are both nil, the given path exists but seems
 // to be a configuration directory instead.
 //
 // Error will be non-nil if path refers to something which looks like a plan
 // file and loading the file fails.
-func (m *Meta) PlanFile(path string) (*planfile.Reader, error) {
+func (m *Meta) PlanFile(path string) (*planfile.WrappedPlanFile, error) {
 	fi, err := os.Stat(path)
 	if err != nil {
 		return nil, err
@@ -45,5 +46,5 @@ func (m *Meta) PlanFile(path string) (*planfile.Reader, error) {
 		return nil, nil
 	}
 
-	return planfile.Open(path)
+	return planfile.OpenWrapped(path)
 }

internal/plans/planfile/planfile_test.go

@@ -5,6 +5,7 @@ package planfile
 
 import (
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -100,10 +101,17 @@ func TestRoundtrip(t *testing.T) {
 		t.Fatalf("failed to create plan file: %s", err)
 	}
 
-	pr, err := Open(planFn)
+	wpf, err := OpenWrapped(planFn)
 	if err != nil {
 		t.Fatalf("failed to open plan file for reading: %s", err)
 	}
+	pr, ok := wpf.Local()
+	if !ok {
+		t.Fatalf("failed to open plan file as a local plan file")
+	}
+	if wpf.IsCloud() {
+		t.Fatalf("wrapped plan claims to be both kinds of plan at once")
+	}
 
 	t.Run("ReadPlan", func(t *testing.T) {
 		planOut, err := pr.ReadPlan()
@@ -167,3 +175,33 @@ func TestRoundtrip(t *testing.T) {
 		}
 	})
 }
+
+func TestWrappedError(t *testing.T) {
+	// Open something that isn't a cloud or local planfile: should error
+	wrongFile := "not a valid zip file"
+	_, err := OpenWrapped(filepath.Join("testdata", "test-config", "root.tf"))
+	if !strings.Contains(err.Error(), wrongFile) {
+		t.Fatalf("expected  %q, got %q", wrongFile, err)
+	}
+
+	// Open something that doesn't exist: should error
+	missingFile := "no such file or directory"
+	_, err = OpenWrapped(filepath.Join("testdata", "absent.tfplan"))
+	if !strings.Contains(err.Error(), missingFile) {
+		t.Fatalf("expected  %q, got %q", missingFile, err)
+	}
+}
+
+func TestWrappedCloud(t *testing.T) {
+	// Loading valid cloud plan results in a wrapped cloud plan
+	wpf, err := OpenWrapped(filepath.Join("testdata", "cloudplan.json"))
+	if err != nil {
+		t.Fatalf("failed to open valid cloud plan: %s", err)
+	}
+	if !wpf.IsCloud() {
+		t.Fatalf("failed to open cloud file as a cloud plan")
+	}
+	if wpf.IsLocal() {
+		t.Fatalf("wrapped plan claims to be both kinds of plan at once")
+	}
+}

internal/plans/planfile/reader.go

@@ -31,8 +31,10 @@ type Reader struct {
 	zip *zip.ReadCloser
 }
 
-// Open creates a Reader for the file at the given filename, or returns an
-// error if the file doesn't seem to be a planfile.
+// Open creates a Reader for the file at the given filename, or returns an error
+// if the file doesn't seem to be a planfile. NOTE: Most commands that accept a
+// plan file should use OpenWrapped instead, so they can support both local and
+// cloud plan files.
 func Open(filename string) (*Reader, error) {
 	r, err := zip.OpenReader(filename)
 	if err != nil {

internal/plans/planfile/testdata/cloudplan.json

@@ -0,0 +1,5 @@
+{
+  "remote_plan_format": 1,
+  "run_id": "run-GXfuHMkbyHccAGUg",
+  "hostname": "app.terraform.io"
+}

internal/plans/planfile/wrapped.go

@@ -0,0 +1,90 @@
+package planfile
+
+import (
+	"fmt"
+
+	"github.com/hashicorp/terraform/internal/cloud/cloudplan"
+)
+
+// WrappedPlanFile is a sum type that represents a saved plan, loaded from a
+// file path passed on the command line. If the specified file was a thick local
+// plan file, the Local field will be populated; if it was a bookmark for a
+// remote cloud plan, the Cloud field will be populated. In both cases, the
+// other field is expected to be nil. Finally, the outer struct is also expected
+// to be used as a pointer, so that a nil value can represent the absence of any
+// plan file.
+type WrappedPlanFile struct {
+	local *Reader
+	cloud *cloudplan.SavedPlanBookmark
+}
+
+func (w *WrappedPlanFile) IsLocal() bool {
+	return w != nil && w.local != nil
+}
+
+func (w *WrappedPlanFile) IsCloud() bool {
+	return w != nil && w.cloud != nil
+}
+
+// Local checks whether the wrapped value is a local plan file, and returns it if available.
+func (w *WrappedPlanFile) Local() (*Reader, bool) {
+	if w != nil && w.local != nil {
+		return w.local, true
+	} else {
+		return nil, false
+	}
+}
+
+// Cloud checks whether the wrapped value is a cloud plan file, and returns it if available.
+func (w *WrappedPlanFile) Cloud() (*cloudplan.SavedPlanBookmark, bool) {
+	if w != nil && w.cloud != nil {
+		return w.cloud, true
+	} else {
+		return nil, false
+	}
+}
+
+// NewWrappedLocal constructs a WrappedPlanFile from an already loaded local
+// plan file reader. Most cases should use OpenWrapped to load from disk
+// instead. If the provided reader is nil, the returned pointer is nil.
+func NewWrappedLocal(l *Reader) *WrappedPlanFile {
+	if l != nil {
+		return &WrappedPlanFile{local: l}
+	} else {
+		return nil
+	}
+}
+
+// NewWrappedCloud constructs a WrappedPlanFile from an already loaded cloud
+// plan file. Most cases should use OpenWrapped to load from disk
+// instead. If the provided plan file is nil, the returned pointer is nil.
+func NewWrappedCloud(c *cloudplan.SavedPlanBookmark) *WrappedPlanFile {
+	if c != nil {
+		return &WrappedPlanFile{cloud: c}
+	} else {
+		return nil
+	}
+}
+
+// OpenWrapped loads a local or cloud plan file from a specified file path, or
+// returns an error if the file doesn't seem to be a plan file of either kind.
+// Most consumers should use this and switch behaviors based on the kind of plan
+// they expected, rather than directly using Open.
+func OpenWrapped(filename string) (*WrappedPlanFile, error) {
+	// First, try to load it as a local planfile.
+	local, localErr := Open(filename)
+	if localErr == nil {
+		return &WrappedPlanFile{local: local}, nil
+	}
+	// Then, try to load it as a cloud plan.
+	cloud, cloudErr := cloudplan.LoadSavedPlanBookmark(filename)
+	if cloudErr == nil {
+		return &WrappedPlanFile{cloud: &cloud}, nil
+	}
+	// If neither worked, return both errors. In general we don't care to give
+	// any advice about how to fix an internal problem in a plan file, since
+	// both formats are opaque, but we do want to give the user the best chance
+	// at resolving whatever their problem was.
+	combinedErr := fmt.Errorf("couldn't load the provided path as either a local plan file (%s) or a saved cloud plan (%s)", localErr, cloudErr)
+	return nil, combinedErr
+}