execgraph: Marshaling an unmarshaling of execution graphs

To support the workflow of saving a plan to disk and applying it on some
other machine we need to be able to represent the execution graph as a byte
stream and then reload it later to produce an equivalent execution graph.

This is an initial implementation of that, based on the way the execgraph
package currently represents execution graphs. We may change that
representation more in future as we get more experience working in the new
architecture, but this is intended as part of our "walking skeleton" phase
where we try to get the new architecture working end-to-end with simple
configurations as soon as possible to help verify that we're even on the
right track with this new approach, and try to find unknown unknowns that
we ought to deal with before we get too deep into this.

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

internal/engine/internal/execgraph/builder.go

@@ -42,6 +42,7 @@ type Builder struct {
 	providerAddrRefs       map[addrs.Provider]ResultRef[addrs.Provider]
 	providerInstConfigRefs addrs.Map[addrs.AbsProviderInstanceCorrect, ResultRef[cty.Value]]
 	openProviderRefs       addrs.Map[addrs.AbsProviderInstanceCorrect, resultWithCloseBlockers[providers.Configured]]
+	emptyWaiterRef         ResultRef[struct{}]
 }
 
 func NewBuilder() *Builder {
@@ -54,6 +55,7 @@ func NewBuilder() *Builder {
 		providerAddrRefs:       make(map[addrs.Provider]ResultRef[addrs.Provider]),
 		providerInstConfigRefs: addrs.MakeMap[addrs.AbsProviderInstanceCorrect, ResultRef[cty.Value]](),
 		openProviderRefs:       addrs.MakeMap[addrs.AbsProviderInstanceCorrect, resultWithCloseBlockers[providers.Configured]](),
+		emptyWaiterRef:         nil, // will be populated on first request for an empty waiter
 	}
 }
 
@@ -192,9 +194,10 @@ func (b *Builder) ProviderInstanceConfig(addr addrs.AbsProviderInstanceCorrect)
 // the graph at once, although each distinct provider instance address gets
 // only one set of nodes added and then subsequent calls get references to
 // the same operation results.
-func (b *Builder) ProviderInstance(addr addrs.AbsProviderInstanceCorrect, waitFor []AnyResultRef) (ResultRef[providers.Configured], RegisterCloseBlockerFunc) {
+func (b *Builder) ProviderInstance(addr addrs.AbsProviderInstanceCorrect, waitFor AnyResultRef) (ResultRef[providers.Configured], RegisterCloseBlockerFunc) {
 	configResult := b.ProviderInstanceConfig(addr)
 	providerAddrResult := b.ConstantProviderAddr(addr.Config.Config.Provider)
+	waiter := b.ensureWaiterRef(waitFor)
 
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -203,7 +206,6 @@ func (b *Builder) ProviderInstance(addr addrs.AbsProviderInstanceCorrect, waitFo
 	if existing, ok := b.openProviderRefs.GetOk(addr); ok {
 		return existing.Result, existing.CloseBlockerFunc
 	}
-	waiter := b.makeWaiter(waitFor)
 	openResult := operationRef[providers.Configured](b, operationDesc{
 		opCode:   opOpenProvider,
 		operands: []AnyResultRef{providerAddrResult, configResult, waiter},
@@ -220,6 +222,45 @@ func (b *Builder) ProviderInstance(addr addrs.AbsProviderInstanceCorrect, waitFo
 	return openResult, registerCloseBlocker
 }
 
+// OpenProviderClient registers an operation for opening a client for a
+// particular provider instance.
+//
+// Direct callers should typically use [Builder.ProviderInstance] instead,
+// because it automatically deduplicates requests for the same provider
+// instance and registers the associated "close" operation for the provider
+// instance. This method is here primarily for the benefit of [UnmarshalGraph],
+// which needs to be able to work with the individual operation nodes when
+// reconstructing a previously-marshaled graph.
+func (b *Builder) OpenProviderClient(providerAddr ResultRef[addrs.Provider], config ResultRef[cty.Value], waitFor AnyResultRef) ResultRef[providers.Configured] {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	waiter := b.ensureWaiterRef(waitFor)
+	return operationRef[providers.Configured](b, operationDesc{
+		opCode:   opOpenProvider,
+		operands: []AnyResultRef{providerAddr, config, waiter},
+	})
+}
+
+// CloseProviderClient registers an operation for closing a provider client
+// that was previously opened through [Builder.OpenProviderClient].
+//
+// Direct callers should typically use [Builder.ProviderInstance] instead,
+// because it deals with all of the ceremony of opening and closing provider
+// clients. This method is here primarily for the benefit of [UnmarshalGraph],
+// which needs to be able to work with the individual operation nodes when
+// reconstructing a previously-marshaled graph.
+func (b *Builder) CloseProviderClient(clientResult ResultRef[providers.Configured], waitFor AnyResultRef) ResultRef[struct{}] {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	waiter := b.ensureWaiterRef(waitFor)
+	return operationRef[struct{}](b, operationDesc{
+		opCode:   opCloseProvider,
+		operands: []AnyResultRef{clientResult, waiter},
+	})
+}
+
 // ManagedResourceObjectFinalPlan registers an operation to decide the "final plan" for a managed
 // resource instance object, which may or may not be "desired".
 //
@@ -250,11 +291,12 @@ func (b *Builder) ManagedResourceObjectFinalPlan(
 	priorState ResultRef[*states.ResourceInstanceObject],
 	plannedVal ResultRef[cty.Value],
 	providerClient ResultRef[providers.Configured],
-	waitFor []AnyResultRef,
+	waitFor AnyResultRef,
 ) ResultRef[*ManagedResourceObjectFinalPlan] {
-	// We'll aggregate all of the waitFor nodes into a waiter node so we
-	// can pass it as just a single argument to the operation.
-	waiter := b.makeWaiter(waitFor)
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	waiter := b.ensureWaiterRef(waitFor)
 	return operationRef[*ManagedResourceObjectFinalPlan](b, operationDesc{
 		opCode:   opManagedFinalPlan,
 		operands: []AnyResultRef{desiredInst, priorState, plannedVal, providerClient, waiter},
@@ -271,6 +313,9 @@ func (b *Builder) ApplyManagedResourceObjectChanges(
 	finalPlan ResultRef[*ManagedResourceObjectFinalPlan],
 	providerClient ResultRef[providers.Configured],
 ) ResultRef[*states.ResourceInstanceObject] {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	return operationRef[*states.ResourceInstanceObject](b, operationDesc{
 		opCode:   opManagedApplyChanges,
 		operands: []AnyResultRef{finalPlan, providerClient},
@@ -280,9 +325,12 @@ func (b *Builder) ApplyManagedResourceObjectChanges(
 func (b *Builder) DataRead(
 	desiredInst ResultRef[*eval.DesiredResourceInstance],
 	providerClient ResultRef[providers.Configured],
-	waitFor []AnyResultRef,
+	waitFor AnyResultRef,
 ) ResultRef[*states.ResourceInstanceObject] {
-	waiter := b.makeWaiter(waitFor)
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	waiter := b.ensureWaiterRef(waitFor)
 	return operationRef[*states.ResourceInstanceObject](b, operationDesc{
 		opCode:   opDataRead,
 		operands: []AnyResultRef{desiredInst, providerClient, waiter},
@@ -307,6 +355,18 @@ func (b *Builder) SetResourceInstanceFinalStateResult(addr addrs.AbsResourceInst
 	b.graph.resourceInstanceResults.Put(addr, result)
 }
 
+// Waiter creates a "fan-in" node where a single result depends on the
+// completion of an arbitrary number of other results.
+//
+// The values produced by the dependencies are discarded; this only creates
+// a "must happen after" relationship with the given dependencies.
+func (b *Builder) Waiter(dependencies ...AnyResultRef) AnyResultRef {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	return b.makeWaiter(dependencies)
+}
+
 // operationRef is a helper used by all of the [Builder] methods that produce
 // "operation" nodes, dealing with the common registration part.
 //
@@ -336,8 +396,36 @@ func (b *Builder) makeCloseBlocker() (ResultRef[struct{}], RegisterCloseBlockerF
 	return ref, registerFunc
 }
 
+// ensureWaiterRef exists to make it more convenient for callers of Builder
+// to populate "waitFor" arguments, by normalizing whatever was provided
+// so that it's definitely a waiter reference.
+//
+// The given ref can be nil to represent waiting for nothing, in which case
+// the result is a reference to an empty waiter. If the given ref is not nil
+// but is also not of the correct result type for a waiter then it'll be
+// wrapped in a waiter with only a single dependency and the result will
+// be a reference to that waiter.
+func (b *Builder) ensureWaiterRef(given AnyResultRef) ResultRef[struct{}] {
+	if ret, ok := given.(waiterResultRef); ok {
+		return ret
+	}
+	if given == nil {
+		return b.makeWaiter(nil)
+	}
+	return b.makeWaiter([]AnyResultRef{given})
+}
+
 func (b *Builder) makeWaiter(waitFor []AnyResultRef) ResultRef[struct{}] {
-	idx := appendIndex(&b.graph.waiters, []AnyResultRef{})
+	if len(waitFor) == 0 {
+		// Empty waiters tend to appear in multiple places, so we'll just
+		// allocate a single one on request and reuse it.
+		if b.emptyWaiterRef == nil {
+			idx := appendIndex(&b.graph.waiters, nil)
+			b.emptyWaiterRef = waiterResultRef{idx}
+		}
+		return b.emptyWaiterRef
+	}
+	idx := appendIndex(&b.graph.waiters, waitFor)
 	return waiterResultRef{idx}
 }
 

internal/engine/internal/execgraph/execgraphproto/doc.go

@@ -0,0 +1,18 @@
+// Copyright (c) The OpenTofu Authors
+// SPDX-License-Identifier: MPL-2.0
+// Copyright (c) 2023 HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+// Package execgraphproto contains just the protocol buffers models we use
+// for marshaling and unmarshaling execution graphs.
+//
+// The logic for converting to and from the types defined in here is
+// encapsulated in package execgraph. The specific serialization format for
+// execution graphs is an implementation detail that may change arbitrarily
+// between OpenTofu versions: it's not supported to take an execution graph
+// marshaled by one OpenTofu version and then try to unmarshal it with a
+// different OpenTofu version. Even the fact that there is an execution
+// graph _at all_ is an implementation detail, with the public-facing model
+// only representing a set of high-level actions against resource instances
+// and output values.
+package execgraphproto

internal/engine/internal/execgraph/execgraphproto/execgraph.proto

@@ -0,0 +1,148 @@
+// Copyright (c) The OpenTofu Authors
+// SPDX-License-Identifier: MPL-2.0
+// Copyright (c) 2023 HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+edition = "2024";
+package opentofu_execgraph;
+
+// For OpenTofu's own parsing, the proto stub types go into an internal Go
+// package. The public API is in github.com/opentofu/opentofu/plans/planfile .
+option go_package = "github.com/opentofu/opentofu/internal/engine/internal/execgraph/execgraphproto";
+
+// ExecutionGraph is the root message type for an execution graph file.
+//
+// Note that we expect execution graphs to be saved as part of a larger plan
+// file structure that should represent general metadata such as which OpenTofu
+// version a particular execution graph was created by. This message type and
+// its nested message types are focused only on representing the details of
+// the execution graph itself.
+message ExecutionGraph {
+    // The various "elements" of the graph.
+    //
+    // This current serialization should be thought of as a sequence of
+    // mutations that each add something new to the graph, where the available
+    // actions correspond to the methods of execgraph.Builder. Each message
+    // in the sequence therefore corresponds to a method called on an internal
+    // builder used to reconstruct the graph during unmarshaling.
+    //
+    // We use execgraph.Builder when reloading because that helps us to
+    // validate the graph data using the result type information produced by
+    // the builder method results, although a round-trip through this
+    // serialization only guarantees a functionally-equivalent graph, not an
+    // exactly-identical graph. (i.e. the internal slices used to represent
+    // the graph elements might be in a different order after round-tripping,
+    // and the text-based debug representation of the graph will reflect
+    // those changes even though the graph _topology_ should be identical.)
+    repeated Element elements = 1;
+
+    // Map from resource instance addresses to indices into the "elements"
+    // field, specifying which element produces the "final state" result
+    // for each resource instance.
+    map<string, uint64> resource_instance_results = 2;
+}
+
+// One "element" of an execution graph. Refer to ExecutionGraph.elements for
+// more information.
+message Element {
+    // Each element has only one request field populated, which therefore
+    // implies the type of element to produce. These fields should all have
+    // identifiers less than 128 so that the discriminant encodes as just one
+    // byte in the protobuf wire format.
+    //
+    // The field types should all be as simple as possible to further shrink the
+    // size of their protobuf encoding. Don't worry about forward-compatibility,
+    // because a specific version of OpenTofu is only required to unmarshal
+    // execution graphs produced by itself: we can always make the
+    // representation of a specific request type more complicated in a later
+    // release if we need to.
+    //
+    // Where one request refers to the result of another request, the referrer
+    // must be serialized after what it refers to and refers back to the earlier
+    // result using a zero-based uint64 index into the ExecutionGraph.elements
+    // list. The fields representing such references conventionally have names
+    // ending in "_result". During unmarshaling the loader will verify that
+    // these references each refer to a result of the appropriate type, which
+    // should always be true for a serialization of any graph produced by
+    // execgraph.Builder that was not subsequently tampered with.
+    oneof request {
+        // A constant value, corresponding to execgraph.Builder.ConstantValue.
+        //
+        // The value is a MessagePack serialization of a cty.Value using
+        // cty.DynamicPseudoType as the serialization type constraint, and so
+        // this represents both the value and the value's dynamic type.
+        bytes constant_value = 1;
+
+        // A constant address for a provider type, corresponding to
+        // execgraph.Builder.ConstantProviderAddr.
+        //
+        // The value is a fully-qualified provider source address in the
+        // usual syntax, such as "registry.opentofu.org/hashicorp/aws".
+        string constant_provider_addr = 2;
+
+        // Request for information about a specific "desired" resource
+        // instance, corresponding to execgraph.Builder.DesiredResourceInstance.
+        //
+        // The value is a fully-qualified resource instance address in the
+        // syntax that addrs.AbsResourceInstance.String produces.
+        string desired_resource_instance = 3;
+
+        // Request for the prior state data for the current object for a
+        // specific resource instance, corresponding to
+        // execgraph.Builder.ResourceInstancePriorState.
+        //
+        // The value is a fully-qualified resource instance address in the
+        // syntax that addrs.AbsResourceInstance.String produces.
+        string resource_instance_prior_state = 4;
+
+        // Request for the state data for a specific deposed object of a
+        // specific resource instance, corresponding to
+        // execgraph.Builder.ResourceDeposedObjectState.
+        DeposedResourceInstanceObject resource_instance_deposed_object_state = 5;
+
+        // Request for the configuration value for a provider instance,
+        // corresponding to execgraph.Builder.ProviderInstanceConfig .
+        string provider_instance_config = 6;
+
+        // An operation to be performed during execution, corresponding to
+        // various different methods of execgraph.Builder depending on the
+        // opcode.
+        //
+        // The unmarshal code is responsible for figuring out which Builder
+        // method to call based on the opcode, handled as part of its validation
+        // logic.
+        Operation operation = 7;
+
+        // A collection of results whose completion an operation can depend
+        // on without actually making any use of the values they produce.
+        //
+        // We use this to represent explicit dependencies between resource
+        // instances and providers so that the execution order will respect
+        // the dependencies regardless of whether the dependencies were
+        // inferred automatically from expressions or declared explicitly
+        // using a "depends_on" argument.
+        Waiter waiter = 8;
+    }
+}
+
+message DeposedResourceInstanceObject {
+    // A fully-qualified resource instance address in the
+    // syntax that addrs.AbsResourceInstance.String produces.
+    string instance_addr = 1;
+    // The key of the requested deposed object associated with the
+    // resource instance.
+    string deposed_key = 2;
+}
+
+message Operation {
+    // The raw opcode of the operation to be performed.
+    uint64 opcode = 1;
+    // The result ids for all of the operands. The number of results and their
+    // meanings depend on the opcode, and are validated during unmarshaling.
+    repeated uint64 operands = 2;
+}
+
+message Waiter {
+    // Result ids of all of the results that must be awaited.
+    repeated uint64 results = 1;
+}

internal/engine/internal/execgraph/graph.go

@@ -7,6 +7,8 @@ package execgraph
 
 import (
 	"fmt"
+	"maps"
+	"slices"
 	"strings"
 
 	"github.com/zclconf/go-cty-debug/ctydebug"
@@ -124,8 +126,16 @@ func (g *Graph) DebugRepr() string {
 	if g.resourceInstanceResults.Len() != 0 && (len(g.ops) != 0 || len(g.constantVals) != 0) {
 		buf.WriteByte('\n')
 	}
+	// We'll sort the resource instance results by instance address key just
+	// so that the resulting order is consistent for comparison in tests.
+	resourceInstanceResults := make(map[string]string)
 	for _, elem := range g.resourceInstanceResults.Elems {
-		fmt.Fprintf(&buf, "%s = %s;\n", elem.Key.String(), g.resultDebugRepr(elem.Value))
+		resourceInstanceResults[elem.Key.String()] = g.resultDebugRepr(elem.Value)
+	}
+	resourceInstanceAddrs := slices.Collect(maps.Keys(resourceInstanceResults))
+	slices.Sort(resourceInstanceAddrs)
+	for _, addrStr := range resourceInstanceAddrs {
+		fmt.Fprintf(&buf, "%s = %s;\n", addrStr, resourceInstanceResults[addrStr])
 	}
 	return buf.String()
 }

internal/engine/internal/execgraph/graph_marshal.go

@@ -0,0 +1,242 @@
+// Copyright (c) The OpenTofu Authors
+// SPDX-License-Identifier: MPL-2.0
+// Copyright (c) 2023 HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package execgraph
+
+import (
+	"fmt"
+
+	"github.com/zclconf/go-cty/cty"
+	ctymsgpack "github.com/zclconf/go-cty/cty/msgpack"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/opentofu/opentofu/internal/addrs"
+	"github.com/opentofu/opentofu/internal/engine/internal/execgraph/execgraphproto"
+	"github.com/opentofu/opentofu/internal/states"
+)
+
+// Marshal produces an opaque byte slice representing the given graph,
+// which can then be passed to [UnmarshalGraph] to produce a
+// functionally-equivalent graph.
+func (g *Graph) Marshal() []byte {
+	m := &graphMarshaler{
+		graph:   g,
+		indices: make(map[AnyResultRef]uint64),
+	}
+
+	// The operations are the main essence of any graph, so we let those
+	// drive the process here and then append anything else they refer to
+	// as we analyze their arguments.
+	//
+	// This approach forces the elements into a topological order as we
+	// go along, which [UnmarshalGraph] relies on so it can validate the
+	// elements gradually during loading by referring back to elements
+	// that it had previously loaded.
+	for idx := range g.ops {
+		m.EnsureOperationPresent(idx)
+	}
+	// We also record which element produces the final state result for
+	// each resource instance.
+	m.EnsureResourceInstanceResultsPresent(g.resourceInstanceResults)
+
+	return m.Bytes()
+}
+
+type graphMarshaler struct {
+	graph                   *Graph
+	elems                   []*execgraphproto.Element
+	indices                 map[AnyResultRef]uint64
+	resourceInstanceResults map[string]uint64
+}
+
+func (m *graphMarshaler) EnsureOperationPresent(idx int) uint64 {
+	// We have a little problem here because we don't retain information
+	// about the result type of each operation directly in the graph -- that's
+	// encoded in the references instead -- but we really only need the
+	// raw operation indices here anyway and so as a private detail for
+	// graphMarshaler alone we'll just use a type-erased operation ref
+	// type, and then the unmarshal code will check that the result types
+	// are actually consistent as part of its validation process.
+	erasedRef := m.operationRefFromIdx(idx)
+	return m.ensureRefTarget(erasedRef)
+}
+
+func (m *graphMarshaler) EnsureResourceInstanceResultsPresent(results addrs.Map[addrs.AbsResourceInstance, ResultRef[*states.ResourceInstanceObject]]) {
+	m.resourceInstanceResults = make(map[string]uint64)
+	for _, mapElem := range results.Elems {
+		instAddr := mapElem.Key
+		resultRef := mapElem.Value
+		resultIdx := m.ensureRefTarget(resultRef)
+		m.resourceInstanceResults[instAddr.String()] = resultIdx
+	}
+}
+
+func (m *graphMarshaler) ensureRefTarget(ref AnyResultRef) uint64 {
+	if ref == nil {
+		panic("ensureRefTarget with nil ref")
+	}
+	if opRef, ok := ref.(anyOperationResultRef); ok {
+		// Our lookup table doesn't care about the result type of each
+		// operation, so we'll erase any type information to obtain
+		// the key that's actually used in the map.
+		ref = m.typeErasedOperationRef(opRef)
+	}
+	if existing, ok := m.indices[ref]; ok {
+		return existing
+	}
+
+	switch ref := ref.(type) {
+	case valueResultRef:
+		return m.addConstantValue(ref, m.graph.constantVals[ref.index])
+	case providerAddrResultRef:
+		return m.addProviderAddr(ref, m.graph.providerAddrs[ref.index])
+	case desiredResourceInstanceResultRef:
+		return m.addDesiredStateRef(ref, m.graph.desiredStateRefs[ref.index])
+	case resourceInstancePriorStateResultRef:
+		return m.addPriorStateRef(ref, m.graph.priorStateRefs[ref.index])
+	case providerInstanceConfigResultRef:
+		return m.addProviderInstanceConfigRef(ref, m.graph.providerInstConfigRefs[ref.index])
+	case operationResultRef[struct{}]:
+		return m.addOperationWithDependencies(ref, m.graph.ops[ref.index])
+	case waiterResultRef:
+		return m.addWaiter(ref, m.graph.waiters[ref.index])
+	default:
+		// Should not get here because the cases above should cover all of
+		// the variants of AnyResultRef.
+		panic(fmt.Sprintf("graphMarshaler doesn't know how to handle %#v", ref))
+	}
+}
+
+func (m *graphMarshaler) addConstantValue(ref valueResultRef, v cty.Value) uint64 {
+	raw, err := ctymsgpack.Marshal(v, cty.DynamicPseudoType)
+	if err != nil {
+		// Here we assume that any value we're given is one either produced by
+		// OpenTofu itself or at least previously unmarshaled by OpenTofu, and
+		// thus we should never encounter an error trying to serialize it.
+		panic(fmt.Sprintf("constant value %d is not MessagePack-compatible: %s", ref.index, err))
+	}
+	return m.newElement(ref, func(elem *execgraphproto.Element) {
+		elem.SetConstantValue(raw)
+	})
+}
+
+func (m *graphMarshaler) addProviderAddr(ref providerAddrResultRef, addr addrs.Provider) uint64 {
+	addrStr := addr.String()
+	return m.newElement(ref, func(elem *execgraphproto.Element) {
+		elem.SetConstantProviderAddr(addrStr)
+	})
+}
+
+func (m *graphMarshaler) addDesiredStateRef(ref desiredResourceInstanceResultRef, addr addrs.AbsResourceInstance) uint64 {
+	addrStr := addr.String()
+	return m.newElement(ref, func(elem *execgraphproto.Element) {
+		elem.SetDesiredResourceInstance(addrStr)
+	})
+}
+
+func (m *graphMarshaler) addPriorStateRef(ref resourceInstancePriorStateResultRef, target resourceInstanceStateRef) uint64 {
+	// We serialize these ones a little differently depending on whether there's
+	// a deposed key, because deposed objects in prior state are relatively
+	// rare and we'd prefer a more compact representation of the more common
+	// case of describing a "current" resource instance object.
+	return m.newElement(ref, func(elem *execgraphproto.Element) {
+		if target.DeposedKey == states.NotDeposed {
+			elem.SetResourceInstancePriorState(target.ResourceInstance.String())
+		} else {
+			var req execgraphproto.DeposedResourceInstanceObject
+			req.Reset()
+			req.SetInstanceAddr(target.ResourceInstance.String())
+			req.SetDeposedKey(target.DeposedKey.String())
+			elem.SetResourceInstanceDeposedObjectState(&req)
+		}
+	})
+}
+
+func (m *graphMarshaler) addProviderInstanceConfigRef(ref providerInstanceConfigResultRef, addr addrs.AbsProviderInstanceCorrect) uint64 {
+	addrStr := addr.String()
+	return m.newElement(ref, func(elem *execgraphproto.Element) {
+		elem.SetProviderInstanceConfig(addrStr)
+	})
+}
+
+func (m *graphMarshaler) addOperationWithDependencies(ref operationResultRef[struct{}], desc operationDesc) uint64 {
+	// During marshaling we just assume the graph was correctly constructed
+	// and save the operation items in their raw form. The unmarshal code
+	// then uses the opcode to decide which method of [Builder] to call
+	// for each operation as part of its validation logic.
+	var rawRefIdxs []uint64
+	if len(desc.operands) != 0 {
+		rawRefIdxs = make([]uint64, len(desc.operands))
+	}
+	for i, operandRef := range desc.operands {
+		rawRefIdxs[i] = m.ensureRefTarget(operandRef)
+	}
+	return m.newElement(ref, func(elem *execgraphproto.Element) {
+		opCode := uint64(desc.opCode)
+		elem.SetOperation(execgraphproto.Operation_builder{
+			Opcode:   &opCode,
+			Operands: rawRefIdxs,
+		}.Build())
+	})
+}
+
+func (m *graphMarshaler) addWaiter(ref waiterResultRef, deps []AnyResultRef) uint64 {
+	rawRefIdxs := make([]uint64, len(deps))
+	for i, depRef := range deps {
+		rawRefIdxs[i] = m.ensureRefTarget(depRef)
+	}
+	return m.newElement(ref, func(elem *execgraphproto.Element) {
+		elem.SetWaiter(execgraphproto.Waiter_builder{
+			Results: rawRefIdxs,
+		}.Build())
+	})
+}
+
+// typeErasedOperationRef erases the specific result type information from an
+// operation result ref because for serialization purposes we only really
+// care about the indices: the unmarshal code will recover the type information
+// again through its validation logic.
+func (m *graphMarshaler) typeErasedOperationRef(ref anyOperationResultRef) operationResultRef[struct{}] {
+	return operationResultRef[struct{}]{ref.operationResultIndex()}
+}
+
+// operationRefFromIdx is similar to [typeErasedOperationRef] but starts with
+// a raw index into the slice of operations, without any type information at all.
+func (m *graphMarshaler) operationRefFromIdx(idx int) operationResultRef[struct{}] {
+	return operationResultRef[struct{}]{idx}
+}
+
+// newElement is the low-level primitive beneath all of the methods that add
+// elements to the marshaled version of the graph.
+//
+// Constructing protobuf messages through the "opaque"-style API is somewhat
+// awkward, so we use a callback here so that the caller can focus just on
+// populating an empty-but-valid [execgraphproto.Element] using one of the
+// setters on that type. The provided function MUST call exactly one of the
+// setters from the oneOf message, since the method chosen also decides which
+// element type gets recorded in the serialized messages.
+func (m *graphMarshaler) newElement(ref AnyResultRef, build func(*execgraphproto.Element)) uint64 {
+	var elem execgraphproto.Element
+	elem.Reset()
+	build(&elem)
+	idx := uint64(len(m.elems))
+	m.elems = append(m.elems, &elem)
+	m.indices[ref] = idx
+	return idx
+}
+
+func (m *graphMarshaler) Bytes() []byte {
+	var root execgraphproto.ExecutionGraph
+	root.Reset()
+	root.SetElements(m.elems)
+	root.SetResourceInstanceResults(m.resourceInstanceResults)
+	ret, err := proto.Marshal(&root)
+	// We constructed everything in this overall message, so if it isn't
+	// marshalable then that's always a bug somewhere in the code in this file.
+	if err != nil {
+		panic(fmt.Sprintf("produced unmarshalable protobuf message: %s", err))
+	}
+	return ret
+}

internal/engine/internal/execgraph/graph_test.go

@@ -0,0 +1,296 @@
+// Copyright (c) The OpenTofu Authors
+// SPDX-License-Identifier: MPL-2.0
+// Copyright (c) 2023 HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package execgraph
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/zclconf/go-cty/cty"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/opentofu/opentofu/internal/addrs"
+	"github.com/opentofu/opentofu/internal/engine/internal/execgraph/execgraphproto"
+)
+
+func TestGraphMarshalUnmarshalValid(t *testing.T) {
+	// This test exercises our graph marshal and unmarshal behavior by
+	// feeding the marshal result directly into unmarshal and then testing
+	// whether the resulting graph matches our expectations. The fine details
+	// of how things get serialized are not particularly important as long
+	// as we can get a functionally-equivalent graph back out again, and so
+	// this is a pragmatic way to get good enough test coverage while avoiding
+	// the need to update lots of fiddly tests each time we change the
+	// serialization format.
+	//
+	// (The specific serialization format is not a compatibility constraint
+	// because we explicitly disallow applying a plan created by one version of
+	// OpenTofu with a different version of OpenTofu, so it's not justified
+	// to unit-test the specific serialization details.)
+
+	tests := map[string]struct {
+		// InputGraph is a function that constructs the graph that should
+		// be round-tripped through the marshalling code. Implementations
+		// of this should typically aim to construct graphs of similar
+		// shape to those that the planning engine might construct.
+		InputGraph func(*Builder) *Graph
+		// WantGraph is a string representation of the expected output graph,
+		// using the syntax returned by [Graph.DebugRepr]. This can use a
+		// raw string literal indented to align with the surrounding code
+		// because we'll trim off the leading and trailing space from each line
+		// before comparing.
+		WantGraph string
+
+		// We intentionally focus only on valid input here because we only
+		// expect to be parsing graphs produced by OpenTofu itself, and so any
+		// errors we encounter are either bugs in OpenTofu or caused by
+		// something outside of OpenTofu tampering with the serialized graph.
+		// The error handling in [UnmarshalGraph] is primarily to help us with
+		// debugging, because end-users should never see those errors unless
+		// we've made a mistake somewhere.
+	}{
+		"managed resource instance final plan and apply": {
+			// This is intended to mimic how the planning engine would represent
+			// the process of final-planning and applying an action on a
+			// managed resource instance.
+			func(builder *Builder) *Graph {
+				instAddr := addrs.Resource{
+					Mode: addrs.ManagedResourceMode,
+					Type: "test",
+					Name: "example",
+				}.Absolute(addrs.RootModuleInstance).Instance(addrs.NoKey)
+				desiredInst := builder.DesiredResourceInstance(instAddr)
+				priorState := builder.ResourceInstancePriorState(instAddr)
+				plannedVal := builder.ConstantValue(cty.ObjectVal(map[string]cty.Value{
+					"name": cty.StringVal("thingy"),
+				}))
+				providerClient, registerUser := builder.ProviderInstance(
+					addrs.AbsProviderInstanceCorrect{
+						Config: addrs.AbsProviderConfigCorrect{
+							Config: addrs.ProviderConfigCorrect{
+								Provider: addrs.NewBuiltInProvider("test"),
+							},
+						},
+					},
+					nil,
+				)
+				finalPlan := builder.ManagedResourceObjectFinalPlan(desiredInst, priorState, plannedVal, providerClient, nil)
+				newState := builder.ApplyManagedResourceObjectChanges(finalPlan, providerClient)
+				registerUser(newState)
+				builder.SetResourceInstanceFinalStateResult(instAddr, newState)
+				return builder.Finish()
+			},
+			`
+				v[0] = cty.ObjectVal(map[string]cty.Value{
+					"name": cty.StringVal("thingy"),
+				});
+
+				r[0] = OpenProvider(provider("terraform.io/builtin/test"), providerInstConfig(provider["terraform.io/builtin/test"]), await());
+				r[1] = ManagedFinalPlan(desired(test.example), priorState(test.example), v[0], r[0], await());
+				r[2] = ManagedApplyChanges(r[1], r[0]);
+				r[3] = CloseProvider(r[0], await(r[2]));
+
+				test.example = r[2];
+			`,
+		},
+		"data resource instance read": {
+			// This is intended to mimic how the planning engine would represent
+			// the process of reading the data for a data resource instance.
+			func(builder *Builder) *Graph {
+				instAddr := addrs.Resource{
+					Mode: addrs.DataResourceMode,
+					Type: "test",
+					Name: "example",
+				}.Absolute(addrs.RootModuleInstance).Instance(addrs.NoKey)
+				desiredInst := builder.DesiredResourceInstance(instAddr)
+				providerClient, registerUser := builder.ProviderInstance(
+					addrs.AbsProviderInstanceCorrect{
+						Config: addrs.AbsProviderConfigCorrect{
+							Config: addrs.ProviderConfigCorrect{
+								Provider: addrs.NewBuiltInProvider("test"),
+							},
+						},
+					},
+					nil,
+				)
+				newState := builder.DataRead(desiredInst, providerClient, nil)
+				registerUser(newState)
+				builder.SetResourceInstanceFinalStateResult(instAddr, newState)
+				return builder.Finish()
+			},
+			`
+				r[0] = OpenProvider(provider("terraform.io/builtin/test"), providerInstConfig(provider["terraform.io/builtin/test"]), await());
+				r[1] = DataRead(desired(data.test.example), r[0], await());
+				r[2] = CloseProvider(r[0], await(r[1]));
+
+				data.test.example = r[1];
+			`,
+		},
+		"data resource instance reads with dependency": {
+			func(builder *Builder) *Graph {
+				instAddr1 := addrs.Resource{
+					Mode: addrs.DataResourceMode,
+					Type: "test",
+					Name: "example1",
+				}.Absolute(addrs.RootModuleInstance).Instance(addrs.NoKey)
+				instAddr2 := addrs.Resource{
+					Mode: addrs.DataResourceMode,
+					Type: "test",
+					Name: "example2",
+				}.Absolute(addrs.RootModuleInstance).Instance(addrs.NoKey)
+				desiredInst1 := builder.DesiredResourceInstance(instAddr1)
+				desiredInst2 := builder.DesiredResourceInstance(instAddr2)
+				providerClient, registerUser := builder.ProviderInstance(
+					addrs.AbsProviderInstanceCorrect{
+						Config: addrs.AbsProviderConfigCorrect{
+							Config: addrs.ProviderConfigCorrect{
+								Provider: addrs.NewBuiltInProvider("test"),
+							},
+						},
+					},
+					nil,
+				)
+				newState1 := builder.DataRead(desiredInst1, providerClient, nil)
+				newState2 := builder.DataRead(desiredInst2, providerClient, builder.Waiter(newState1))
+				registerUser(newState1)
+				registerUser(newState2)
+				builder.SetResourceInstanceFinalStateResult(instAddr1, newState1)
+				builder.SetResourceInstanceFinalStateResult(instAddr2, newState2)
+				return builder.Finish()
+			},
+			`
+				r[0] = OpenProvider(provider("terraform.io/builtin/test"), providerInstConfig(provider["terraform.io/builtin/test"]), await());
+				r[1] = DataRead(desired(data.test.example1), r[0], await());
+				r[2] = DataRead(desired(data.test.example2), r[0], await(r[1]));
+				r[3] = CloseProvider(r[0], await(r[1], r[2]));
+
+				data.test.example1 = r[1];
+				data.test.example2 = r[2];
+			`,
+		},
+
+		////////
+		// The remaining test cases are covering some weird cases just to
+		// make sure we can handle them without crashing or otherwise
+		// misbehaving. We don't need to go overboard here because this code
+		// only really needs to support graphs that OpenTofu's planning engine
+		// could reasonably generate.
+		////////
+		"empty": {
+			func(builder *Builder) *Graph {
+				return builder.Finish()
+			},
+			``,
+		},
+		"unused values discarded": {
+			// The graph marshaling is driven by what is refered to by the
+			// operations in the graph, and so anything that isn't actually
+			// used by an operation is irrelevant and discarded.
+			// (This test case is just here to remind us that this is the
+			// behavior. We don't actually rely on this behavior for
+			// correctness, because a value that isn't used in any operation
+			// is effectively ignored during execution anyway.)
+			func(b *Builder) *Graph {
+				b.ConstantValue(cty.True)
+				return b.Finish()
+			},
+			``,
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			inputGraph := test.InputGraph(NewBuilder())
+			marshaled := inputGraph.Marshal()
+			// The debug representation of the marshaled graph can be pretty
+			// verbose, so we'll print it only when we're reporting a failure
+			// so that a reader can look up any element indices that appear
+			// in the error messages.
+			showMarshaled := func() {
+				marshaledText := graphProtoDebugRepr(marshaled)
+				t.Log("graph marshaled as:\n" + marshaledText)
+			}
+			outputGraph, err := UnmarshalGraph(marshaled)
+			if err != nil {
+				// This test function only deals with valid cases, so we
+				// don't expect any errors here.
+				showMarshaled()
+				t.Fatalf("unexpected unmarshal error: %s", err)
+			}
+			gotRepr := trimLineSpaces(outputGraph.DebugRepr())
+			wantRepr := trimLineSpaces(test.WantGraph)
+			if diff := cmp.Diff(wantRepr, gotRepr); diff != "" {
+				showMarshaled()
+				t.Error("wrong output graph:\n" + diff)
+			}
+		})
+	}
+}
+
+// trimLineSpaces returns a modified version of the given string where each
+// individual line has its leading and trailing spaces removed, where
+// "spaces" is defined the same way as for [strings.TrimSpace].
+//
+// This is here just to make it easier to compare results from [Graph.DebugRepr]
+// with string constants written in test code, while still having those
+// string constants indented consistently with the surrounding code. Apply this
+// function to both the constant string and the [Graph.DebugRepr] result and
+// then compare the two e.g. using [cmp.Diff].
+func trimLineSpaces(input string) string {
+	var buf strings.Builder
+
+	// Since this function is tailored for Graph.DebugRepr in particular
+	// we'll use simplistic string cutting instead of all of the complexity
+	// of bufio.Scanner here, which also means we can minimize copying
+	// in conversions between string and []byte.
+	remain := input
+	for len(remain) != 0 {
+		line, extra, _ := strings.Cut(remain, "\n")
+		remain = extra
+		buf.WriteString(strings.TrimSpace(line))
+		buf.WriteByte('\n')
+	}
+
+	// We also ignore any leading and trailing spaces in the result, which
+	// could caused if there's an extra newline at the start or end of the
+	// string, as tends to happen when formatting a raw string to match the
+	// indentation of its surroundings.
+	return strings.TrimSpace(buf.String())
+}
+
+// graphProtoDebugRepr produces a human-oriented string representation of
+// a serialized execution graph for test debugging purposes only.
+func graphProtoDebugRepr(wire []byte) string {
+	var buf strings.Builder
+	var protoGraph execgraphproto.ExecutionGraph
+	err := proto.Unmarshal(wire, &protoGraph)
+	if err != nil {
+		panic(fmt.Sprintf("invalid protobuf serialization of %T: %s", &protoGraph, err))
+	}
+
+	// Various parts of the graph serialization involve indices into the
+	// same array of elements, so it's helpful to include the indices
+	// explicitly in the output.
+	for i, elem := range protoGraph.GetElements() {
+		asJSON := protojson.Format(elem)
+		fmt.Fprintf(&buf, "%d: %s\n", i, asJSON)
+	}
+
+	resourceInstResults := protoGraph.GetResourceInstanceResults()
+	if len(resourceInstResults) != 0 {
+		resourceInstResultsJSON, err := json.MarshalIndent(resourceInstResults, "", "  ")
+		if err != nil {
+			panic(fmt.Sprintf("can't marshal resource instance results: %s", err))
+		}
+		fmt.Fprintf(&buf, "resource instance results: %s\n", resourceInstResultsJSON)
+	}
+
+	return buf.String()
+}

internal/engine/internal/execgraph/graph_unmarshal.go

@@ -0,0 +1,350 @@
+// Copyright (c) The OpenTofu Authors
+// SPDX-License-Identifier: MPL-2.0
+// Copyright (c) 2023 HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package execgraph
+
+import (
+	"fmt"
+	"math"
+
+	"github.com/zclconf/go-cty/cty"
+	ctymsgpack "github.com/zclconf/go-cty/cty/msgpack"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/opentofu/opentofu/internal/addrs"
+	"github.com/opentofu/opentofu/internal/engine/internal/execgraph/execgraphproto"
+	"github.com/opentofu/opentofu/internal/lang/eval"
+	"github.com/opentofu/opentofu/internal/providers"
+	"github.com/opentofu/opentofu/internal/states"
+)
+
+// UnmarshalGraph takes some bytes previously returned by [Graph.Marshal] and
+// returns a graph that is functionally-equivalent to (but not necessarily
+// identical to) the original graph.
+//
+// Because this is working with data loaded from outside OpenTofu it returns
+// errors when encountering problems, but if it fails when unmarshaling an
+// unmodified result from [Graph.Marshal] then that represents a bug in either
+// this or that function: they should always be updated together so they are
+// implementing the same file format.
+func UnmarshalGraph(src []byte) (*Graph, error) {
+	var root execgraphproto.ExecutionGraph
+	err := proto.Unmarshal(src, &root)
+	if err != nil {
+		return nil, fmt.Errorf("invalid wire format: %w", err)
+	}
+
+	elems := root.GetElements()
+	// During decoding we'll track the typed result ref corresponding to
+	// each element from the serialized graph, which then allows us to
+	// make sure that operands are actually of the types they ought to be
+	// during our validation work.
+	results := make([]AnyResultRef, len(elems))
+	builder := NewBuilder()
+
+	for idx, elem := range elems {
+		switch reqType := elem.WhichRequest(); reqType {
+		case execgraphproto.Element_Operation_case:
+			resultRef, err := unmarshalOperationElem(elem.GetOperation(), results, builder)
+			if err != nil {
+				return nil, fmt.Errorf("invalid operation in element %d: %w", idx, err)
+			}
+			results[idx] = resultRef
+
+		case execgraphproto.Element_Waiter_case:
+			resultRef, err := unmarshalWaiterElem(elem.GetWaiter(), results, builder)
+			if err != nil {
+				return nil, fmt.Errorf("invalid waiter in element %d: %w", idx, err)
+			}
+			results[idx] = resultRef
+
+		case execgraphproto.Element_ConstantValue_case:
+			val, err := unmarshalConstantValueElem(elem.GetConstantValue())
+			if err != nil {
+				return nil, fmt.Errorf("invalid constant value in element %d: %w", idx, err)
+			}
+			results[idx] = builder.ConstantValue(val)
+
+		case execgraphproto.Element_ConstantProviderAddr_case:
+			addr, err := unmarshalConstantProviderAddr(elem.GetConstantProviderAddr())
+			if err != nil {
+				return nil, fmt.Errorf("invalid provider address in element %d: %w", idx, err)
+			}
+			results[idx] = builder.ConstantProviderAddr(addr)
+
+		case execgraphproto.Element_DesiredResourceInstance_case:
+			addr, err := unmarshalDesiredResourceInstance(elem.GetDesiredResourceInstance())
+			if err != nil {
+				return nil, fmt.Errorf("invalid desired resource instance address in element %d: %w", idx, err)
+			}
+			results[idx] = builder.DesiredResourceInstance(addr)
+
+		case execgraphproto.Element_ResourceInstancePriorState_case:
+			addr, err := unmarshalResourceInstancePriorState(elem.GetResourceInstancePriorState())
+			if err != nil {
+				return nil, fmt.Errorf("invalid prior state resource instance address in element %d: %w", idx, err)
+			}
+			results[idx] = builder.ResourceInstancePriorState(addr)
+
+		case execgraphproto.Element_ResourceInstanceDeposedObjectState_case:
+			objRef, err := unmarshalResourceInstanceDeposedObjectState(elem.GetResourceInstanceDeposedObjectState())
+			if err != nil {
+				return nil, fmt.Errorf("invalid deposed resource instance object reference in element %d: %w", idx, err)
+			}
+			results[idx] = builder.ResourceDeposedObjectState(objRef.ResourceInstance, objRef.DeposedKey)
+
+		case execgraphproto.Element_ProviderInstanceConfig_case:
+			addr, err := unmarshalProviderInstanceConfig(elem.GetProviderInstanceConfig())
+			if err != nil {
+				return nil, fmt.Errorf("invalid provider instance address in element %d: %w", idx, err)
+			}
+			results[idx] = builder.ProviderInstanceConfig(addr)
+
+		default:
+			// The above cases should cover all of the valid values of
+			// execgraphproto.case_Element_Request, so we should not get here
+			// for any serialized graph that was produced by this version
+			// of OpenTofu.
+			return nil, fmt.Errorf("unrecognized request type %#v for element %d", reqType, idx)
+		}
+	}
+
+	for instAddrStr, resultIdx := range root.GetResourceInstanceResults() {
+		instAddr, diags := addrs.ParseAbsResourceInstanceStr(instAddrStr)
+		if diags.HasErrors() {
+			return nil, fmt.Errorf("invalid resource instance address %q: %w", instAddrStr, diags.Err())
+		}
+		resultRef, err := unmarshalGetPrevResultOf[*states.ResourceInstanceObject](results, resultIdx)
+		if err != nil {
+			return nil, fmt.Errorf("invalid result element for %s: %w", instAddr, err)
+		}
+		builder.SetResourceInstanceFinalStateResult(instAddr, resultRef)
+	}
+
+	return builder.Finish(), nil
+}
+
+func unmarshalOperationElem(protoOp *execgraphproto.Operation, prevResults []AnyResultRef, builder *Builder) (AnyResultRef, error) {
+	switch c := protoOp.GetOpcode(); opCode(c) {
+	case opManagedFinalPlan:
+		return unmarshalOpManagedFinalPlan(protoOp.GetOperands(), prevResults, builder)
+	case opManagedApplyChanges:
+		return unmarshalOpManagedApplyChanges(protoOp.GetOperands(), prevResults, builder)
+	case opDataRead:
+		return unmarshalOpDataRead(protoOp.GetOperands(), prevResults, builder)
+	case opOpenProvider:
+		return unmarshalOpOpenProvider(protoOp.GetOperands(), prevResults, builder)
+	case opCloseProvider:
+		return unmarshalOpCloseProvider(protoOp.GetOperands(), prevResults, builder)
+
+		// TODO: All of the other opcodes
+	default:
+		// The above cases should cover all valid values of [opCode], so we
+		// should not get here unless the serialized graph was tampered
+		// with outside of OpenTofu.
+		return nil, fmt.Errorf("unrecognized opcode %d", c)
+	}
+}
+
+func unmarshalOpManagedFinalPlan(rawOperands []uint64, prevResults []AnyResultRef, builder *Builder) (AnyResultRef, error) {
+	if len(rawOperands) != 5 {
+		return nil, fmt.Errorf("wrong number of operands (%d) for opManagedFinalPlan", len(rawOperands))
+	}
+	desiredInst, err := unmarshalGetPrevResultOf[*eval.DesiredResourceInstance](prevResults, rawOperands[0])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opManagedFinalPlan desiredInst: %w", err)
+	}
+	priorState, err := unmarshalGetPrevResultOf[*states.ResourceInstanceObject](prevResults, rawOperands[1])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opManagedFinalPlan priorState: %w", err)
+	}
+	plannedVal, err := unmarshalGetPrevResultOf[cty.Value](prevResults, rawOperands[2])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opManagedFinalPlan plannedVal: %w", err)
+	}
+	providerClient, err := unmarshalGetPrevResultOf[providers.Configured](prevResults, rawOperands[3])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opManagedFinalPlan providerClient: %w", err)
+	}
+	waitFor, err := unmarshalGetPrevResultWaiter(prevResults, rawOperands[4])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opManagedFinalPlan waitFor: %w", err)
+	}
+	return builder.ManagedResourceObjectFinalPlan(desiredInst, priorState, plannedVal, providerClient, waitFor), nil
+}
+
+func unmarshalOpManagedApplyChanges(rawOperands []uint64, prevResults []AnyResultRef, builder *Builder) (AnyResultRef, error) {
+	if len(rawOperands) != 2 {
+		return nil, fmt.Errorf("wrong number of operands (%d) for opManagedApplyChanges", len(rawOperands))
+	}
+	finalPlan, err := unmarshalGetPrevResultOf[*ManagedResourceObjectFinalPlan](prevResults, rawOperands[0])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opManagedApplyChanges finalPlan: %w", err)
+	}
+	providerClient, err := unmarshalGetPrevResultOf[providers.Configured](prevResults, rawOperands[1])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opManagedApplyChanges providerClient: %w", err)
+	}
+	return builder.ApplyManagedResourceObjectChanges(finalPlan, providerClient), nil
+}
+
+func unmarshalOpDataRead(rawOperands []uint64, prevResults []AnyResultRef, builder *Builder) (AnyResultRef, error) {
+	if len(rawOperands) != 3 {
+		return nil, fmt.Errorf("wrong number of operands (%d) for opDataRead", len(rawOperands))
+	}
+	desiredInst, err := unmarshalGetPrevResultOf[*eval.DesiredResourceInstance](prevResults, rawOperands[0])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opDataRead desiredInst: %w", err)
+	}
+	providerClient, err := unmarshalGetPrevResultOf[providers.Configured](prevResults, rawOperands[1])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opDataRead providerClient: %w", err)
+	}
+	waitFor, err := unmarshalGetPrevResultWaiter(prevResults, rawOperands[2])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opDataRead waitFor: %w", err)
+	}
+	return builder.DataRead(desiredInst, providerClient, waitFor), nil
+}
+
+func unmarshalOpOpenProvider(rawOperands []uint64, prevResults []AnyResultRef, builder *Builder) (AnyResultRef, error) {
+	if len(rawOperands) != 3 {
+		return nil, fmt.Errorf("wrong number of operands (%d) for opOpenProvider", len(rawOperands))
+	}
+	providerAddr, err := unmarshalGetPrevResultOf[addrs.Provider](prevResults, rawOperands[0])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opOpenProvider providerAddr: %w", err)
+	}
+	config, err := unmarshalGetPrevResultOf[cty.Value](prevResults, rawOperands[1])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opOpenProvider config: %w", err)
+	}
+	waitFor, err := unmarshalGetPrevResultWaiter(prevResults, rawOperands[2])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opOpenProvider waitFor: %w", err)
+	}
+	return builder.OpenProviderClient(providerAddr, config, waitFor), nil
+}
+
+func unmarshalOpCloseProvider(rawOperands []uint64, prevResults []AnyResultRef, builder *Builder) (AnyResultRef, error) {
+	if len(rawOperands) != 2 {
+		return nil, fmt.Errorf("wrong number of operands (%d) for opCloseProvider", len(rawOperands))
+	}
+	client, err := unmarshalGetPrevResultOf[providers.Configured](prevResults, rawOperands[0])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opCloseProvider client: %w", err)
+	}
+	waitFor, err := unmarshalGetPrevResultWaiter(prevResults, rawOperands[1])
+	if err != nil {
+		return nil, fmt.Errorf("invalid opCloseProvider waitFor: %w", err)
+	}
+	return builder.CloseProviderClient(client, waitFor), nil
+}
+
+func unmarshalWaiterElem(protoWaiter *execgraphproto.Waiter, prevResults []AnyResultRef, builder *Builder) (AnyResultRef, error) {
+	waitForIdxs := protoWaiter.GetResults()
+	waitFor := make([]AnyResultRef, len(waitForIdxs))
+	for i, prevResultIdx := range waitForIdxs {
+		result := unmarshalGetPrevResult(prevResults, prevResultIdx)
+		if result == nil {
+			return nil, fmt.Errorf("refers to later element %d", prevResultIdx)
+		}
+		waitFor[i] = result
+	}
+	waiter := builder.Waiter(waitFor...)
+	return waiter, nil
+}
+
+func unmarshalConstantValueElem(src []byte) (cty.Value, error) {
+	v, err := ctymsgpack.Unmarshal(src, cty.DynamicPseudoType)
+	if err != nil {
+		return cty.NilVal, fmt.Errorf("invalid MessagePack encoding: %w", err)
+	}
+	return v, nil
+}
+
+func unmarshalConstantProviderAddr(addrStr string) (addrs.Provider, error) {
+	// This address parser returns diagnostics rather than just an error,
+	// which is inconvenient. :(
+	ret, diags := addrs.ParseProviderSourceString(addrStr)
+	return ret, diags.Err()
+}
+
+func unmarshalDesiredResourceInstance(addrStr string) (addrs.AbsResourceInstance, error) {
+	// This address parser returns diagnostics rather than just an error,
+	// which is inconvenient. :(
+	ret, diags := addrs.ParseAbsResourceInstanceStr(addrStr)
+	return ret, diags.Err()
+}
+
+func unmarshalResourceInstancePriorState(addrStr string) (addrs.AbsResourceInstance, error) {
+	// This address parser returns diagnostics rather than just an error,
+	// which is inconvenient. :(
+	ret, diags := addrs.ParseAbsResourceInstanceStr(addrStr)
+	return ret, diags.Err()
+}
+
+func unmarshalProviderInstanceConfig(addrStr string) (addrs.AbsProviderInstanceCorrect, error) {
+	// This address parser returns diagnostics rather than just an error,
+	// which is inconvenient. :(
+	//
+	// FIXME: AbsProviderInstanceCorrect doesn't currently have its own string
+	// parsing function, we'll borrow the one we introduced for the world where
+	// instance keys are tracked externally from the address. This means that
+	// we're limited to only the address forms that the old system could handle,
+	// including not supporting provider instances inside multi-instance modules,
+	// but that's okay for now.
+	var ret addrs.AbsProviderInstanceCorrect
+	configAddr, instKey, diags := addrs.ParseAbsProviderConfigInstanceStr(addrStr)
+	if diags.HasErrors() {
+		return ret, fmt.Errorf("invalid provider instance address: %w", diags.Err())
+	}
+	ret.Config = configAddr.Correct()
+	ret.Key = instKey
+	return ret, diags.Err()
+}
+
+func unmarshalResourceInstanceDeposedObjectState(protoRef *execgraphproto.DeposedResourceInstanceObject) (resourceInstanceStateRef, error) {
+	var ret resourceInstanceStateRef
+	instAddrStr := protoRef.GetInstanceAddr()
+	deposedKeyStr := protoRef.GetDeposedKey()
+
+	// This address parser returns diagnostics rather than just an error,
+	// which is inconvenient. :(
+	instAddr, diags := addrs.ParseAbsResourceInstanceStr(instAddrStr)
+	if diags.HasErrors() {
+		return ret, fmt.Errorf("invalid resource instance address: %w", diags.Err())
+	}
+	ret.ResourceInstance = instAddr
+	ret.DeposedKey = states.DeposedKey(deposedKeyStr)
+	return ret, nil
+}
+
+func unmarshalGetPrevResult(prevResults []AnyResultRef, resultIdx uint64) AnyResultRef {
+	if resultIdx > math.MaxInt || int(resultIdx) > len(prevResults) {
+		return nil
+	}
+	return prevResults[int(resultIdx)]
+}
+
+func unmarshalGetPrevResultOf[T any](prevResults []AnyResultRef, resultIdx uint64) (ResultRef[T], error) {
+	dynResult := unmarshalGetPrevResult(prevResults, resultIdx)
+	if dynResult == nil {
+		return nil, fmt.Errorf("refers to later element %d", resultIdx)
+	}
+	ret, ok := dynResult.(ResultRef[T])
+	if !ok {
+		return nil, fmt.Errorf("refers to %T, but need %T", dynResult, ret)
+	}
+	return ret, nil
+}
+
+func unmarshalGetPrevResultWaiter(prevResults []AnyResultRef, resultIdx uint64) (AnyResultRef, error) {
+	ret := unmarshalGetPrevResult(prevResults, resultIdx)
+	if ret == nil {
+		return nil, fmt.Errorf("refers to later element %d", resultIdx)
+	}
+	return ret, nil
+}

tools/protobuf-compile/protobuf-compile.go

@@ -60,6 +60,11 @@ var protocSteps = []protocStep{
 		"internal/plans/internal/planproto",
 		[]string{"--go_out=.", "--go_opt=paths=source_relative", "planfile.proto"},
 	},
+	{
+		"opentofu_execgraph (execution graph serialization)",
+		"internal/engine/internal/execgraph/execgraphproto",
+		[]string{"--go_out=.", "--go_opt=paths=source_relative", "execgraph.proto"},
+	},
 }
 
 func main() {