diff --git a/internal/stacks/stackplan/component.go b/internal/stacks/stackplan/component.go
index c87424eb4a..f119b5b130 100644
--- a/internal/stacks/stackplan/component.go
+++ b/internal/stacks/stackplan/component.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/hashicorp/terraform/internal/addrs"
 	"github.com/hashicorp/terraform/internal/plans"
+	"github.com/hashicorp/terraform/internal/states"
 )
 
 // Component is a container for a set of changes that all belong to the same
@@ -21,6 +22,10 @@ type Component struct {
 	// ResourceInstancePlanned describes the changes that Terraform is proposing
 	// to make to try to converge the real system state with the desired state
 	// as described by the configuration.
+	//
+	// FIXME: This modelling is incorrect, because it doesn't handle the fact that
+	// a resource instance change might actually be for a deposed object
+	// rather than the current object.
 	ResourceInstancePlanned addrs.Map[addrs.AbsResourceInstance, *plans.ResourceInstanceChangeSrc]
 
 	// TODO: Something for deferred resource instance changes, once we have
@@ -31,3 +36,44 @@ type Component struct {
 	// function during apply and must not be used for any other purpose.
 	PlanTimestamp time.Time
 }
+
+// ForModulesRuntime translates the component instance plan into the form
+// expected by the modules runtime, which is what would ultimately be used
+// to apply the plan.
+//
+// The stack component planning model preserves only the most crucial details
+// of a component plan produced by the modules runtime, and so the result
+// will not exactly match the [plans.Plan] that the component plan was produced
+// from, but should be complete enough to successfully apply the plan.
+//
+// Conversion with this method should always succeed if the given previous
+// run state is truly the one that the plan was created from. If this method
+// returns an error then that suggests that the recieving plan is inconsistent
+// with the given previous run state, which should not happen if the caller
+// is using Terraform Core correctly.
+func (c *Component) ForModulesRuntime(prevRunState *states.State) (*plans.Plan, error) {
+	changes := plans.NewChanges()
+	priorState := prevRunState.DeepCopy()
+	plan := &plans.Plan{
+		Changes:      changes,
+		Timestamp:    c.PlanTimestamp,
+		PrevRunState: prevRunState,
+		PriorState:   priorState,
+	}
+
+	sc := changes.SyncWrapper()
+	for _, elem := range c.ResourceInstancePlanned.Elems {
+		changeSrc := elem.Value
+		sc.AppendResourceInstanceChange(changeSrc)
+	}
+
+	// FIXME: For ResourceInstanceChangedOutside we actually need to modify
+	// priorState, since that will mimick what the modules runtime would've
+	// done itself during its own refresh and plan process during the
+	// planning phase. But we can't do that here because we'd need providers
+	// to help us convert from msgpack to JSON. So we'll just ignore the
+	// "changed outside" stuff for now and figure out what to do with this
+	// problem later.
+
+	return plan, nil
+}
diff --git a/internal/stacks/stackruntime/internal/stackeval/applying.go b/internal/stacks/stackruntime/internal/stackeval/applying.go
index 22adacb60f..bd5744d47a 100644
--- a/internal/stacks/stackruntime/internal/stackeval/applying.go
+++ b/internal/stacks/stackruntime/internal/stackeval/applying.go
@@ -1,5 +1,36 @@
 package stackeval
 
+import (
+	"context"
+
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
+	"github.com/hashicorp/terraform/internal/tfdiags"
+)
+
 type ApplyOpts struct {
 	ProviderFactories ProviderFactories
 }
+
+// ApplyChecker is an interface implemented by types which represent objects
+// that can potentially produce diagnostics and object change reports during
+// the apply phase.
+//
+// Unlike [Plannable], ApplyChecker implementations do not actually apply
+// changes themselves. Instead, the real changes get driven separately using
+// the [ChangeExec] function (see [ApplyPlan]) and then we collect up any
+// reports to send to the caller separately using this interface.
+type ApplyChecker interface {
+	// CheckApply checks the receiver's apply-time result and returns zero
+	// or more applied change descriptions and zero or more diagnostics
+	// describing any problems that occured for this specific object during
+	// the apply phase.
+	//
+	// CheckApply must not report any diagnostics raised indirectly by
+	// evaluating other objects. Those will be collected separately by calling
+	// this same method on those other objects.
+	CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics)
+
+	// Our general async planning helper relies on this to name its
+	// tracing span.
+	tracingNamer
+}
diff --git a/internal/stacks/stackruntime/internal/stackeval/change_exec.go b/internal/stacks/stackruntime/internal/stackeval/change_exec.go
index 6702571026..4c77b2e95c 100644
--- a/internal/stacks/stackruntime/internal/stackeval/change_exec.go
+++ b/internal/stacks/stackruntime/internal/stackeval/change_exec.go
@@ -166,6 +166,16 @@ func (r *ChangeExecResults) ComponentInstanceResult(ctx context.Context, addr st
 	return valWithDiags.Result, valWithDiags.Diagnostics, err
 }
 
+// AwaitCompletion blocks until all of the scheduled changes have completed.
+func (r *ChangeExecResults) AwaitCompletion(ctx context.Context) {
+	// We don't have any single signal that everything is complete here,
+	// but it's sufficient for us to just visit each of our saved promise
+	// getters in turn and read from them.
+	for _, elem := range r.componentInstances.Elems() {
+		elem.V(ctx) // intentionally discards result; we only care that it's complete
+	}
+}
+
 type ErrChangeExecUnregistered struct {
 	Addr fmt.Stringer
 }
diff --git a/internal/stacks/stackruntime/internal/stackeval/component.go b/internal/stacks/stackruntime/internal/stackeval/component.go
index a8ba10bbaa..e078ad42a6 100644
--- a/internal/stacks/stackruntime/internal/stackeval/component.go
+++ b/internal/stacks/stackruntime/internal/stackeval/component.go
@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/terraform/internal/stacks/stackaddrs"
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 )
@@ -285,6 +286,17 @@ func (c *Component) ExprReferenceValue(ctx context.Context, phase EvalPhase) cty
 	return c.ResultValue(ctx, phase)
 }
 
+func (c *Component) checkValid(ctx context.Context, phase EvalPhase) tfdiags.Diagnostics {
+	var diags tfdiags.Diagnostics
+
+	_, moreDiags := c.CheckForEachValue(ctx, phase)
+	diags = diags.Append(moreDiags)
+	_, moreDiags = c.CheckInstances(ctx, phase)
+	diags = diags.Append(moreDiags)
+
+	return diags
+}
+
 // PlanChanges implements Plannable by performing plan-time validation of
 // the component call itself.
 //
@@ -292,14 +304,12 @@ func (c *Component) ExprReferenceValue(ctx context.Context, phase EvalPhase) cty
 // PlanChanges for each instance separately in order to produce a complete
 // plan.
 func (c *Component) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
-	var diags tfdiags.Diagnostics
-
-	_, moreDiags := c.CheckForEachValue(ctx, PlanPhase)
-	diags = diags.Append(moreDiags)
-	_, moreDiags = c.CheckInstances(ctx, PlanPhase)
-	diags = diags.Append(moreDiags)
+	return nil, c.checkValid(ctx, PlanPhase)
+}
 
-	return nil, diags
+// CheckApply implements ApplyChecker.
+func (c *Component) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	return nil, c.checkValid(ctx, ApplyPhase)
 }
 
 func (c *Component) tracingName() string {
diff --git a/internal/stacks/stackruntime/internal/stackeval/component_instance.go b/internal/stacks/stackruntime/internal/stackeval/component_instance.go
index 6ea367c00e..2859730c23 100644
--- a/internal/stacks/stackruntime/internal/stackeval/component_instance.go
+++ b/internal/stacks/stackruntime/internal/stackeval/component_instance.go
@@ -17,6 +17,8 @@ import (
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig/stackconfigtypes"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
 	"github.com/hashicorp/terraform/internal/stacks/stackruntime/hooks"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
+	"github.com/hashicorp/terraform/internal/states"
 	"github.com/hashicorp/terraform/internal/terraform"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 )
@@ -527,14 +529,54 @@ func (c *ComponentInstance) CheckModuleTreePlan(ctx context.Context) (*plans.Pla
 	)
 }
 
+// ApplyModuleTreePlan applies a plan returned by a previous call to
+// [ComponentInstance.CheckModuleTreePlan].
+//
+// Applying a plan often has significant externally-visible side-effects, and
+// so this method should be called only once for a given plan. In practice
+// we currently ensure that is true by calling it only from the package-level
+// [ApplyPlan] function, which arranges for this function to be called
+// concurrently with the same method on other component instances and with
+// a whole-tree walk to gather up results and diagnostics.
+func (c *ComponentInstance) ApplyModuleTreePlan(ctx context.Context, plan *plans.Plan) (*states.State, tfdiags.Diagnostics) {
+	panic("unimplemented")
+}
+
+// ApplyResultState returns the new state resulting from applying a plan for
+// this object using [ApplyModuleTreePlan], or nil if the apply failed and
+// so there is no new state to return.
+func (c *ComponentInstance) ApplyResultState(ctx context.Context) *states.State {
+	ret, _ := c.CheckApplyResultState(ctx)
+	return ret
+}
+
+// CheckApplyResultState returns the new state resulting from applying a plan for
+// this object using [ApplyModuleTreePlan] and diagnostics describing any
+// problems encountered when applying it.
+func (c *ComponentInstance) CheckApplyResultState(ctx context.Context) (*states.State, tfdiags.Diagnostics) {
+	var diags tfdiags.Diagnostics
+	changes := c.main.ApplyChangeResults()
+	newState, moreDiags, err := changes.ComponentInstanceResult(ctx, c.Addr())
+	diags = diags.Append(moreDiags)
+	if err != nil {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			"Component instance apply not scheduled",
+			fmt.Sprintf("Terraform needs the result from applying changes to %s, but that apply was apparently not scheduled to run. This is a bug in Terraform.", c.Addr()),
+		))
+	}
+	return newState, diags
+}
+
 func (c *ComponentInstance) ResultValue(ctx context.Context, phase EvalPhase) cty.Value {
 	switch phase {
 	case PlanPhase:
 		plan := c.ModuleTreePlan(ctx)
 		if plan == nil {
 			// Planning seems to have failed so we cannot decide a result value yet.
-			// FIXME: Should use an unknown value of a specific object type
-			// constraint derived from the root module's output values.
+			// We can't do any better than DynamicVal here because in the
+			// modules language output values don't have statically-declared
+			// result types.
 			return cty.DynamicVal
 		}
 
@@ -552,10 +594,30 @@ func (c *ComponentInstance) ResultValue(ctx context.Context, phase EvalPhase) ct
 		}
 		return cty.ObjectVal(attrs)
 
+	case ApplyPhase:
+		newState := c.ApplyResultState(ctx)
+		if newState == nil {
+			// Applying seems to have failed so we cannot provide a result
+			// value, and so we'll return a placeholder to help our caller
+			// unwind gracefully with its own placeholder result.
+			// We can't do any better than DynamicVal here because in the
+			// modules language output values don't have statically-declared
+			// result types.
+			return cty.DynamicVal
+		}
+
+		// During the apply phase we use the root module output values from
+		// the new state to construct our value.
+		outputVals := newState.RootModule().OutputValues
+		attrs := make(map[string]cty.Value, len(outputVals))
+		for _, ov := range outputVals {
+			name := ov.Addr.OutputValue.Name
+			attrs[name] = ov.Value
+		}
+		return cty.ObjectVal(attrs)
+
 	default:
 		// We can't produce a concrete value for any other phase.
-		// FIXME: Should use an unknown value of a specific object type
-		// constraint derived from the root module's output values.
 		return cty.DynamicVal
 	}
 }
@@ -612,6 +674,27 @@ func (c *ComponentInstance) PlanChanges(ctx context.Context) ([]stackplan.Planne
 	return changes, diags
 }
 
+// CheckApply implements ApplyChecker.
+func (c *ComponentInstance) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	var changes []stackstate.AppliedChange
+	var diags tfdiags.Diagnostics
+
+	// FIXME: We need to report AppliedChange objects for the component
+	// instance itself and each of the affected resource instances inside it.
+	// For now we're only reporting diagnostics as an initial stub.
+
+	_, moreDiags := c.CheckInputVariableValues(ctx, ApplyPhase)
+	diags = diags.Append(moreDiags)
+
+	_, moreDiags = c.CheckProviders(ctx, ApplyPhase)
+	diags = diags.Append(moreDiags)
+
+	_, moreDiags = c.CheckApplyResultState(ctx)
+	diags = diags.Append(moreDiags)
+
+	return changes, diags
+}
+
 func (c *ComponentInstance) tracingName() string {
 	return c.Addr().String()
 }
diff --git a/internal/stacks/stackruntime/internal/stackeval/input_variable.go b/internal/stacks/stackruntime/internal/stackeval/input_variable.go
index 69ff7d1449..37cd87763b 100644
--- a/internal/stacks/stackruntime/internal/stackeval/input_variable.go
+++ b/internal/stacks/stackruntime/internal/stackeval/input_variable.go
@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/terraform/internal/stacks/stackaddrs"
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 	"github.com/zclconf/go-cty/cty/convert"
@@ -152,15 +153,24 @@ func (v *InputVariable) ExprReferenceValue(ctx context.Context, phase EvalPhase)
 	return v.Value(ctx, phase)
 }
 
-// PlanChanges implements Plannable as a plan-time validation of the variable's
-// declaration and of the caller's definition of the variable.
-func (v *InputVariable) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+func (v *InputVariable) checkValid(ctx context.Context, phase EvalPhase) tfdiags.Diagnostics {
 	var diags tfdiags.Diagnostics
 
 	_, moreDiags := v.CheckValue(ctx, PlanPhase)
 	diags = diags.Append(moreDiags)
 
-	return nil, diags
+	return diags
+}
+
+// PlanChanges implements Plannable as a plan-time validation of the variable's
+// declaration and of the caller's definition of the variable.
+func (v *InputVariable) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+	return nil, v.checkValid(ctx, PlanPhase)
+}
+
+// CheckApply implements ApplyChecker.
+func (v *InputVariable) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	return nil, v.checkValid(ctx, ApplyPhase)
 }
 
 func (v *InputVariable) tracingName() string {
diff --git a/internal/stacks/stackruntime/internal/stackeval/main.go b/internal/stacks/stackruntime/internal/stackeval/main.go
index 17cf724424..0fd018897a 100644
--- a/internal/stacks/stackruntime/internal/stackeval/main.go
+++ b/internal/stacks/stackruntime/internal/stackeval/main.go
@@ -64,7 +64,8 @@ type mainPlanning struct {
 }
 
 type mainApplying struct {
-	opts ApplyOpts
+	opts    ApplyOpts
+	results *ChangeExecResults
 }
 
 func NewForValidating(config *stackconfig.Config, opts ValidateOpts) *Main {
@@ -92,6 +93,17 @@ func NewForPlanning(config *stackconfig.Config, opts PlanOpts) *Main {
 	}
 }
 
+func NewForApplying(config *stackconfig.Config, execResults *ChangeExecResults, opts ApplyOpts) *Main {
+	return &Main{
+		config: config,
+		applying: &mainApplying{
+			opts: opts,
+		},
+		providerFactories: opts.ProviderFactories,
+		providerTypes:     make(map[addrs.Provider]*ProviderType),
+	}
+}
+
 // Validating returns true if the receiving [Main] is configured for validating.
 //
 // If this returns false then validation methods may panic or return strange
@@ -113,10 +125,10 @@ func (m *Main) Planning() bool {
 // If this returns false then applying methods may panic or return strange
 // results.
 func (m *Main) Applying() bool {
-	return m.applying != nil && m.Planning()
+	return m.applying != nil
 }
 
-// PlanningMode returns the planning options to use during the planning phase,
+// PlanningOpts returns the planning options to use during the planning phase,
 // or panics if this [Main] was not instantiated for planning.
 //
 // Do not modify anything reachable through the returned pointer.
@@ -127,6 +139,16 @@ func (m *Main) PlanningOpts() *PlanOpts {
 	return &m.planning.opts
 }
 
+// ApplyChangeResults returns the object that tracks the results of the actual
+// changes being made during the apply phase, or panics if this [Main] is not
+// instantiated for applying.
+func (m *Main) ApplyChangeResults() *ChangeExecResults {
+	if !m.Applying() {
+		panic("stacks language runtime is not instantiated for applying")
+	}
+	return m.applying.results
+}
+
 // SourceBundle returns the source code bundle that the stack configuration
 // was originally loaded from and that should also contain the source code
 // for any modules that "component" blocks refer to.
diff --git a/internal/stacks/stackruntime/internal/stackeval/main_apply.go b/internal/stacks/stackruntime/internal/stackeval/main_apply.go
new file mode 100644
index 0000000000..8053dcc16e
--- /dev/null
+++ b/internal/stacks/stackruntime/internal/stackeval/main_apply.go
@@ -0,0 +1,273 @@
+package stackeval
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+
+	"github.com/hashicorp/terraform/internal/promising"
+	"github.com/hashicorp/terraform/internal/stacks/stackconfig"
+	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
+	"github.com/hashicorp/terraform/internal/states"
+	"github.com/hashicorp/terraform/internal/tfdiags"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+// ApplyPlan internally instantiates a [Main] configured to apply the given
+// raw plan, and then visits all of the relevant objects to collect up any
+// diagnostics they emit while evaluating in terms of the change results.
+func ApplyPlan(ctx context.Context, config *stackconfig.Config, rawPlan []*anypb.Any, opts ApplyOpts, outp ApplyOutput) error {
+	plan, err := stackplan.LoadFromProto(rawPlan)
+	if err != nil {
+		return fmt.Errorf("invalid raw plan: %w", err)
+	}
+	if !plan.Applyable {
+		// We should not get here because a caller should not ask us to try
+		// to apply a plan that wasn't marked as applyable, but we'll check
+		// it anyway just to be robust in case there's a bug further up
+		// the call stack.
+		return fmt.Errorf("plan is not applyable")
+	}
+
+	// We'll register all of the changes we intend to make up front, so we
+	// can error rather than deadlock if something goes wrong and causes
+	// us to try to depend on a result that isn't coming.
+	results, begin := ChangeExec(ctx, func(ctx context.Context, reg *ChangeExecRegistry[*Main]) {
+		for _, elem := range plan.Components.Elems() {
+			addr := elem.K
+			componentInstPlan := elem.V
+			reg.RegisterComponentInstanceChange(
+				ctx, addr,
+				func(ctx context.Context, main *Main) (*states.State, tfdiags.Diagnostics) {
+					stack := main.Stack(ctx, addr.Stack, ApplyPhase)
+					component := stack.Component(ctx, addr.Item.Component)
+					insts := component.Instances(ctx, ApplyPhase)
+					inst, ok := insts[addr.Item.Key]
+					if !ok {
+						// If we managed to plan a change for this instance
+						// during the plan phase but yet it doesn't exist
+						// during the apply phase then that suggests that
+						// something upstream has failed in a strange way
+						// during apply and so this component's for_each or
+						// count argument can't be properly evaluated anymore.
+						// This is an unlikely case but we'll tolerate it by
+						// returning a placeholder value and expect the cause
+						// to be reported by some object when we do the apply
+						// checking walk below.
+						return nil, nil
+					}
+
+					// TODO: We should also turn the prior state into the form
+					// the modules runtime expects and pass that in here,
+					// instead of an empty prior state.
+					modulesRuntimePlan, err := componentInstPlan.ForModulesRuntime(states.NewState())
+					if err != nil {
+						// Suggests that the state is inconsistent with the
+						// plan, which is a bug in whatever provided us with
+						// those two artifacts, but we don't know who that
+						// caller is (it probably came from a client of the
+						// Core RPC API) so we don't include our typical
+						// "This is a bug in Terraform" language here.
+						var diags tfdiags.Diagnostics
+						diags = diags.Append(tfdiags.Sourceless(
+							tfdiags.Error,
+							"Inconsistent component instance plan",
+							fmt.Sprintf("The plan for %s is inconsistent with its prior state: %s.", addr, err),
+						))
+						return nil, diags
+					}
+
+					return inst.ApplyModuleTreePlan(ctx, modulesRuntimePlan)
+				},
+			)
+		}
+	})
+
+	main := NewForApplying(config, results, opts)
+	begin(ctx, main) // the change tasks registered above become runnable
+
+	// With the planned changes now in progress, we'll visit everything and
+	// each object to check itself (producing diagnostics) and announce any
+	// changes that were applied to it.
+	diags, err := promising.MainTask(ctx, func(ctx context.Context) (tfdiags.Diagnostics, error) {
+		var seenSelfDepDiag atomic.Bool
+		ws, complete := newWalkStateCustomDiags(
+			func(diags tfdiags.Diagnostics) {
+				for _, diag := range diags {
+					if diagIsPromiseSelfReference(diag) {
+						// We'll discard all but the first promise-self-reference
+						// diagnostic we see; these tend to get duplicated
+						// because they emerge from all codepaths participating
+						// in the self-reference at once.
+						if !seenSelfDepDiag.CompareAndSwap(false, true) {
+							continue
+						}
+					}
+					outp.AnnounceDiagnostics(ctx, tfdiags.Diagnostics{diag})
+				}
+			},
+			func() tfdiags.Diagnostics {
+				// We emit all diagnostics immediately as they arrive, so
+				// we never have any accumulated diagnostics to emit at the end.
+				return nil
+			},
+		)
+		walk := &applyWalk{
+			state: ws,
+			out:   &outp,
+		}
+
+		// walkCheckAppliedChanges, and all of the downstream functions it calls,
+		// must take care to ensure that there's always at least one
+		// planWalk-tracked async task running until the entire process is
+		// complete. If one task launches another then the child task call
+		// must come before the caller's implementation function returns.
+		main.walkCheckAppliedChanges(ctx, walk, main.MainStack(ctx))
+
+		// Note: in practice this "complete" cannot actually return any
+		// diagnostics because our custom walkstate hooks above just announce
+		// the diagnostics immediately. But "complete" still serves the purpose
+		// of blocking until all of the async jobs are complete.
+		diags := complete()
+
+		// By the time we get here all of the scheduled changes should be
+		// complete already anyway, since we should have visited them all
+		// in walkCheckAppliedChanges, but just to make sure we don't leave
+		// anything hanging in the background if walkCheckAppliedChanges is
+		// buggy we'll also pause here until the ChangeExec scheduler thinks
+		// everything it's supervising is complete.
+		results.AwaitCompletion(ctx)
+
+		return diags, nil
+	})
+	diags = diags.Append(diagnosticsForPromisingTaskError(err, main))
+	if len(diags) > 0 {
+		outp.AnnounceDiagnostics(ctx, diags)
+	}
+
+	return nil
+}
+
+type ApplyOutput struct {
+	// Called each time we confirm that a planned change has now been applied.
+	//
+	// Each announced change can have a raw element, an external-facing
+	// element, or both. The raw element is opaque to anything outside of
+	// Terraform Core, while the external-facing element is never consumed
+	// by Terraform Core and is instead for other uses such as presenting
+	// changes in the UI.
+	//
+	// The callback should return relatively quickly to minimize the
+	// backpressure applied to the planning process.
+	AnnounceAppliedChange func(context.Context, stackstate.AppliedChange)
+
+	// Called each time we encounter some diagnostics. These are asynchronous
+	// from planned changes because the evaluator will sometimes need to
+	// aggregate together some diagnostics and post-process the set before
+	// announcing them. Callers should not try to correlate diagnostics
+	// with planned changes by announcement-time-proximity.
+	//
+	// The callback should return relatively quickly to minimize the
+	// backpressure applied to the planning process.
+	AnnounceDiagnostics func(context.Context, tfdiags.Diagnostics)
+}
+
+// applyWalk just bundles a [walkState] and an [ApplyOutput] together so we can
+// concisely pass them both as a single argument between the all the apply walk
+// driver functions below.
+type applyWalk = walkWithOutput[*ApplyOutput]
+
+func (m *Main) walkCheckAppliedChanges(ctx context.Context, walk *applyWalk, stack *Stack) {
+	// We'll get the expansion of any child stack calls going first, so that
+	// we can explore downstream stacks concurrently with this one. Each
+	// stack call can represent zero or more child stacks that we'll analyze
+	// by recursive calls to this function.
+	for _, call := range stack.EmbeddedStackCalls(ctx) {
+		call := call // separate symbol per loop iteration
+
+		m.walkApplyCheckObjectChanges(ctx, walk, call)
+
+		// We need to perform the whole expansion in an overall async task
+		// because it involves evaluating for_each expressions, and one
+		// stack call's for_each might depend on the results of another.
+		walk.AsyncTask(ctx, func(ctx context.Context) {
+			insts := call.Instances(ctx, PlanPhase)
+			for _, inst := range insts {
+				m.walkApplyCheckObjectChanges(ctx, walk, inst)
+
+				childStack := inst.CalledStack(ctx)
+				m.walkCheckAppliedChanges(ctx, walk, childStack)
+			}
+		})
+	}
+
+	// We also need to visit and check all of the other declarations in
+	// the current stack.
+
+	for _, component := range stack.Components(ctx) {
+		component := component // separate symbol per loop iteration
+
+		m.walkApplyCheckObjectChanges(ctx, walk, component)
+
+		// We need to perform the instance expansion in an overall async task
+		// because it involves potentially evaluating a for_each expression.
+		// and that might depend on data from elsewhere in the same stack.
+		walk.AsyncTask(ctx, func(ctx context.Context) {
+			insts := component.Instances(ctx, PlanPhase)
+			for _, inst := range insts {
+				// This is the means by which we learn of any diagnostics from
+				// applying the component's plan and report that we've applied
+				// the changes; this indirectly consumes the results from
+				// the change actions scheduled earlier in [ApplyPlan].
+				m.walkApplyCheckObjectChanges(ctx, walk, inst)
+			}
+		})
+	}
+	for _, provider := range stack.Providers(ctx) {
+		provider := provider // separate symbol per loop iteration
+
+		m.walkApplyCheckObjectChanges(ctx, walk, provider)
+
+		// We need to perform the instance expansion in an overall async
+		// task because it involves potentially evaluating a for_each expression,
+		// and that might depend on data from elsewhere in the same stack.
+		walk.AsyncTask(ctx, func(ctx context.Context) {
+			insts := provider.Instances(ctx, PlanPhase)
+			for _, inst := range insts {
+				m.walkApplyCheckObjectChanges(ctx, walk, inst)
+			}
+		})
+	}
+	for _, variable := range stack.InputVariables(ctx) {
+		m.walkApplyCheckObjectChanges(ctx, walk, variable)
+	}
+	// TODO: Local values
+	for _, output := range stack.OutputValues(ctx) {
+		m.walkApplyCheckObjectChanges(ctx, walk, output)
+	}
+
+	// Finally we'll also check the stack itself, to deal with any problems
+	// with the stack as a whole rather than individual declarations inside.
+	m.walkApplyCheckObjectChanges(ctx, walk, stack)
+}
+
+// walkApplyCheckObjectChanges deals with the leaf objects that can directly
+// contribute changes and/or diagnostics to the apply result, which should each
+// implement [ApplyChecker].
+//
+// This function is not responsible for actually making the changes; they must
+// be scheduled separately or this function will either block forever or
+// return strange errors. (See [ApplyPlan] for more about how the apply phase
+// deals with changes.)
+func (m *Main) walkApplyCheckObjectChanges(ctx context.Context, walk *applyWalk, obj ApplyChecker) {
+	walk.AsyncTask(ctx, func(ctx context.Context) {
+		changes, diags := obj.CheckApply(ctx)
+		for _, change := range changes {
+			walk.out.AnnounceAppliedChange(ctx, change)
+		}
+		if len(diags) != 0 {
+			walk.out.AnnounceDiagnostics(ctx, diags)
+		}
+	})
+}
diff --git a/internal/stacks/stackruntime/internal/stackeval/main_plan.go b/internal/stacks/stackruntime/internal/stackeval/main_plan.go
index c147272531..ca3b07fe39 100644
--- a/internal/stacks/stackruntime/internal/stackeval/main_plan.go
+++ b/internal/stacks/stackruntime/internal/stackeval/main_plan.go
@@ -122,14 +122,7 @@ type PlanOutput struct {
 // planWalk just bundles a [walkState] and a [PlanOutput] together so we can
 // concisely pass them both as a single argument between the all the plan walk
 // driver functions below.
-type planWalk struct {
-	state *walkState
-	out   *PlanOutput
-}
-
-func (w *planWalk) AsyncTask(ctx context.Context, impl func(ctx context.Context)) {
-	w.state.AsyncTask(ctx, impl)
-}
+type planWalk = walkWithOutput[*PlanOutput]
 
 func (m *Main) walkPlanChanges(ctx context.Context, walk *planWalk, stack *Stack) {
 	// We'll get the expansion of any child stack calls going first, so that
diff --git a/internal/stacks/stackruntime/internal/stackeval/output_value.go b/internal/stacks/stackruntime/internal/stackeval/output_value.go
index 5ea9e8146b..0548b34c64 100644
--- a/internal/stacks/stackruntime/internal/stackeval/output_value.go
+++ b/internal/stacks/stackruntime/internal/stackeval/output_value.go
@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig/stackconfigtypes"
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig/typeexpr"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 	"github.com/zclconf/go-cty/cty/convert"
@@ -144,19 +145,28 @@ func (v *OutputValue) CheckResultValue(ctx context.Context, phase EvalPhase) (ct
 	))
 }
 
-// PlanChanges implements Plannable as a plan-time validation of the variable's
-// declaration and of the caller's definition of the variable.
-func (v *OutputValue) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+func (v *OutputValue) checkValid(ctx context.Context, phase EvalPhase) tfdiags.Diagnostics {
 	var diags tfdiags.Diagnostics
 
 	// FIXME: We should really check the type during the validation phase
 	// in OutputValueConfig, rather than the planning phase in OutputValue.
 	_, _, moreDiags := v.CheckResultType(ctx)
 	diags = diags.Append(moreDiags)
-	_, moreDiags = v.CheckResultValue(ctx, PlanPhase)
+	_, moreDiags = v.CheckResultValue(ctx, phase)
 	diags = diags.Append(moreDiags)
 
-	return nil, diags
+	return diags
+}
+
+// PlanChanges implements Plannable as a plan-time validation of the variable's
+// declaration and of the caller's definition of the variable.
+func (v *OutputValue) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+	return nil, v.checkValid(ctx, PlanPhase)
+}
+
+// CheckApply implements ApplyChecker.
+func (v *OutputValue) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	return nil, v.checkValid(ctx, ApplyPhase)
 }
 
 func (v *OutputValue) tracingName() string {
diff --git a/internal/stacks/stackruntime/internal/stackeval/provider.go b/internal/stacks/stackruntime/internal/stackeval/provider.go
index 8d621d1f5b..a4fde076cf 100644
--- a/internal/stacks/stackruntime/internal/stackeval/provider.go
+++ b/internal/stacks/stackruntime/internal/stackeval/provider.go
@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/terraform/internal/stacks/stackaddrs"
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 )
@@ -281,8 +282,7 @@ func (p *Provider) ExprReferenceValue(ctx context.Context, phase EvalPhase) cty.
 	}
 }
 
-// PlanChanges implements Plannable.
-func (p *Provider) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+func (p *Provider) checkValid(ctx context.Context, phase EvalPhase) tfdiags.Diagnostics {
 	var diags tfdiags.Diagnostics
 
 	_, moreDiags := p.CheckForEachValue(ctx, PlanPhase)
@@ -292,7 +292,17 @@ func (p *Provider) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange,
 	// Everything else is instance-specific and so the plan walk driver must
 	// call p.Instances and ask each instance to plan itself.
 
-	return nil, diags
+	return diags
+}
+
+// PlanChanges implements Plannable.
+func (p *Provider) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+	return nil, p.checkValid(ctx, PlanPhase)
+}
+
+// CheckApply implements ApplyChecker.
+func (p *Provider) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	return nil, p.checkValid(ctx, ApplyPhase)
 }
 
 // tracingName implements Plannable.
diff --git a/internal/stacks/stackruntime/internal/stackeval/provider_instance.go b/internal/stacks/stackruntime/internal/stackeval/provider_instance.go
index 88215573b7..e26b4e0935 100644
--- a/internal/stacks/stackruntime/internal/stackeval/provider_instance.go
+++ b/internal/stacks/stackruntime/internal/stackeval/provider_instance.go
@@ -12,6 +12,7 @@ import (
 	"github.com/hashicorp/terraform/internal/providers"
 	"github.com/hashicorp/terraform/internal/stacks/stackaddrs"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/hashicorp/terraform/version"
 	"github.com/zclconf/go-cty/cty"
@@ -257,20 +258,29 @@ func (p *ProviderInstance) ResolveExpressionReference(ctx context.Context, ref s
 	return stack.resolveExpressionReference(ctx, ref, nil, p.repetition)
 }
 
-// PlanChanges implements Plannable.
-func (p *ProviderInstance) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+func (p *ProviderInstance) checkValid(ctx context.Context, phase EvalPhase) tfdiags.Diagnostics {
 	var diags tfdiags.Diagnostics
 
-	_, moreDiags := p.CheckProviderArgs(ctx, PlanPhase)
+	_, moreDiags := p.CheckProviderArgs(ctx, phase)
 	diags = diags.Append(moreDiags)
 
 	// NOTE: CheckClient starts and configures the provider as a side-effect.
 	// If this is a plugin-based provider then the plugin process will stay
-	// running for the remainder of the planning phase.
-	_, moreDiags = p.CheckClient(ctx, PlanPhase)
+	// running for the remainder of the specified evaluation phase.
+	_, moreDiags = p.CheckClient(ctx, phase)
 	diags = diags.Append(moreDiags)
 
-	return nil, diags
+	return diags
+}
+
+// PlanChanges implements Plannable.
+func (p *ProviderInstance) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
+	return nil, p.checkValid(ctx, PlanPhase)
+}
+
+// CheckApply implements ApplyChecker.
+func (p *ProviderInstance) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	return nil, p.checkValid(ctx, ApplyPhase)
 }
 
 // tracingName implements Plannable.
diff --git a/internal/stacks/stackruntime/internal/stackeval/stack.go b/internal/stacks/stackruntime/internal/stackeval/stack.go
index 138bf3c57a..5b52a9b035 100644
--- a/internal/stacks/stackruntime/internal/stackeval/stack.go
+++ b/internal/stacks/stackruntime/internal/stackeval/stack.go
@@ -13,6 +13,7 @@ import (
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig"
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig/typeexpr"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 )
@@ -504,6 +505,13 @@ func (s *Stack) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfd
 	return changes, nil
 }
 
+// CheckApply implements ApplyChecker.
+func (s *Stack) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	// TODO: We should emit an AppliedChange for each output value,
+	// reporting its final value.
+	return nil, nil
+}
+
 func (s *Stack) tracingName() string {
 	addr := s.Addr()
 	if addr.IsRoot() {
diff --git a/internal/stacks/stackruntime/internal/stackeval/stack_call.go b/internal/stacks/stackruntime/internal/stackeval/stack_call.go
index 6bd8220ef4..5a0db30747 100644
--- a/internal/stacks/stackruntime/internal/stackeval/stack_call.go
+++ b/internal/stacks/stackruntime/internal/stackeval/stack_call.go
@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/terraform/internal/stacks/stackaddrs"
 	"github.com/hashicorp/terraform/internal/stacks/stackconfig"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 )
@@ -278,6 +279,21 @@ func (c *StackCall) ExprReferenceValue(ctx context.Context, phase EvalPhase) cty
 	return c.ResultValue(ctx, phase)
 }
 
+func (c *StackCall) checkValid(ctx context.Context, phase EvalPhase) tfdiags.Diagnostics {
+	var diags tfdiags.Diagnostics
+
+	_, moreDiags := c.CheckForEachValue(ctx, phase)
+	diags = diags.Append(moreDiags)
+	_, moreDiags = c.CheckInstances(ctx, phase)
+	diags = diags.Append(moreDiags)
+
+	// All of the other arguments in a stack call get evaluated separately
+	// for each instance of the call, so [StackCallInstance] must deal
+	// with those.
+
+	return diags
+}
+
 // PlanChanges implements Plannable to perform "plan-time validation" of the
 // stack call.
 //
@@ -289,18 +305,12 @@ func (c *StackCall) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange,
 	// can potentially generate diagnostics if the call configuration is
 	// invalid. This is therefore more a "plan-time validation" than actually
 	// planning.
-	var diags tfdiags.Diagnostics
-
-	_, moreDiags := c.CheckForEachValue(ctx, PlanPhase)
-	diags = diags.Append(moreDiags)
-	_, moreDiags = c.CheckInstances(ctx, PlanPhase)
-	diags = diags.Append(moreDiags)
-
-	// All of the other arguments in a stack call get evaluated separately
-	// for each instance of the call, so [StackCallInstance.PlanChanges]
-	// must deal with those.
+	return nil, c.checkValid(ctx, PlanPhase)
+}
 
-	return nil, diags
+// CheckApply implements ApplyChecker.
+func (c *StackCall) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	return nil, c.checkValid(ctx, ApplyPhase)
 }
 
 func (c *StackCall) tracingName() string {
diff --git a/internal/stacks/stackruntime/internal/stackeval/stack_call_instance.go b/internal/stacks/stackruntime/internal/stackeval/stack_call_instance.go
index 8ed5f120a0..17f7b698ff 100644
--- a/internal/stacks/stackruntime/internal/stackeval/stack_call_instance.go
+++ b/internal/stacks/stackruntime/internal/stackeval/stack_call_instance.go
@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/terraform/internal/instances"
 	"github.com/hashicorp/terraform/internal/stacks/stackaddrs"
 	"github.com/hashicorp/terraform/internal/stacks/stackplan"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 	"github.com/zclconf/go-cty/cty/convert"
@@ -157,6 +158,15 @@ func (c *StackCallInstance) ResolveExpressionReference(ctx context.Context, ref
 	return stack.resolveExpressionReference(ctx, ref, nil, c.repetition)
 }
 
+func (c *StackCallInstance) checkValid(ctx context.Context, phase EvalPhase) tfdiags.Diagnostics {
+	var diags tfdiags.Diagnostics
+
+	_, moreDiags := c.CheckInputVariableValues(ctx, PlanPhase)
+	diags = diags.Append(moreDiags)
+
+	return diags
+}
+
 // PlanChanges implements Plannable by performing plan-time validation of
 // all of the per-instance arguments in the stack call configuration.
 //
@@ -166,12 +176,13 @@ func (c *StackCallInstance) ResolveExpressionReference(ctx context.Context, ref
 func (c *StackCallInstance) PlanChanges(ctx context.Context) ([]stackplan.PlannedChange, tfdiags.Diagnostics) {
 	// This is really just a "plan-time validation" behavior, since stack
 	// calls never contribute directly to the planned changes.
-	var diags tfdiags.Diagnostics
-
-	_, moreDiags := c.CheckInputVariableValues(ctx, PlanPhase)
-	diags = diags.Append(moreDiags)
+	return nil, c.checkValid(ctx, PlanPhase)
+}
 
-	return nil, diags
+// CheckApply implements ApplyChecker by confirming that the input variable
+// values are still valid after resolving any upstream changes.
+func (c *StackCallInstance) CheckApply(ctx context.Context) ([]stackstate.AppliedChange, tfdiags.Diagnostics) {
+	return nil, c.checkValid(ctx, ApplyPhase)
 }
 
 // tracingName implements Plannable.
diff --git a/internal/stacks/stackruntime/internal/stackeval/walk.go b/internal/stacks/stackruntime/internal/stackeval/walk.go
index f9f3eef3bb..df86a052de 100644
--- a/internal/stacks/stackruntime/internal/stackeval/walk.go
+++ b/internal/stacks/stackruntime/internal/stackeval/walk.go
@@ -20,8 +20,7 @@ import (
 // to share, with everything else dealt with using -- as much as possible --
 // "normal code".
 type walkState struct {
-	wg    sync.WaitGroup
-	diags syncDiagnostics
+	wg sync.WaitGroup
 
 	// handleDiags is called for each call to [walkState.AddDiags], with
 	// a set of diagnostics produced from that call's arguments.
@@ -126,3 +125,15 @@ func (ws *walkState) AddDiags(new ...any) {
 }
 
 type walkTaskContextKey struct{}
+
+// walkWithOutput combines a [walkState] with some other object that allows
+// emitting output events to a caller, so that walk codepaths can conveniently
+// pass these both together as a single argument.
+type walkWithOutput[Output any] struct {
+	state *walkState
+	out   Output
+}
+
+func (w *walkWithOutput[Output]) AsyncTask(ctx context.Context, impl func(ctx context.Context)) {
+	w.state.AsyncTask(ctx, impl)
+}
diff --git a/internal/stacks/stackstate/applied_change.go b/internal/stacks/stackstate/applied_change.go
new file mode 100644
index 0000000000..19aae61605
--- /dev/null
+++ b/internal/stacks/stackstate/applied_change.go
@@ -0,0 +1,19 @@
+package stackstate
+
+import (
+	"github.com/hashicorp/terraform/internal/rpcapi/terraform1"
+)
+
+// AppliedChange represents a single isolated change, emitted as
+// part of a stream of applied changes during the ApplyStackChanges RPC API
+// operation.
+//
+// Each AppliedChange becomes a single event in the RPC API, which itself
+// has zero or more opaque raw plan messages that the caller must collect and
+// provide verbatim during planning and zero or more "description" messages
+// that are to give the caller realtime updates about the planning process.
+type AppliedChange interface {
+	// AppliedChangeProto returns the protocol buffers representation of
+	// the change, ready to be sent verbatim to an RPC API client.
+	AppliedChangeProto() (*terraform1.AppliedChange, error)
+}