tofu: Allow dynamic expressions in prevent_destroy arguments

Previously we evaluated prevent_destroy expressions immediately inside the
config loader, thereby forcing it to always be a constant expression
producing a bool value.

Now the config loader just saves whatever expression it was given and we
let the language runtime deal with it instead, which means we can allow
references to dynamically-chosen values from elsewhere in the same module.

The language runtime's "validate" phase still performs a type check for
bool that's equivalent to what we used to do during config loading to make
sure that the "tofu validate" command can catch a similar subset of
problems as it used to be able to catch, but we have more information
available during the plan phase that allows us to produce more complete and
relevant error messages, so for any expression that we can't evaluate
with a nil evaluation context we'll now let the plan phase deal with the
checks instead.

The policy for handling annoying cases such as unknown values, ephemeral
values, sensitive values, and references to local symbols like count.index
is intentionally the most conservative choice to start, because future
versions of OpenTofu can allow more once we've got more experience but
cannot permit less if we find that we've made a mistake. Future changes
could potentially make these rules a little more liberal, once we have
learned from feedback on this initial functionality.

Signed-off-by: Martin Atkins <mart@degeneration.co.uk>

internal/configs/module_merge.go

@@ -262,9 +262,8 @@ func (r *Resource) merge(or *Resource, rps map[string]*RequiredProvider) hcl.Dia
 		if or.Managed.IgnoreAllChanges {
 			r.Managed.IgnoreAllChanges = true
 		}
-		if or.Managed.PreventDestroySet {
+		if or.Managed.PreventDestroy != nil {
 			r.Managed.PreventDestroy = or.Managed.PreventDestroy
-			r.Managed.PreventDestroySet = or.Managed.PreventDestroySet
 		}
 		if len(or.Managed.Provisioners) != 0 {
 			r.Managed.Provisioners = or.Managed.Provisioners

internal/configs/resource.go

@@ -70,12 +70,11 @@ type ManagedResource struct {
 	Provisioners []*Provisioner
 
 	CreateBeforeDestroy bool
-	PreventDestroy      bool
+	PreventDestroy      hcl.Expression
 	IgnoreChanges       []hcl.Traversal
 	IgnoreAllChanges    bool
 
 	CreateBeforeDestroySet bool
-	PreventDestroySet      bool
 }
 
 func (r *Resource) moduleUniqueKey() string {
@@ -209,9 +208,7 @@ func decodeResourceBlock(block *hcl.Block, override bool) (*Resource, hcl.Diagno
 			}
 
 			if attr, exists := lcContent.Attributes["prevent_destroy"]; exists {
-				valDiags := gohcl.DecodeExpression(attr.Expr, nil, &r.Managed.PreventDestroy)
-				diags = append(diags, valDiags...)
-				r.Managed.PreventDestroySet = true
+				r.Managed.PreventDestroy = attr.Expr
 			}
 
 			if attr, exists := lcContent.Attributes["replace_triggered_by"]; exists {

internal/tofu/context_plan_test.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -18,6 +19,8 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
+	"time"
 
 	"github.com/davecgh/go-spew/spew"
 	"github.com/google/go-cmp/cmp"
@@ -1394,7 +1397,7 @@ func TestContext2Plan_preventDestroy_bad(t *testing.T) {
 
 	plan, err := ctx.Plan(context.Background(), m, state, DefaultPlanOpts)
 
-	expectedErr := "aws_instance.foo has lifecycle.prevent_destroy"
+	expectedErr := "aws_instance.foo has prevent_destroy"
 	if !strings.Contains(fmt.Sprintf("%s", err), expectedErr) {
 		if plan != nil {
 			t.Logf("%s", legacyDiffComparisonString(plan.Changes))
@@ -1442,6 +1445,378 @@ func TestContext2Plan_preventDestroy_good(t *testing.T) {
 	}
 }
 
+func TestContext2Plan_preventDestroy_dynamic(t *testing.T) {
+	// We'll run the same set of tests with equivalent configuration written
+	// with different syntaxes.
+	fixtures := map[string]string{
+		"main.tf": `
+			variable "prevent_destroy" {
+				# intentionally not constraining type here because some
+				# of the test cases intentionally use the wrong type to
+				# test how we handle that.
+				type = any
+			}
+
+			resource "test_instance" "foo" {
+				require_new = "yes"
+
+				lifecycle {
+					prevent_destroy = var.prevent_destroy
+				}
+			}
+		`,
+		// This is a JSON equivalent of the above, to make sure that HCL's
+		// slightly different treatment of template-based dynamic expressions
+		// in JSON does not interfere with the handling of prevent_destroy.
+		"main.tf.json": `
+			{
+				"variable": {
+					"prevent_destroy": {
+						"type": "any"
+					}
+				},
+				"resource": {
+					"test_instance": {
+						"foo": {
+							"require_new": "yes",
+							"lifecycle": {
+								"prevent_destroy": "${var.prevent_destroy}"
+							}
+						}
+					}
+				}
+			}
+		`,
+	}
+
+	for filename, content := range fixtures {
+		t.Run(filename, func(t *testing.T) {
+			m := testModuleInline(t, map[string]string{
+				filename: content,
+			})
+			p := testProvider("test")
+			p.PlanResourceChangeFn = testDiffFn
+			state := states.NewState()
+			root := state.EnsureModule(addrs.RootModuleInstance)
+			root.SetResourceInstanceCurrent(
+				mustResourceInstanceAddr("test_instance.foo").Resource,
+				&states.ResourceInstanceObjectSrc{
+					Status:    states.ObjectReady,
+					AttrsJSON: []byte(`{"id":"i-abc123"}`),
+				},
+				mustProviderConfig(`provider["registry.opentofu.org/hashicorp/test"]`),
+				addrs.NoKey,
+			)
+
+			tests := []struct {
+				PreventDestroy cty.Value
+				WantErr        string
+			}{
+				{
+					PreventDestroy: cty.False,
+					WantErr:        ``,
+				},
+				{
+					PreventDestroy: cty.True,
+					WantErr:        `test_instance.foo has prevent_destroy set`,
+				},
+				{
+					PreventDestroy: cty.StringVal("false"), // automatic type conversion
+					WantErr:        ``,
+				},
+				{
+					PreventDestroy: cty.StringVal("true"), // automatic type conversion
+					WantErr:        `test_instance.foo has prevent_destroy set`,
+				},
+				{
+					PreventDestroy: cty.UnknownVal(cty.Bool),
+					WantErr:        `test_instance.foo has a prevent_destroy argument but its value will not be known until the apply step`,
+				},
+				{
+					PreventDestroy: cty.DynamicVal,
+					WantErr:        `test_instance.foo has a prevent_destroy argument but its value will not be known until the apply step`,
+				},
+				{
+					PreventDestroy: cty.False.Mark(marks.Sensitive),
+					WantErr:        `test_instance.foo has a sensitive value for its prevent_destroy argument`,
+				},
+				// NOTE: can't test marks.Ephemeral here, because the test fixture's
+				// input variable is not declared as ephemeral. Refer to
+				// [TestContext2Plan_preventDestroy_dynamicEphemeral] instead.
+				// NOTE: can't test references to deprecated output values here, because
+				// the test fixture doesn't include any deprecated output values and
+				// an input variable cannot be marked as deprecated, and because
+				// deprecation causes warnings rather than errors. Refer to
+				// [TestContext2Plan_preventDestroy_dynamicDeprecated] instead.
+				{
+					PreventDestroy: cty.Zero,
+					WantErr:        `test_instance.foo has an invalid value for its prevent_destroy argument: bool required, but have number`,
+				},
+				{
+					PreventDestroy: cty.NullVal(cty.Bool),
+					WantErr:        `test_instance.foo has prevent_destroy set to null`,
+				},
+				{
+					PreventDestroy: cty.NullVal(cty.DynamicPseudoType),
+					WantErr:        `test_instance.foo has prevent_destroy set to null`,
+				},
+			}
+			for _, test := range tests {
+				t.Run(test.PreventDestroy.GoString(), func(t *testing.T) {
+					ctx := testContext2(t, &ContextOpts{
+						Providers: map[addrs.Provider]providers.Factory{
+							addrs.NewDefaultProvider("test"): testProviderFuncFixed(p),
+						},
+					})
+
+					_, diags := ctx.Plan(context.Background(), m, state, SimplePlanOpts(plans.NormalMode, InputValues{
+						"prevent_destroy": {
+							Value:      test.PreventDestroy,
+							SourceType: ValueFromCaller,
+						},
+					}))
+					if test.WantErr == "" {
+						if diags.HasErrors() {
+							t.Errorf("unexpected errors: %s", diags)
+						}
+					} else {
+						if !diags.HasErrors() {
+							t.Fatalf("unexpected success\nwant error diagnostics with substring %q", test.WantErr)
+						}
+						gotErr := diags.Err().Error()
+						if !strings.Contains(gotErr, test.WantErr) {
+							t.Fatalf("missing expected error\ngot: %s\nwant error diagnostics with substring %q", gotErr, test.WantErr)
+						}
+					}
+				})
+			}
+		})
+	}
+}
+
+func TestContext2Plan_preventDestroy_dynamicEphemeral(t *testing.T) {
+	m := testModuleInline(t, map[string]string{
+		"main.tf": `
+			variable "prevent_destroy" {
+				type      = bool
+				default   = false
+				ephemeral = true
+			}
+
+			resource "test_instance" "foo" {
+				require_new = "yes"
+
+				lifecycle {
+					prevent_destroy = var.prevent_destroy
+				}
+			}
+		`,
+	})
+	p := testProvider("test")
+	p.PlanResourceChangeFn = testDiffFn
+	state := states.NewState()
+	root := state.EnsureModule(addrs.RootModuleInstance)
+	root.SetResourceInstanceCurrent(
+		mustResourceInstanceAddr("test_instance.foo").Resource,
+		&states.ResourceInstanceObjectSrc{
+			Status:    states.ObjectReady,
+			AttrsJSON: []byte(`{"id":"i-abc123"}`),
+		},
+		mustProviderConfig(`provider["registry.opentofu.org/hashicorp/test"]`),
+		addrs.NoKey,
+	)
+
+	ctx := testContext2(t, &ContextOpts{
+		Providers: map[addrs.Provider]providers.Factory{
+			addrs.NewDefaultProvider("test"): testProviderFuncFixed(p),
+		},
+	})
+
+	_, diags := ctx.Plan(context.Background(), m, state, SimplePlanOpts(plans.NormalMode, InputValues{
+		"prevent_destroy": {
+			Value:      cty.NilVal, // as if not set at all
+			SourceType: ValueFromCaller,
+		},
+	}))
+	if !diags.HasErrors() {
+		t.Fatalf("unexpected success\nwant error about ephemeral values")
+	}
+	gotErr := diags.Err().Error()
+	wantErr := `test_instance.foo has an ephemeral value for its prevent_destroy argument`
+	if !strings.Contains(gotErr, wantErr) {
+		t.Fatalf("missing expected error\ngot: %s\nwant error diagnostics with substring %q", gotErr, wantErr)
+	}
+}
+
+func TestContext2Plan_preventDestroy_dynamicDeprecated(t *testing.T) {
+	m := testModuleInline(t, map[string]string{
+		"main.tf": `
+			module "child" {
+				source = "./child"
+			}
+
+			resource "test_instance" "foo" {
+				require_new = "yes"
+
+				lifecycle {
+					prevent_destroy = module.child.prevent_destroy
+				}
+			}
+		`,
+		"child/main.tf": `
+			output "prevent_destroy" {
+				value      = false
+				deprecated = "Deprecated for testing purposes!"
+			}
+		`,
+	})
+	p := testProvider("test")
+	p.PlanResourceChangeFn = testDiffFn
+	state := states.NewState()
+	root := state.EnsureModule(addrs.RootModuleInstance)
+	root.SetResourceInstanceCurrent(
+		mustResourceInstanceAddr("test_instance.foo").Resource,
+		&states.ResourceInstanceObjectSrc{
+			Status:    states.ObjectReady,
+			AttrsJSON: []byte(`{"id":"i-abc123"}`),
+		},
+		mustProviderConfig(`provider["registry.opentofu.org/hashicorp/test"]`),
+		addrs.NoKey,
+	)
+
+	ctx := testContext2(t, &ContextOpts{
+		Providers: map[addrs.Provider]providers.Factory{
+			addrs.NewDefaultProvider("test"): testProviderFuncFixed(p),
+		},
+	})
+
+	_, diags := ctx.Plan(context.Background(), m, state, SimplePlanOpts(plans.NormalMode, nil))
+	assertNoErrors(t, diags)
+	gotErr := diags.ErrWithWarnings().Error()
+	wantErr := `This value is derived from module.child.prevent_destroy`
+	if !strings.Contains(gotErr, wantErr) {
+		t.Fatalf("missing expected warning\ngot: %s\nwant warning diagnostic with substring %q", gotErr, wantErr)
+	}
+}
+
+func TestContext2Plan_preventDestroy_dynamicFromDataResource(t *testing.T) {
+	// This test is intentionally a little redundant with
+	// [TestContext2Plan_preventDestroy_dynamic], but intentionally involves
+	// a dependency on another resource so that we're more likely to catch
+	// situations where references from the prevent_destroy argument are
+	// not detected when we're building the dependency graph, and therefore
+	// the prevent_destroy argument might be evaluated too early.
+	//
+	// If that _does_ happen then the most likely symptom is that this will
+	// return an error about prevent_destroy being unknown or null rather
+	// than about the resource having prevent_destroy set, because the
+	// result from the upstream resource would not have been written to
+	// the state yet. Exactly what would happen depends on the nature of
+	// the bug that caused the dependencies to be incorrect, but we've
+	// used synctest here to ensure that it should at least fail _reliably_
+	// when such a bug is present, rather than being a flake.
+	synctest.Test(t, func(t *testing.T) {
+		m := testModuleInline(t, map[string]string{
+			"main.tf": `
+			data "source" "foo" {
+			}
+
+			resource "test_instance" "foo" {
+				require_new = "yes"
+
+				lifecycle {
+					prevent_destroy = data.source.foo.prevent_destroy
+				}
+			}
+		`,
+		})
+		mainP := testProvider("test")
+		mainP.PlanResourceChangeFn = func(req providers.PlanResourceChangeRequest) providers.PlanResourceChangeResponse {
+			// Because we're simulating a "replace" situation here, the
+			// plan function actually gets called twice with the second
+			// one planning the "create" part of the replace. We want
+			// do do our fake synctest sleep only on the first plan call
+			// (where there's a prior state) because that's the one whose
+			// execution is supposed to wait until the data resource read
+			// has completed.
+			if !req.PriorState.IsNull() {
+				log.Printf("[TRACE] TestContext2Plan_preventDestroy_dynamicFromDataResource: fake sleep before test_instance.foo plan")
+				// The duration here doesn't really matter as long as it's
+				// shorter than the one in the other provider below. Refer to
+				// the comment there for more information.
+				time.Sleep(5 * time.Second)
+				log.Printf("[TRACE] TestContext2Plan_preventDestroy_dynamicFromDataResource: test_instance.foo plan")
+			} else {
+				log.Printf("[TRACE] TestContext2Plan_preventDestroy_dynamicFromDataResource: test_instance.foo followup plan")
+			}
+			return testDiffFn(req)
+		}
+		dataP := &MockProvider{
+			GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+				DataSources: map[string]providers.Schema{
+					"source": {
+						Block: &configschema.Block{
+							Attributes: map[string]*configschema.Attribute{
+								"prevent_destroy": {
+									Type:     cty.Bool,
+									Computed: true,
+								},
+							},
+						},
+					},
+				},
+			},
+			ReadDataSourceFn: func(_ providers.ReadDataSourceRequest) providers.ReadDataSourceResponse {
+				log.Printf("[TRACE] TestContext2Plan_preventDestroy_dynamicFromDataResource: fake sleep before data.source.foo read")
+				// The duration here doesn't really matter as long as it's
+				// longer than the one in the other provider above, meaning
+				// that this data read will definitely complete after the
+				// resource is planned unless the dependency between the
+				// two resources was correctly detected.
+				//
+				// Note that because we're in a synctest bubble this does not
+				// actually cause a "real" sleep, and instead just interacts
+				// with the synctest bubble's fake clock.
+				time.Sleep(10 * time.Second)
+				log.Printf("[TRACE] TestContext2Plan_preventDestroy_dynamicFromDataResource: data.source.foo read")
+				return providers.ReadDataSourceResponse{
+					State: cty.ObjectVal(map[string]cty.Value{
+						"prevent_destroy": cty.True,
+					}),
+				}
+			},
+		}
+		state := states.NewState()
+		root := state.EnsureModule(addrs.RootModuleInstance)
+		root.SetResourceInstanceCurrent(
+			mustResourceInstanceAddr("test_instance.foo").Resource,
+			&states.ResourceInstanceObjectSrc{
+				Status:    states.ObjectReady,
+				AttrsJSON: []byte(`{"id":"i-abc123"}`),
+			},
+			mustProviderConfig(`provider["registry.opentofu.org/hashicorp/test"]`),
+			addrs.NoKey,
+		)
+
+		ctx := testContext2(t, &ContextOpts{
+			Providers: map[addrs.Provider]providers.Factory{
+				addrs.NewDefaultProvider("test"):   testProviderFuncFixed(mainP),
+				addrs.NewDefaultProvider("source"): testProviderFuncFixed(dataP),
+			},
+		})
+
+		_, diags := ctx.Plan(context.Background(), m, state, SimplePlanOpts(plans.NormalMode, nil))
+		if !diags.HasErrors() {
+			t.Fatalf("unexpected success; want error about prevent_destroy being set")
+		}
+		gotErr := diags.Err().Error()
+		wantErr := `test_instance.foo has prevent_destroy set`
+		if !strings.Contains(gotErr, wantErr) {
+			t.Fatalf("missing expected error\ngot: %s\nwant error diagnostics with substring %q", gotErr, wantErr)
+		}
+	})
+}
+
 func TestContext2Plan_preventDestroy_countBad(t *testing.T) {
 	m := testModule(t, "plan-prevent-destroy-count-bad")
 	p := testProvider("aws")
@@ -1475,7 +1850,7 @@ func TestContext2Plan_preventDestroy_countBad(t *testing.T) {
 
 	plan, err := ctx.Plan(context.Background(), m, state, DefaultPlanOpts)
 
-	expectedErr := "aws_instance.foo[1] has lifecycle.prevent_destroy"
+	expectedErr := "aws_instance.foo[1] has prevent_destroy"
 	if !strings.Contains(fmt.Sprintf("%s", err), expectedErr) {
 		if plan != nil {
 			t.Logf("%s", legacyDiffComparisonString(plan.Changes))
@@ -1611,7 +1986,7 @@ func TestContext2Plan_preventDestroy_destroyPlan(t *testing.T) {
 		Mode: plans.DestroyMode,
 	})
 
-	expectedErr := "aws_instance.foo has lifecycle.prevent_destroy"
+	expectedErr := "aws_instance.foo has prevent_destroy"
 	if !strings.Contains(fmt.Sprintf("%s", diags.Err()), expectedErr) {
 		if plan != nil {
 			t.Logf("%s", legacyDiffComparisonString(plan.Changes))

internal/tofu/node_resource_abstract.go

@@ -206,6 +206,11 @@ func (n *NodeAbstractResource) References() []*addrs.Reference {
 		}
 
 		if c.Managed != nil {
+			if c.Managed.PreventDestroy != nil {
+				refs, _ := lang.ReferencesInExpr(addrs.ParseRef, c.Managed.PreventDestroy)
+				result = append(result, refs...)
+			}
+
 			if c.Managed.Connection != nil {
 				refs, _ = lang.ReferencesInBlock(addrs.ParseRef, c.Managed.Connection.Config, shared.ConnectionBlockSupersetSchema)
 				result = append(result, refs...)

internal/tofu/node_resource_abstract_instance.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/hashicorp/hcl/v2"
 	"github.com/zclconf/go-cty/cty"
+	"github.com/zclconf/go-cty/cty/convert"
 
 	"github.com/opentofu/opentofu/internal/addrs"
 	"github.com/opentofu/opentofu/internal/checks"
@@ -23,6 +24,8 @@ import (
 	"github.com/opentofu/opentofu/internal/configs/configschema"
 	"github.com/opentofu/opentofu/internal/encryption"
 	"github.com/opentofu/opentofu/internal/instances"
+	"github.com/opentofu/opentofu/internal/lang"
+	"github.com/opentofu/opentofu/internal/lang/evalchecks"
 	"github.com/opentofu/opentofu/internal/lang/marks"
 	"github.com/opentofu/opentofu/internal/plans"
 	"github.com/opentofu/opentofu/internal/plans/objchange"
@@ -398,28 +401,195 @@ func (n *NodeAbstractResourceInstance) readDiff(evalCtx EvalContext, providerSch
 	return change, nil
 }
 
-func (n *NodeAbstractResourceInstance) checkPreventDestroy(change *plans.ResourceInstanceChange) error {
-	if change == nil || n.Config == nil || n.Config.Managed == nil {
+func (n *NodeAbstractResourceInstance) checkPreventDestroy(ctx context.Context, evalCtx EvalContext, change *plans.ResourceInstanceChange) tfdiags.Diagnostics {
+	if change == nil || n.Config == nil || n.Config.Managed == nil || n.Config.Managed.PreventDestroy == nil {
 		return nil
 	}
 
-	preventDestroy := n.Config.Managed.PreventDestroy
+	var diags tfdiags.Diagnostics
 
-	if (change.Action == plans.Delete || change.Action.IsReplace()) && preventDestroy {
-		var diags tfdiags.Diagnostics
+	// NOTE: Some of the following would probably be similar if we later
+	// implement support for dynamic create_before_destroy too, but it's
+	// all written in a simpler, non-general way for now to keep it relatively
+	// simple until we actually know what subset of these rules is going to
+	// be common between the two.
+
+	preventDestroyExpr := n.Config.Managed.PreventDestroy
+	preventDestroyRefs, moreDiags := lang.ReferencesInExpr(addrs.ParseRef, preventDestroyExpr)
+	diags = diags.Append(moreDiags)
+	if moreDiags.HasErrors() {
+		return diags
+	}
+
+	// We have some special error messages for the instance-related symbols
+	// here, because it's reasonable for someone to try to use them to
+	// set prevent_destroy for only certain instances of a resource but we
+	// don't yet know how to support that.
+	for _, ref := range preventDestroyRefs {
+		switch addr := ref.Subject.(type) {
+		case addrs.ForEachAttr, addrs.CountAttr:
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Invalid reference in prevent_destroy",
+				Detail: fmt.Sprintf(
+					"A prevent_destroy argument cannot refer to %s, because OpenTofu needs to evaluate this argument for instances that have already been removed from the configuration and so whose per-instance data is no longer available.",
+					addr.String(),
+				),
+				Subject: ref.SourceRange.ToHCL().Ptr(),
+			})
+		}
+	}
+	if diags.HasErrors() {
+		// If we already have errors then we'll stop here because otherwise
+		// we'll redundantly re-report the invalid references during
+		// expression evaluation with lower-relevance error messages.
+		return diags
+	}
+
+	scope := evalCtx.EvaluationScope(nil, nil, EvalDataForNoInstanceKey)
+	hclCtx, moreDiags := scope.EvalContext(ctx, preventDestroyRefs)
+	diags = diags.Append(moreDiags)
+	if moreDiags.HasErrors() {
+		return diags
+	}
+	preventDestroyVal, hclDiags := preventDestroyExpr.Value(hclCtx)
+	diags = diags.Append(hclDiags)
+	if hclDiags.HasErrors() {
+		return diags
+	}
+
+	const errSummary = "Invalid value for prevent_destroy"
+	preventDestroyVal, err := convert.Convert(preventDestroyVal, cty.Bool)
+	if err != nil {
 		diags = diags.Append(&hcl.Diagnostic{
 			Severity: hcl.DiagError,
-			Summary:  "Instance cannot be destroyed",
+			Summary:  errSummary,
 			Detail: fmt.Sprintf(
-				"Resource %s has lifecycle.prevent_destroy set, but the plan calls for this resource to be destroyed. To avoid this error and continue with the plan, either disable lifecycle.prevent_destroy or reduce the scope of the plan using the -target flag.",
+				"Resource instance %s has an invalid value for its prevent_destroy argument: %s.",
+				n.Addr.String(), tfdiags.FormatError(err),
+			),
+			Subject:     preventDestroyExpr.Range().Ptr(),
+			Expression:  preventDestroyExpr,
+			EvalContext: hclCtx,
+		})
+	}
+	preventDestroyVal, moreDiags = marks.ExtractDeprecatedDiagnosticsWithExpr(preventDestroyVal, preventDestroyExpr)
+	diags = diags.Append(moreDiags)
+	preventDestroyVal, pdMarks := preventDestroyVal.Unmark()
+	for mark := range pdMarks {
+		switch mark {
+		case marks.Sensitive:
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  errSummary,
+				Detail: fmt.Sprintf(
+					"Resource instance %s has a sensitive value for its prevent_destroy argument, which is invalid because it would cause OpenTofu to disclose the sensitive value by whether deletion is blocked.\n\nIf you know this value is not sensitive in practice, consider using the nonsensitive function to declare that.",
+					n.Addr.String(),
+				),
+				Subject:     preventDestroyExpr.Range().Ptr(),
+				Expression:  preventDestroyExpr,
+				EvalContext: hclCtx,
+				Extra:       evalchecks.DiagnosticCausedByConfidentialValues(true),
+			})
+		case marks.Ephemeral:
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  errSummary,
+				Detail: fmt.Sprintf(
+					"Resource instance %s has an ephemeral value for its prevent_destroy argument, which is invalid because the decision for whether it's okay to destroy instances of this resource instance must stay consistent between plan and apply.",
+					n.Addr.String(),
+				),
+				Subject:     preventDestroyExpr.Range().Ptr(),
+				Expression:  preventDestroyExpr,
+				EvalContext: hclCtx,
+			})
+		default:
+			// This is a generic error message just to make sure that we'll
+			// fail if a new kind of mark gets added in future which we've
+			// not yet considered whether to allow here. If we add a new mark
+			// kind then we should add a new case for it above, even if the
+			// behavior is to do absolutely nothing because that mark is
+			// allowed in prevent_destroy.
+			diags = diags.Append(&hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  errSummary,
+				Detail: fmt.Sprintf(
+					"Resource instance %s has a prevent_destroy value derived from something that isn't allowed for deciding whether a resource instance may be destroyed (has internal mark %#v). The fact that OpenTofu cannot give more details about this is a bug, so please report it!",
+					n.Addr.String(), mark,
+				),
+				Subject:     preventDestroyExpr.Range().Ptr(),
+				Expression:  preventDestroyExpr,
+				EvalContext: hclCtx,
+			})
+		}
+	}
+
+	if diags.HasErrors() {
+		// If we already have errors then we'll stop early here.
+		return diags
+	}
+	if change.Action != plans.Delete && !change.Action.IsReplace() {
+		// If we're not attempting to destroy then the above checks are
+		// sufficient to reject an expression that cannot possibly be valid
+		// for prevent_destroy. If we're not actually planning to destroy
+		// then we'll skip the remaining checks because they are likely to
+		// fail dynamically in non-destroy situations even though they
+		// could be valid by the time this object actually is planned for
+		// destroy.
+		return nil
+	}
+
+	if !preventDestroyVal.IsKnown() {
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  errSummary,
+			Detail: fmt.Sprintf(
+				"Resource instance %s has a prevent_destroy argument but its value will not be known until the apply step, so OpenTofu can't predict whether destroying this is acceptable.\n\nTo proceed, exclude instances of this resource from this round using:\n    -exclude=%q",
+				n.Addr.String(), n.Addr.ContainingResource().String(),
+			),
+			Subject:     preventDestroyExpr.Range().Ptr(),
+			Expression:  preventDestroyExpr,
+			EvalContext: hclCtx,
+			Extra:       evalchecks.DiagnosticCausedByUnknown(true),
+		})
+	}
+	if preventDestroyVal.IsNull() {
+		// We could potentially treat null as equivalent to false here, matching
+		// how OpenTofu would behave if there were no expression present at all,
+		// but "false" is just as easy to specify as "null" in a conditional
+		// expression and doesn't require a reader to know what the default
+		// is, so we'll require that to make life easier for a future maintainer
+		// that isn't necessarily familiar with the prevent_destroy behavior yet.
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  errSummary,
+			Detail: fmt.Sprintf(
+				"Resource instance %s has prevent_destroy set to null. When making a dynamic decision to allow destroy, use false instead.",
 				n.Addr.String(),
 			),
+			Subject:     preventDestroyExpr.Range().Ptr(),
+			Expression:  preventDestroyExpr,
+			EvalContext: hclCtx,
+		})
+	}
+	if diags.HasErrors() {
+		// Any errors so far means that preventDestroyVal.True is likely to
+		// either panic or return nonsense.
+		return diags
+	}
+
+	if preventDestroyVal.True() {
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Resource instance cannot be destroyed",
+			Detail: fmt.Sprintf(
+				"Resource instance %s has prevent_destroy set, but the plan calls for it to be destroyed.\n\nTo proceed, either disable prevent_destroy for this resource or exclude instances of this resource from this round using:\n    -exclude=%q",
+				n.Addr.String(), n.Addr.ContainingResource().String(),
+			),
 			Subject: &n.Config.DeclRange,
 		})
-		return diags.Err()
 	}
-
-	return nil
+	return diags
 }
 
 // preApplyHook calls the pre-Apply hook

internal/tofu/node_resource_plan_destroy.go

@@ -143,7 +143,7 @@ func (n *NodePlanDestroyableResourceInstance) managedResourceExecute(ctx context
 		return diags
 	}
 
-	diags = diags.Append(n.checkPreventDestroy(change))
+	diags = diags.Append(n.checkPreventDestroy(ctx, evalCtx, change))
 	return diags
 }
 

internal/tofu/node_resource_plan_instance.go

@@ -433,7 +433,7 @@ func (n *NodePlannableResourceInstance) managedResourceExecute(ctx context.Conte
 		if diags.HasErrors() {
 			return diags
 		}
-		diags = diags.Append(n.checkPreventDestroy(change))
+		diags = diags.Append(n.checkPreventDestroy(ctx, evalCtx, change))
 		if diags.HasErrors() {
 			return diags
 		}

internal/tofu/node_resource_plan_orphan.go

@@ -227,7 +227,7 @@ func (n *NodePlannableResourceInstanceOrphan) managedResourceExecute(ctx context
 		return diags
 	}
 
-	diags = diags.Append(n.checkPreventDestroy(change))
+	diags = diags.Append(n.checkPreventDestroy(ctx, evalCtx, change))
 	if diags.HasErrors() {
 		return diags
 	}

internal/tofu/node_resource_validate.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/hashicorp/hcl/v2"
 	"github.com/zclconf/go-cty/cty"
+	"github.com/zclconf/go-cty/cty/convert"
 
 	"github.com/opentofu/opentofu/internal/addrs"
 	"github.com/opentofu/opentofu/internal/communicator/shared"
@@ -68,6 +69,31 @@ func (n *NodeValidatableResource) Execute(ctx context.Context, evalCtx EvalConte
 	diags = diags.Append(n.validateCheckRules(ctx, evalCtx, n.Config))
 
 	if managed := n.Config.Managed; managed != nil {
+		if pdExpr := managed.PreventDestroy; pdExpr != nil {
+			// This validation focuses only on the simple case of a valid
+			// constant expression, because it's replacing some static
+			// type-checking that was previously done during config loading,
+			// before we allowed dynamic expressions here. If the expression
+			// refers to anything else in the configuration, or if it fails
+			// evaluation for any other reason, then we'll wait until the plan
+			// phase to check it properly so we can have more information
+			// available to generate better error messages.
+			if val, hclDiags := pdExpr.Value(nil); !hclDiags.HasErrors() {
+				_, err := convert.Convert(val, cty.Bool)
+				if err != nil {
+					diags = diags.Append(&hcl.Diagnostic{
+						Severity: hcl.DiagError,
+						Summary:  "Invalid value for prevent_destroy",
+						Detail: fmt.Sprintf(
+							"Resource instance %s has an invalid value for its prevent_destroy argument: %s.",
+							n.Addr.String(), tfdiags.FormatError(err),
+						),
+						Subject: pdExpr.Range().Ptr(),
+					})
+				}
+			}
+		}
+
 		// Validate all the provisioners
 		for _, p := range managed.Provisioners {
 			// Create a local shallow copy of the provisioner