Repository: kurtosis-tech/kurtosis
Pull request #1280 (2023-09-11 09:33:02)

feat: The current enclave plan is now persisted to the enclave DB every times the execution finishes
## Description:
This makes the APIC resilient to restarts w.r.t. idempotent runs.

## Is this change user facing?
NO
<!-- If yes, please add the "user facing" label to the PR -->
<!-- If yes, don't forget to include docs changes where relevant -->

## References (if applicable):
<!-- Add relevant Github Issues, Discord threads, or other helpful information. -->



Patch:
diff --git a/core/server/api_container/main.go b/core/server/api_container/main.go
index fc26f6ad5..ea378246e 100644
--- a/core/server/api_container/main.go
+++ b/core/server/api_container/main.go
@@ -20,6 +20,7 @@ import (
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_types"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/runtime_value_store"
 	"github.com/kurtosis-tech/kurtosis/core/server/commons/enclave_data_directory"
@@ -170,11 +171,17 @@ func runMain() error {
 		return stacktrace.Propagate(err, "An error occurred creating the runtime value store")
 	}
 
+	// Load the current enclave plan, in case the enclave is being restarted
+	enclavePlan, err := enclave_plan_persistence.Load(enclaveDb)
+	if err != nil {
+		return stacktrace.Propagate(err, "An error occurred loading stored enclave plan")
+	}
+
 	// TODO: Consolidate Interpreter, Validator and Executor into a single interface
 	startosisRunner := startosis_engine.NewStartosisRunner(
-		startosis_engine.NewStartosisInterpreter(serviceNetwork, gitPackageContentProvider, runtimeValueStore, serverArgs.EnclaveEnvVars),
+		startosis_engine.NewStartosisInterpreter(serviceNetwork, gitPackageContentProvider, runtimeValueStore, starlarkValueSerde, serverArgs.EnclaveEnvVars),
 		startosis_engine.NewStartosisValidator(&kurtosisBackend, serviceNetwork, filesArtifactStore),
-		startosis_engine.NewStartosisExecutor(runtimeValueStore))
+		startosis_engine.NewStartosisExecutor(starlarkValueSerde, runtimeValueStore, enclavePlan, enclaveDb))
 
 	//Creation of ApiContainerService
 	apiContainerService, err := server.NewApiContainerService(
diff --git a/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan.go b/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan.go
new file mode 100644
index 000000000..f8b302417
--- /dev/null
+++ b/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan.go
@@ -0,0 +1,94 @@
+package enclave_plan_persistence
+
+import (
+	"encoding/json"
+	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/database_accessors/enclave_db"
+	"github.com/kurtosis-tech/stacktrace"
+	"go.etcd.io/bbolt"
+)
+
+const (
+	enclavePlanBucketName = "EnclavePlan"
+	// We persist only one enclave plan per enclave, so we can use a constant key here
+	enclavePlanConstantKey = "EnclavePlan"
+)
+
+type EnclavePlan struct {
+	EnclavePlanInstructions []*EnclavePlanInstruction `json:"enclavePlanInstructions"`
+}
+
+func NewEnclavePlan() *EnclavePlan {
+	return &EnclavePlan{
+		EnclavePlanInstructions: []*EnclavePlanInstruction{},
+	}
+}
+
+func (enclavePlan *EnclavePlan) PartialDeepClone(indexOfFirstInstructionToNotClone int) *EnclavePlan {
+	newPlan := &EnclavePlan{
+		EnclavePlanInstructions: []*EnclavePlanInstruction{},
+	}
+	for idx, instruction := range enclavePlan.EnclavePlanInstructions {
+		if idx >= indexOfFirstInstructionToNotClone {
+			return newPlan
+		}
+		newPlan.AppendInstruction(instruction.Clone())
+	}
+	return newPlan
+}
+
+func Load(enclaveDb *enclave_db.EnclaveDB) (*EnclavePlan, error) {
+	var persistedEnclavePlanSerializedMaybe []byte
+	err := enclaveDb.View(func(tx *bbolt.Tx) error {
+		enclavePlanBucket := tx.Bucket([]byte(enclavePlanBucketName))
+		if enclavePlanBucket == nil {
+			return nil
+		}
+		persistedEnclavePlanSerializedMaybe = enclavePlanBucket.Get([]byte(enclavePlanConstantKey))
+		return nil
+	})
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "An error occurred reading enclave plan value from enclave database")
+	}
+	if persistedEnclavePlanSerializedMaybe == nil {
+		// No enclave plan persisted for this enclave just yet, return an empty plan
+		return NewEnclavePlan(), nil
+	}
+	persistedEnclavePlan := new(EnclavePlan)
+	if err := json.Unmarshal(persistedEnclavePlanSerializedMaybe, persistedEnclavePlan); err != nil {
+		return nil, stacktrace.Propagate(err, "An error occurred deserializing enclave plan from enclave database")
+	}
+	return persistedEnclavePlan, nil
+}
+
+func (enclavePlan *EnclavePlan) Persist(enclaveDb *enclave_db.EnclaveDB) error {
+	serializedEnclavePlanBytes, err := json.Marshal(enclavePlan)
+	if err != nil {
+		return stacktrace.Propagate(err, "An error occurred serializing enclave plan to enclave database")
+	}
+	err = enclaveDb.Update(func(tx *bbolt.Tx) error {
+		enclavePlanBucket, err := tx.CreateBucketIfNotExists([]byte(enclavePlanBucketName))
+		if err != nil {
+			return stacktrace.Propagate(err, "An error occurred getting or creating enclave plan bucket into Bbolt")
+		}
+		if err := enclavePlanBucket.Put([]byte(enclavePlanConstantKey), serializedEnclavePlanBytes); err != nil {
+			return stacktrace.Propagate(err, "An error occurred persisting enclave plan to to Bbolt database")
+		}
+		return nil
+	})
+	if err != nil {
+		return stacktrace.Propagate(err, "An error occurred persisting enclave plan to enclave database")
+	}
+	return nil
+}
+
+func (enclavePlan *EnclavePlan) AppendInstruction(instruction *EnclavePlanInstruction) {
+	enclavePlan.EnclavePlanInstructions = append(enclavePlan.EnclavePlanInstructions, instruction)
+}
+
+func (enclavePlan *EnclavePlan) GeneratePlan() []*EnclavePlanInstruction {
+	return enclavePlan.EnclavePlanInstructions
+}
+
+func (enclavePlan *EnclavePlan) Size() int {
+	return len(enclavePlan.EnclavePlanInstructions)
+}
diff --git a/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan_instruction.go b/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan_instruction.go
new file mode 100644
index 000000000..b0cfa86a1
--- /dev/null
+++ b/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan_instruction.go
@@ -0,0 +1,74 @@
+package enclave_plan_persistence
+
+import (
+	"bytes"
+	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
+)
+
+type EnclavePlanInstruction struct {
+	Uuid string `json:"uuid"`
+
+	Type string `json:"type"`
+
+	StarlarkCode string `json:"starlarkCode"`
+
+	ReturnedValue string `json:"returnedValue"`
+
+	ServiceNames []string `json:"serviceNames"`
+
+	// mapping between files artifact name and files artifact MD5
+	FilesArtifacts map[string][]byte `json:"filesArtifacts"` // FSA byte arrays are automatically serialized as base64 encoded strings
+}
+
+// HasOnlyServiceName is a convenience function that returns true if the enclave plan instruction has only
+// one service and its name is equal to the parameter of this function
+func (enclavePlanInstruction *EnclavePlanInstruction) HasOnlyServiceName(serviceName service.ServiceName) bool {
+	if len(enclavePlanInstruction.ServiceNames) != 1 {
+		return false
+	}
+	return enclavePlanInstruction.ServiceNames[0] == string(serviceName)
+}
+
+// HasOnlyFilesArtifactName is a convenience function that returns true if the enclave plan instruction has only
+// one files artifact and its name is equal to the parameter of this function
+func (enclavePlanInstruction *EnclavePlanInstruction) HasOnlyFilesArtifactName(filesArtifactName string) bool {
+	if len(enclavePlanInstruction.FilesArtifacts) != 1 {
+		return false
+	}
+	_, found := enclavePlanInstruction.FilesArtifacts[filesArtifactName]
+	return found
+}
+
+// HasOnlyFilesArtifactMd5 is a convenience function that returns true if the enclave plan instruction has only
+// one files artifact and its MD5 is equal to the parameter of this function
+func (enclavePlanInstruction *EnclavePlanInstruction) HasOnlyFilesArtifactMd5(filesArtifactMd5 []byte) bool {
+	if len(enclavePlanInstruction.FilesArtifacts) != 1 {
+		return false
+	}
+	for _, filesArtifactMd5FromInstruction := range enclavePlanInstruction.FilesArtifacts {
+		if bytes.Equal(filesArtifactMd5, filesArtifactMd5FromInstruction) {
+			return true
+		}
+	}
+	return false
+}
+
+func (enclavePlanInstruction *EnclavePlanInstruction) Clone() *EnclavePlanInstruction {
+	clonedServiceNames := make([]string, len(enclavePlanInstruction.ServiceNames))
+	copy(enclavePlanInstruction.ServiceNames, clonedServiceNames)
+
+	clonedFilesArtifacts := make(map[string][]byte, len(enclavePlanInstruction.FilesArtifacts))
+	for filesArtifactName, filesArtifactMd5 := range enclavePlanInstruction.FilesArtifacts {
+		clonedFilesArtifactMd5 := make([]byte, len(filesArtifactMd5))
+		copy(filesArtifactMd5, clonedFilesArtifactMd5)
+		clonedFilesArtifacts[filesArtifactName] = clonedFilesArtifactMd5
+	}
+	return &EnclavePlanInstruction{
+		Uuid:           enclavePlanInstruction.Uuid,
+		Type:           enclavePlanInstruction.Type,
+		StarlarkCode:   enclavePlanInstruction.StarlarkCode,
+		ReturnedValue:  enclavePlanInstruction.ReturnedValue,
+		ServiceNames:   clonedServiceNames,
+		FilesArtifacts: clonedFilesArtifacts,
+	}
+}
diff --git a/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan_instruction_builder.go b/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan_instruction_builder.go
new file mode 100644
index 000000000..737757fd8
--- /dev/null
+++ b/core/server/api_container/server/startosis_engine/enclave_plan_persistence/enclave_plan_instruction_builder.go
@@ -0,0 +1,75 @@
+package enclave_plan_persistence
+
+import (
+	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
+	"github.com/kurtosis-tech/stacktrace"
+)
+
+type EnclavePlanInstructionBuilder struct {
+	uuid string
+
+	instructionType string
+
+	starlarkCode string
+
+	returnedValue string
+
+	serviceNames []string
+
+	filesArtifacts map[string][]byte
+}
+
+func NewEnclavePlanInstructionBuilder() *EnclavePlanInstructionBuilder {
+	return &EnclavePlanInstructionBuilder{
+		uuid:            "",
+		instructionType: "",
+		starlarkCode:    "",
+		returnedValue:   "",
+		serviceNames:    []string{},
+		filesArtifacts:  map[string][]byte{},
+	}
+}
+
+func (builder *EnclavePlanInstructionBuilder) SetUuid(uuid string) *EnclavePlanInstructionBuilder {
+	builder.uuid = uuid
+	return builder
+}
+
+func (builder *EnclavePlanInstructionBuilder) SetType(instructionType string) *EnclavePlanInstructionBuilder {
+	builder.instructionType = instructionType
+	return builder
+}
+
+func (builder *EnclavePlanInstructionBuilder) SetStarlarkCode(starlarkCode string) *EnclavePlanInstructionBuilder {
+	builder.starlarkCode = starlarkCode
+	return builder
+}
+
+func (builder *EnclavePlanInstructionBuilder) SetReturnedValue(serializedReturnedValue string) *EnclavePlanInstructionBuilder {
+	builder.returnedValue = serializedReturnedValue
+	return builder
+}
+
+func (builder *EnclavePlanInstructionBuilder) AddServiceName(serviceName service.ServiceName) *EnclavePlanInstructionBuilder {
+	builder.serviceNames = append(builder.serviceNames, string(serviceName))
+	return builder
+}
+
+func (builder *EnclavePlanInstructionBuilder) AddFilesArtifact(filesArtifactName string, filesArtifactMd5 []byte) *EnclavePlanInstructionBuilder {
+	builder.filesArtifacts[filesArtifactName] = filesArtifactMd5
+	return builder
+}
+
+func (builder *EnclavePlanInstructionBuilder) Build() (*EnclavePlanInstruction, error) {
+	if builder.uuid == "" || builder.instructionType == "" || builder.starlarkCode == "" || builder.returnedValue == "" {
+		return nil, stacktrace.NewError("Some required attributes aren't set on this builder")
+	}
+	return &EnclavePlanInstruction{
+		Uuid:           builder.uuid,
+		Type:           builder.instructionType,
+		StarlarkCode:   builder.starlarkCode,
+		ReturnedValue:  builder.returnedValue,
+		ServiceNames:   builder.serviceNames,
+		FilesArtifacts: builder.filesArtifacts,
+	}, nil
+}
diff --git a/core/server/api_container/server/startosis_engine/instructions_plan/instructions_plan.go b/core/server/api_container/server/startosis_engine/instructions_plan/instructions_plan.go
index 87c5a8285..caabaf9d9 100644
--- a/core/server/api_container/server/startosis_engine/instructions_plan/instructions_plan.go
+++ b/core/server/api_container/server/startosis_engine/instructions_plan/instructions_plan.go
@@ -30,21 +30,6 @@ func NewInstructionsPlan() *InstructionsPlan {
 	}
 }
 
-func (plan *InstructionsPlan) PartialClone(indexOfFirstInstructionToNotClone int) (*InstructionsPlan, error) {
-	newPlan := NewInstructionsPlan()
-	for idx, instructionUuid := range plan.instructionsSequence {
-		if idx >= indexOfFirstInstructionToNotClone {
-			return newPlan, nil
-		}
-		instructionToTransfer, found := plan.scheduledInstructionsIndex[instructionUuid]
-		if !found {
-			return nil, stacktrace.NewError("Instruction with UUID '%s' was part of the instruction sequence but could not be found in the instruction index. This is a Kurtosis internal issue", instructionUuid)
-		}
-		newPlan.AddScheduledInstruction(instructionToTransfer)
-	}
-	return newPlan, nil
-}
-
 func (plan *InstructionsPlan) SetIndexOfFirstInstruction(indexOfFirstInstruction int) {
 	plan.indexOfFirstInstruction = indexOfFirstInstruction
 }
diff --git a/core/server/api_container/server/startosis_engine/instructions_plan/resolver/instructions_plan_mask.go b/core/server/api_container/server/startosis_engine/instructions_plan/resolver/instructions_plan_mask.go
index 16714493e..f66050d70 100644
--- a/core/server/api_container/server/startosis_engine/instructions_plan/resolver/instructions_plan_mask.go
+++ b/core/server/api_container/server/startosis_engine/instructions_plan/resolver/instructions_plan_mask.go
@@ -1,38 +1,40 @@
 package resolver
 
-import "github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
+import (
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
+)
 
 type InstructionsPlanMask struct {
-	readIdx               int
-	scheduledInstructions []*instructions_plan.ScheduledInstruction
-	isValid               bool
+	readIdx                 int
+	enclavePlanInstructions []*enclave_plan_persistence.EnclavePlanInstruction
+	isValid                 bool
 }
 
 func NewInstructionsPlanMask(size int) *InstructionsPlanMask {
 	return &InstructionsPlanMask{
-		readIdx:               0,
-		scheduledInstructions: make([]*instructions_plan.ScheduledInstruction, size),
-		isValid:               true, // the mask is considered valid until it's proven to be invalid
+		readIdx:                 0,
+		enclavePlanInstructions: make([]*enclave_plan_persistence.EnclavePlanInstruction, size),
+		isValid:                 true, // the mask is considered valid until it's proven to be invalid
 	}
 }
 
-func (mask *InstructionsPlanMask) InsertAt(idx int, instruction *instructions_plan.ScheduledInstruction) {
-	mask.scheduledInstructions[idx] = instruction
+func (mask *InstructionsPlanMask) InsertAt(idx int, instruction *enclave_plan_persistence.EnclavePlanInstruction) {
+	mask.enclavePlanInstructions[idx] = instruction
 }
 
 func (mask *InstructionsPlanMask) HasNext() bool {
-	return mask.readIdx < len(mask.scheduledInstructions)
+	return mask.readIdx < len(mask.enclavePlanInstructions)
 }
 
-func (mask *InstructionsPlanMask) Next() (int, *instructions_plan.ScheduledInstruction) {
+func (mask *InstructionsPlanMask) Next() (int, *enclave_plan_persistence.EnclavePlanInstruction) {
 	instructionIdx := mask.readIdx
-	scheduledInstruction := mask.scheduledInstructions[instructionIdx]
+	scheduledInstruction := mask.enclavePlanInstructions[instructionIdx]
 	mask.readIdx += 1
 	return instructionIdx, scheduledInstruction
 }
 
 func (mask *InstructionsPlanMask) Size() int {
-	return len(mask.scheduledInstructions)
+	return len(mask.enclavePlanInstructions)
 }
 
 func (mask *InstructionsPlanMask) MarkAsInvalid() {
diff --git a/core/server/api_container/server/startosis_engine/instructions_plan/scheduled_instruction.go b/core/server/api_container/server/startosis_engine/instructions_plan/scheduled_instruction.go
index 149da462d..19d83da87 100644
--- a/core/server/api_container/server/startosis_engine/instructions_plan/scheduled_instruction.go
+++ b/core/server/api_container/server/startosis_engine/instructions_plan/scheduled_instruction.go
@@ -31,6 +31,10 @@ func NewScheduledInstruction(uuid ScheduledInstructionUuid, kurtosisInstruction
 	}
 }
 
+func (instruction *ScheduledInstruction) GetUuid() ScheduledInstructionUuid {
+	return instruction.uuid
+}
+
 func (instruction *ScheduledInstruction) GetInstruction() kurtosis_instruction.KurtosisInstruction {
 	return instruction.kurtosisInstruction
 }
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_service.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_service.go
index 8683dea56..6cbf13ecd 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_service.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_service.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -159,20 +160,20 @@ func (builtin *AddServiceCapabilities) Execute(ctx context.Context, _ *builtin_a
 	return instructionResult, nil
 }
 
-func (builtin *AddServiceCapabilities) TryResolveWith(instructionsAreEqual bool, other kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *AddServiceCapabilities) TryResolveWith(instructionsAreEqual bool, other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	// if other instruction is nil or other instruction is not an add_service instruction, status is unknown
 	if other == nil {
 		enclaveComponents.AddService(builtin.serviceName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
-	otherAddServiceCapabilities, ok := other.(*AddServiceCapabilities)
-	if !ok {
+
+	if other.Type != AddServiceBuiltinName {
 		enclaveComponents.AddService(builtin.serviceName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
 
 	// if service names don't match, status is unknown, instructions can't be resolved together
-	if otherAddServiceCapabilities.serviceName != builtin.serviceName {
+	if !other.HasOnlyServiceName(builtin.serviceName) {
 		enclaveComponents.AddService(builtin.serviceName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
@@ -200,6 +201,14 @@ func (builtin *AddServiceCapabilities) TryResolveWith(instructionsAreEqual bool,
 	return enclave_structure.InstructionIsEqual
 }
 
+func (builtin *AddServiceCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(
+		AddServiceBuiltinName,
+	).AddServiceName(
+		builtin.serviceName,
+	)
+}
+
 func validateAndConvertConfigAndReadyCondition(
 	serviceNetwork service_network.ServiceNetwork,
 	rawConfig starlark.Value,
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_services.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_services.go
index 710d3633c..0dee620cf 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_services.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/add_service/add_services.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -203,7 +204,7 @@ func (builtin *AddServicesCapabilities) Execute(ctx context.Context, _ *builtin_
 	return instructionResult.String(), nil
 }
 
-func (builtin *AddServicesCapabilities) TryResolveWith(instructionsAreEqual bool, other kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *AddServicesCapabilities) TryResolveWith(instructionsAreEqual bool, other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual {
 		for serviceName := range builtin.serviceConfigs {
 			enclaveComponents.AddService(serviceName, enclave_structure.ComponentWasLeftIntact)
@@ -217,8 +218,8 @@ func (builtin *AddServicesCapabilities) TryResolveWith(instructionsAreEqual bool
 		}
 		return enclave_structure.InstructionIsUnknown
 	}
-	otherAddServicesCapabilities, ok := other.(*AddServicesCapabilities)
-	if !ok {
+
+	if other.Type != AddServicesBuiltinName {
 		for serviceName := range builtin.serviceConfigs {
 			enclaveComponents.AddService(serviceName, enclave_structure.ComponentIsNew)
 		}
@@ -227,10 +228,10 @@ func (builtin *AddServicesCapabilities) TryResolveWith(instructionsAreEqual bool
 
 	// The instruction can be re-run only if the set of added services is a superset of what was added by the
 	// instruction it's being compared to, so we check that first
-	atLeastOneFilehasBeenUpdated := false
+	atLeastOneFileHasBeenUpdated := false
 	previouslyAddedService := map[service.ServiceName]bool{}
-	for serviceName := range otherAddServicesCapabilities.serviceConfigs {
-		previouslyAddedService[serviceName] = false
+	for _, serviceName := range other.ServiceNames {
+		previouslyAddedService[service.ServiceName(serviceName)] = false
 	}
 	for serviceName, serviceConfig := range builtin.serviceConfigs {
 		if _, found := previouslyAddedService[serviceName]; found {
@@ -242,7 +243,7 @@ func (builtin *AddServicesCapabilities) TryResolveWith(instructionsAreEqual bool
 		if filesArtifactsExpansion != nil {
 			for _, filesArtifactName := range filesArtifactsExpansion.ServiceDirpathsToArtifactIdentifiers {
 				if enclaveComponents.HasFilesArtifactBeenUpdated(filesArtifactName) {
-					atLeastOneFilehasBeenUpdated = true
+					atLeastOneFileHasBeenUpdated = true
 				}
 			}
 		}
@@ -257,7 +258,7 @@ func (builtin *AddServicesCapabilities) TryResolveWith(instructionsAreEqual bool
 		}
 	}
 
-	if !instructionsAreEqual || atLeastOneFilehasBeenUpdated {
+	if !instructionsAreEqual || atLeastOneFileHasBeenUpdated {
 		for serviceName := range builtin.serviceConfigs {
 			if _, found := previouslyAddedService[serviceName]; found {
 				enclaveComponents.AddService(serviceName, enclave_structure.ComponentIsUpdated)
@@ -275,6 +276,13 @@ func (builtin *AddServicesCapabilities) TryResolveWith(instructionsAreEqual bool
 	return enclave_structure.InstructionIsEqual
 }
 
+func (builtin *AddServicesCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(AddServicesBuiltinName)
+	for serviceName := range builtin.serviceConfigs {
+		builder.AddServiceName(serviceName)
+	}
+}
+
 func (builtin *AddServicesCapabilities) removeAllStartedServices(
 	ctx context.Context,
 	startedServices map[service.ServiceName]*service.Service,
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/assert/assert.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/assert/assert.go
index da0c954b0..149e0f258 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/assert/assert.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/assert/assert.go
@@ -3,6 +3,7 @@ package assert
 import (
 	"context"
 	"fmt"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers/magic_string_helper"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
@@ -143,13 +144,17 @@ func (builtin *AssertCapabilities) Execute(_ context.Context, _ *builtin_argumen
 	return instructionResult, nil
 }
 
-func (builtin *AssertCapabilities) TryResolveWith(instructionsAreEqual bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *AssertCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual {
 		return enclave_structure.InstructionIsEqual
 	}
 	return enclave_structure.InstructionIsUnknown
 }
 
+func (builtin *AssertCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(AssertBuiltinName)
+}
+
 // Assert verifies whether the currentValue matches the targetValue w.r.t. the assertion operator
 // TODO: This and ValidateAssertionToken below are used by both assert and wait. Refactor it to a better place
 func Assert(currentValue starlark.Comparable, assertion string, targetValue starlark.Comparable) error {
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/exec/exec.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/exec/exec.go
index 67fc13fd4..7f95de1e0 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/exec/exec.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/exec/exec.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -175,7 +176,7 @@ func (builtin *ExecCapabilities) Execute(ctx context.Context, _ *builtin_argumen
 	return instructionResult, err
 }
 
-func (builtin *ExecCapabilities) TryResolveWith(instructionsAreEqual bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *ExecCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual && enclaveComponents.HasServiceBeenUpdated(builtin.serviceName) {
 		return enclave_structure.InstructionIsUpdate
 	} else if instructionsAreEqual {
@@ -184,6 +185,10 @@ func (builtin *ExecCapabilities) TryResolveWith(instructionsAreEqual bool, _ kur
 	return enclave_structure.InstructionIsUnknown
 }
 
+func (builtin *ExecCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(ExecBuiltinName)
+}
+
 func (builtin *ExecCapabilities) isAcceptableCode(recipeResult map[string]starlark.Comparable) bool {
 	isAcceptableCode := false
 	for _, acceptableCode := range builtin.acceptableCodes {
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_instruction.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_instruction.go
index fc71494d8..89d809662 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_instruction.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_instruction.go
@@ -3,19 +3,10 @@ package kurtosis_instruction
 import (
 	"context"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/kurtosis_core_rpc_api_bindings"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_validator"
-	"go.starlark.net/starlark"
-)
-
-const (
-	Representative    = true
-	NotRepresentative = false
-)
-
-var (
-	NoArgs []starlark.Value
 )
 
 type KurtosisInstruction interface {
@@ -34,5 +25,12 @@ type KurtosisInstruction interface {
 	ValidateAndUpdateEnvironment(environment *startosis_validator.ValidatorEnvironment) error
 
 	// TryResolveWith assesses whether the instruction can be resolved with the one passed as an argument.
-	TryResolveWith(other KurtosisInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus
+	TryResolveWith(other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus
+
+	// GetPersistableAttributes returns an EnclavePlanInstructionBuilder object which contains the persistable attributes
+	// for this instruction. Persistable attributes are what will be written to the enclave database so that even
+	// if the APIC is restarted, idempotent runs will continue to work.
+	// It returns a builder and not the built object b/c the caller of this method might want to set some attributes
+	// itself. In the current case, this is called in the executor, and it sets the UUID and the returned value.
+	GetPersistableAttributes() *enclave_plan_persistence.EnclavePlanInstructionBuilder
 }
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_print/kurtosis_print.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_print/kurtosis_print.go
index dd3938341..ee70b3db9 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_print/kurtosis_print.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_print/kurtosis_print.go
@@ -3,6 +3,7 @@ package kurtosis_print
 import (
 	"context"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers/magic_string_helper"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
@@ -88,9 +89,13 @@ func (builtin *PrintCapabilities) Execute(_ context.Context, _ *builtin_argument
 	return maybeSerializedArgsWithRuntimeValue, nil
 }
 
-func (builtin *PrintCapabilities) TryResolveWith(instructionsAreEqual bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *PrintCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual {
 		return enclave_structure.InstructionIsEqual
 	}
 	return enclave_structure.InstructionIsUnknown
 }
+
+func (builtin *PrintCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(PrintBuiltinName)
+}
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/mock_instruction/mock_kurtosis_instruction.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/mock_instruction/mock_kurtosis_instruction.go
index 68e9ff119..580183615 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/mock_instruction/mock_kurtosis_instruction.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/mock_instruction/mock_kurtosis_instruction.go
@@ -1,14 +1,15 @@
-// Code generated by mockery v2.31.1. DO NOT EDIT.
+// Code generated by mockery v2.22.1. DO NOT EDIT.
 
 package mock_instruction
 
 import (
 	context "context"
-	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction"
 
-	kurtosis_core_rpc_api_bindings "github.com/kurtosis-tech/kurtosis/api/golang/core/kurtosis_core_rpc_api_bindings"
+	enclave_plan_persistence "github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	enclave_structure "github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 
+	kurtosis_core_rpc_api_bindings "github.com/kurtosis-tech/kurtosis/api/golang/core/kurtosis_core_rpc_api_bindings"
+
 	kurtosis_starlark_framework "github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 
 	mock "github.com/stretchr/testify/mock"
@@ -127,6 +128,49 @@ func (_c *MockKurtosisInstruction_GetCanonicalInstruction_Call) RunAndReturn(run
 	return _c
 }
 
+// GetPersistableAttributes provides a mock function with given fields:
+func (_m *MockKurtosisInstruction) GetPersistableAttributes() *enclave_plan_persistence.EnclavePlanInstructionBuilder {
+	ret := _m.Called()
+
+	var r0 *enclave_plan_persistence.EnclavePlanInstructionBuilder
+	if rf, ok := ret.Get(0).(func() *enclave_plan_persistence.EnclavePlanInstructionBuilder); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*enclave_plan_persistence.EnclavePlanInstructionBuilder)
+		}
+	}
+
+	return r0
+}
+
+// MockKurtosisInstruction_GetPersistableAttributes_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetPersistableAttributes'
+type MockKurtosisInstruction_GetPersistableAttributes_Call struct {
+	*mock.Call
+}
+
+// GetPersistableAttributes is a helper method to define mock.On call
+func (_e *MockKurtosisInstruction_Expecter) GetPersistableAttributes() *MockKurtosisInstruction_GetPersistableAttributes_Call {
+	return &MockKurtosisInstruction_GetPersistableAttributes_Call{Call: _e.mock.On("GetPersistableAttributes")}
+}
+
+func (_c *MockKurtosisInstruction_GetPersistableAttributes_Call) Run(run func()) *MockKurtosisInstruction_GetPersistableAttributes_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run()
+	})
+	return _c
+}
+
+func (_c *MockKurtosisInstruction_GetPersistableAttributes_Call) Return(_a0 *enclave_plan_persistence.EnclavePlanInstructionBuilder) *MockKurtosisInstruction_GetPersistableAttributes_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MockKurtosisInstruction_GetPersistableAttributes_Call) RunAndReturn(run func() *enclave_plan_persistence.EnclavePlanInstructionBuilder) *MockKurtosisInstruction_GetPersistableAttributes_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // GetPositionInOriginalScript provides a mock function with given fields:
 func (_m *MockKurtosisInstruction) GetPositionInOriginalScript() *kurtosis_starlark_framework.KurtosisBuiltinPosition {
 	ret := _m.Called()
@@ -212,11 +256,11 @@ func (_c *MockKurtosisInstruction_String_Call) RunAndReturn(run func() string) *
 }
 
 // TryResolveWith provides a mock function with given fields: other, enclaveComponents
-func (_m *MockKurtosisInstruction) TryResolveWith(other kurtosis_instruction.KurtosisInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (_m *MockKurtosisInstruction) TryResolveWith(other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	ret := _m.Called(other, enclaveComponents)
 
 	var r0 enclave_structure.InstructionResolutionStatus
-	if rf, ok := ret.Get(0).(func(kurtosis_instruction.KurtosisInstruction, *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus); ok {
+	if rf, ok := ret.Get(0).(func(*enclave_plan_persistence.EnclavePlanInstruction, *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus); ok {
 		r0 = rf(other, enclaveComponents)
 	} else {
 		r0 = ret.Get(0).(enclave_structure.InstructionResolutionStatus)
@@ -231,15 +275,15 @@ type MockKurtosisInstruction_TryResolveWith_Call struct {
 }
 
 // TryResolveWith is a helper method to define mock.On call
-//   - other KurtosisInstruction
+//   - other *enclave_plan_persistence.EnclavePlanInstruction
 //   - enclaveComponents *enclave_structure.EnclaveComponents
 func (_e *MockKurtosisInstruction_Expecter) TryResolveWith(other interface{}, enclaveComponents interface{}) *MockKurtosisInstruction_TryResolveWith_Call {
 	return &MockKurtosisInstruction_TryResolveWith_Call{Call: _e.mock.On("TryResolveWith", other, enclaveComponents)}
 }
 
-func (_c *MockKurtosisInstruction_TryResolveWith_Call) Run(run func(other kurtosis_instruction.KurtosisInstruction, enclaveComponents *enclave_structure.EnclaveComponents)) *MockKurtosisInstruction_TryResolveWith_Call {
+func (_c *MockKurtosisInstruction_TryResolveWith_Call) Run(run func(other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents)) *MockKurtosisInstruction_TryResolveWith_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(kurtosis_instruction.KurtosisInstruction), args[1].(*enclave_structure.EnclaveComponents))
+		run(args[0].(*enclave_plan_persistence.EnclavePlanInstruction), args[1].(*enclave_structure.EnclaveComponents))
 	})
 	return _c
 }
@@ -249,7 +293,7 @@ func (_c *MockKurtosisInstruction_TryResolveWith_Call) Return(_a0 enclave_struct
 	return _c
 }
 
-func (_c *MockKurtosisInstruction_TryResolveWith_Call) RunAndReturn(run func(kurtosis_instruction.KurtosisInstruction, *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus) *MockKurtosisInstruction_TryResolveWith_Call {
+func (_c *MockKurtosisInstruction_TryResolveWith_Call) RunAndReturn(run func(*enclave_plan_persistence.EnclavePlanInstruction, *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus) *MockKurtosisInstruction_TryResolveWith_Call {
 	_c.Call.Return(run)
 	return _c
 }
@@ -296,12 +340,13 @@ func (_c *MockKurtosisInstruction_ValidateAndUpdateEnvironment_Call) RunAndRetur
 	return _c
 }
 
-// NewMockKurtosisInstruction creates a new instance of MockKurtosisInstruction. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
-// The first argument is typically a *testing.T value.
-func NewMockKurtosisInstruction(t interface {
+type mockConstructorTestingTNewMockKurtosisInstruction interface {
 	mock.TestingT
 	Cleanup(func())
-}) *MockKurtosisInstruction {
+}
+
+// NewMockKurtosisInstruction creates a new instance of MockKurtosisInstruction. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewMockKurtosisInstruction(t mockConstructorTestingTNewMockKurtosisInstruction) *MockKurtosisInstruction {
 	mock := &MockKurtosisInstruction{}
 	mock.Mock.Test(t)
 
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/plan_module/plan_module.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/plan_module/plan_module.go
index 3090b9bd5..722dccb77 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/plan_module/plan_module.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/plan_module/plan_module.go
@@ -5,6 +5,7 @@ import (
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan/resolver"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_types"
 	"go.starlark.net/starlark"
 	"go.starlark.net/starlarkstruct"
 )
@@ -16,12 +17,13 @@ const (
 func PlanModule(
 	instructionsPlan *instructions_plan.InstructionsPlan,
 	enclaveComponents *enclave_structure.EnclaveComponents,
+	starlarkValueSerde *kurtosis_types.StarlarkValueSerde,
 	instructionsPlanMask *resolver.InstructionsPlanMask,
 	kurtosisPlanInstructions []*kurtosis_plan_instruction.KurtosisPlanInstruction,
 ) *starlarkstruct.Module {
 	moduleBuiltins := starlark.StringDict{}
 	for _, planInstruction := range kurtosisPlanInstructions {
-		wrappedPlanInstruction := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(planInstruction, enclaveComponents, instructionsPlanMask, instructionsPlan)
+		wrappedPlanInstruction := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(planInstruction, enclaveComponents, starlarkValueSerde, instructionsPlanMask, instructionsPlan)
 		moduleBuiltins[planInstruction.GetName()] = starlark.NewBuiltin(planInstruction.GetName(), wrappedPlanInstruction.CreateBuiltin())
 	}
 
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/remove_service/remove_service.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/remove_service/remove_service.go
index a13d8b01e..153b4d973 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/remove_service/remove_service.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/remove_service/remove_service.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -89,6 +90,14 @@ func (builtin *RemoveServiceCapabilities) Execute(ctx context.Context, _ *builti
 	return instructionResult, nil
 }
 
-func (builtin *RemoveServiceCapabilities) TryResolveWith(_ bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *RemoveServiceCapabilities) TryResolveWith(_ bool, _ *enclave_plan_persistence.EnclavePlanInstruction, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	return enclave_structure.InstructionIsNotResolvableAbort
 }
+
+func (builtin *RemoveServiceCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(
+		RemoveServiceBuiltinName,
+	).AddServiceName(
+		builtin.serviceName,
+	)
+}
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/render_templates/render_templates.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/render_templates/render_templates.go
index 62edc3426..6277148eb 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/render_templates/render_templates.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/render_templates/render_templates.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network/render_templates"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -130,20 +131,19 @@ func (builtin *RenderTemplatesCapabilities) Execute(_ context.Context, _ *builti
 	return instructionResult, nil
 }
 
-func (builtin *RenderTemplatesCapabilities) TryResolveWith(instructionsAreEqual bool, other kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *RenderTemplatesCapabilities) TryResolveWith(instructionsAreEqual bool, other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	// if other instruction is nil or other instruction is not an add_service instruction, status is unknown
 	if other == nil {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
-	otherRenderTemplateCapabilities, ok := other.(*RenderTemplatesCapabilities)
-	if !ok {
+	if other.Type != RenderTemplatesBuiltinName {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
 
 	// if artifact names don't match, status is unknown, instructions can't be resolved together
-	if otherRenderTemplateCapabilities.artifactName != builtin.artifactName {
+	if !other.HasOnlyFilesArtifactName(builtin.artifactName) {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
@@ -157,6 +157,17 @@ func (builtin *RenderTemplatesCapabilities) TryResolveWith(instructionsAreEqual
 	return enclave_structure.InstructionIsEqual
 }
 
+func (builtin *RenderTemplatesCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	// technically, we need the MD5 of the files artifact here but because the template is passed in plaintext to the
+	// instruction the check for instruction equality also checks that the content of the artifact is identical.
+	// So we can safely ignore the MD5
+	builder.SetType(
+		RenderTemplatesBuiltinName,
+	).AddFilesArtifact(
+		builtin.artifactName, nil,
+	)
+}
+
 func parseTemplatesAndData(templatesAndData *starlark.Dict) (map[string]*render_templates.TemplateData, *startosis_errors.InterpretationError) {
 	templateAndDataByDestRelFilepath := make(map[string]*render_templates.TemplateData)
 	for _, relPathInFilesArtifactKey := range templatesAndData.Keys() {
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/request/request.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/request/request.go
index 635ec478c..15ed20994 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/request/request.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/request/request.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -187,7 +188,7 @@ func (builtin *RequestCapabilities) Execute(ctx context.Context, _ *builtin_argu
 	return instructionResult, err
 }
 
-func (builtin *RequestCapabilities) TryResolveWith(instructionsAreEqual bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *RequestCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual && enclaveComponents.HasServiceBeenUpdated(builtin.serviceName) {
 		return enclave_structure.InstructionIsUpdate
 	} else if instructionsAreEqual {
@@ -196,6 +197,10 @@ func (builtin *RequestCapabilities) TryResolveWith(instructionsAreEqual bool, _
 	return enclave_structure.InstructionIsUnknown
 }
 
+func (builtin *RequestCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(RequestBuiltinName)
+}
+
 func (builtin *RequestCapabilities) isAcceptableCode(recipeResult map[string]starlark.Comparable) bool {
 	isAcceptableCode := false
 	for _, acceptableCode := range builtin.acceptableCodes {
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/start_service/start_service.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/start_service/start_service.go
index cfb360e7a..9aa6ecc24 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/start_service/start_service.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/start_service/start_service.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -85,6 +86,14 @@ func (builtin *StartServiceCapabilities) Execute(ctx context.Context, _ *builtin
 	return instructionResult, nil
 }
 
-func (builtin *StartServiceCapabilities) TryResolveWith(_ bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *StartServiceCapabilities) TryResolveWith(_ bool, _ *enclave_plan_persistence.EnclavePlanInstruction, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	return enclave_structure.InstructionIsNotResolvableAbort
 }
+
+func (builtin *StartServiceCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(
+		StartServiceBuiltinName,
+	).AddServiceName(
+		builtin.serviceName,
+	)
+}
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/stop_service/stop_service.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/stop_service/stop_service.go
index 4a7cf8b72..13f8dba3d 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/stop_service/stop_service.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/stop_service/stop_service.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -85,6 +86,14 @@ func (builtin *StopServiceCapabilities) Execute(ctx context.Context, _ *builtin_
 	return instructionResult, nil
 }
 
-func (builtin *StopServiceCapabilities) TryResolveWith(_ bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *StopServiceCapabilities) TryResolveWith(_ bool, _ *enclave_plan_persistence.EnclavePlanInstruction, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	return enclave_structure.InstructionIsNotResolvableAbort
 }
+
+func (builtin *StopServiceCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(
+		StopServiceBuiltinName,
+	).AddServiceName(
+		builtin.serviceName,
+	)
+}
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/store_service_files/store_service_files.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/store_service_files/store_service_files.go
index 6ff5b06ed..c180717ae 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/store_service_files/store_service_files.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/store_service_files/store_service_files.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	kurtosis_backend_service "github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -126,20 +127,19 @@ func (builtin *StoreServiceFilesCapabilities) Execute(ctx context.Context, _ *bu
 	return instructionResult, nil
 }
 
-func (builtin *StoreServiceFilesCapabilities) TryResolveWith(instructionsAreEqual bool, other kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *StoreServiceFilesCapabilities) TryResolveWith(instructionsAreEqual bool, other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	// if other instruction is nil or other instruction is not an add_service instruction, status is unknown
 	if other == nil {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
-	otherStoreServiceFilesCapabilities, ok := other.(*StoreServiceFilesCapabilities)
-	if !ok {
+	if other.Type != StoreServiceFilesBuiltinName {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
 
 	// if artifact names don't match, status is unknown, instructions can't be resolved together
-	if otherStoreServiceFilesCapabilities.artifactName != builtin.artifactName {
+	if !other.HasOnlyFilesArtifactName(builtin.artifactName) {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
@@ -160,3 +160,14 @@ func (builtin *StoreServiceFilesCapabilities) TryResolveWith(instructionsAreEqua
 	enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentWasLeftIntact)
 	return enclave_structure.InstructionIsEqual
 }
+
+func (builtin *StoreServiceFilesCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	// No need for the MD5 here because "store_service_files" is an atomic operation at the service_network level.
+	// Here we just consider that if the service has been updated, we store the file again (b/c it the content might
+	// have changed), otherwise we don't
+	builder.SetType(
+		StoreServiceFilesBuiltinName,
+	).AddFilesArtifact(
+		builtin.artifactName, nil,
+	)
+}
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_python.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_python.go
index 6ce2e29c2..1ca65cfe4 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_python.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_python.go
@@ -8,6 +8,7 @@ import (
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service_directory"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers/magic_string_helper"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
@@ -330,13 +331,17 @@ func (builtin *RunPythonCapabilities) Execute(ctx context.Context, _ *builtin_ar
 	return instructionResult, err
 }
 
-func (builtin *RunPythonCapabilities) TryResolveWith(instructionsAreEqual bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *RunPythonCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual {
 		return enclave_structure.InstructionIsEqual
 	}
 	return enclave_structure.InstructionIsUnknown
 }
 
+func (builtin *RunPythonCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(RunPythonBuiltinName)
+}
+
 func setupRequiredPackages(ctx context.Context, builtin *RunPythonCapabilities) (*exec_result.ExecResult, error) {
 	if len(builtin.packages) == 0 {
 		return nil, nil
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_sh.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_sh.go
index 2b5c56c21..971686b79 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_sh.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/tasks/run_sh.go
@@ -6,6 +6,7 @@ import (
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service_directory"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers/magic_string_helper"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
@@ -234,13 +235,17 @@ func (builtin *RunShCapabilities) Execute(ctx context.Context, _ *builtin_argume
 	return instructionResult, err
 }
 
-func (builtin *RunShCapabilities) TryResolveWith(instructionsAreEqual bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *RunShCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, _ *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual {
 		return enclave_structure.InstructionIsEqual
 	}
 	return enclave_structure.InstructionIsUnknown
 }
 
+func (builtin *RunShCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(RunShBuiltinName)
+}
+
 func getCommandToRun(builtin *RunShCapabilities) (string, error) {
 	// replace future references to actual strings
 	maybeSubCommandWithRuntimeValues, err := magic_string_helper.ReplaceRuntimeValueInString(builtin.run, builtin.runtimeValueStore)
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/upload_files/upload_files.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/upload_files/upload_files.go
index a5d51a93a..d9b62504a 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/upload_files/upload_files.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/upload_files/upload_files.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/lib/shared_utils"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
@@ -160,20 +161,20 @@ func (builtin *UploadFilesCapabilities) Execute(_ context.Context, _ *builtin_ar
 	return instructionResult, nil
 }
 
-func (builtin *UploadFilesCapabilities) TryResolveWith(instructionsAreEqual bool, other kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *UploadFilesCapabilities) TryResolveWith(instructionsAreEqual bool, other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	// if other instruction is nil or other instruction is not an add_service instruction, status is unknown
 	if other == nil {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
-	otherUploadFilesCapabilities, ok := other.(*UploadFilesCapabilities)
-	if !ok {
+
+	if other.Type != UploadFilesBuiltinName {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
 
 	// if artifact names don't match, status is unknown, instructions can't be resolved together
-	if otherUploadFilesCapabilities.artifactName != builtin.artifactName {
+	if !other.HasOnlyFilesArtifactName(builtin.artifactName) {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsNew)
 		return enclave_structure.InstructionIsUnknown
 	}
@@ -186,10 +187,18 @@ func (builtin *UploadFilesCapabilities) TryResolveWith(instructionsAreEqual bool
 
 	// From here the instructions are equal
 	// If the hash of the files don't match, the instruction needs to be re-run
-	if !bytes.Equal(otherUploadFilesCapabilities.filesArtifactMd5, builtin.filesArtifactMd5) {
+	if !other.HasOnlyFilesArtifactMd5(builtin.filesArtifactMd5) {
 		enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentIsUpdated)
 		return enclave_structure.InstructionIsUpdate
 	}
 	enclaveComponents.AddFilesArtifact(builtin.artifactName, enclave_structure.ComponentWasLeftIntact)
 	return enclave_structure.InstructionIsEqual
 }
+
+func (builtin *UploadFilesCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(
+		UploadFilesBuiltinName,
+	).AddFilesArtifact(
+		builtin.artifactName, builtin.filesArtifactMd5,
+	)
+}
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_instruction/wait/wait.go b/core/server/api_container/server/startosis_engine/kurtosis_instruction/wait/wait.go
index 18d7a250a..79b117453 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_instruction/wait/wait.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_instruction/wait/wait.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/assert"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers"
@@ -275,7 +276,7 @@ func (builtin *WaitCapabilities) Execute(ctx context.Context, _ *builtin_argumen
 	return instructionResult, nil
 }
 
-func (builtin *WaitCapabilities) TryResolveWith(instructionsAreEqual bool, _ kurtosis_plan_instruction.KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *WaitCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual && enclaveComponents.HasServiceBeenUpdated(builtin.serviceName) {
 		return enclave_structure.InstructionIsUpdate
 	} else if instructionsAreEqual {
@@ -283,3 +284,7 @@ func (builtin *WaitCapabilities) TryResolveWith(instructionsAreEqual bool, _ kur
 	}
 	return enclave_structure.InstructionIsUnknown
 }
+
+func (builtin *WaitCapabilities) FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder) {
+	builder.SetType(WaitBuiltinName)
+}
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction.go b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction.go
index fbb3220a1..843658930 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction.go
@@ -1,10 +1,12 @@
 package kurtosis_plan_instruction
 
 import (
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan/resolver"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_types"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_errors"
 	"github.com/kurtosis-tech/stacktrace"
 	"github.com/sirupsen/logrus"
@@ -29,14 +31,17 @@ type KurtosisPlanInstructionWrapper struct {
 
 	// TODO: This can be changed to KurtosisPlanInstructionInternal when we get rid of KurtosisInstruction
 	instructionsPlan *instructions_plan.InstructionsPlan
+
+	starlarkValueSerde *kurtosis_types.StarlarkValueSerde
 }
 
-func NewKurtosisPlanInstructionWrapper(instruction *KurtosisPlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents, instructionPlanMask *resolver.InstructionsPlanMask, instructionsPlan *instructions_plan.InstructionsPlan) *KurtosisPlanInstructionWrapper {
+func NewKurtosisPlanInstructionWrapper(instruction *KurtosisPlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents, starlarkValueSerde *kurtosis_types.StarlarkValueSerde, instructionPlanMask *resolver.InstructionsPlanMask, instructionsPlan *instructions_plan.InstructionsPlan) *KurtosisPlanInstructionWrapper {
 	return &KurtosisPlanInstructionWrapper{
 		KurtosisPlanInstruction: instruction,
 		enclaveComponents:       enclaveComponents,
 		instructionPlanMask:     instructionPlanMask,
 		instructionsPlan:        instructionsPlan,
+		starlarkValueSerde:      starlarkValueSerde,
 	}
 }
 
@@ -54,12 +59,12 @@ func (builtin *KurtosisPlanInstructionWrapper) CreateBuiltin() func(thread *star
 			return nil, interpretationErr
 		}
 
-		var scheduledInstructionPulledFromMaskMaybe *instructions_plan.ScheduledInstruction
+		var enclavePlanInstructionPulledFromMaskMaybe *enclave_plan_persistence.EnclavePlanInstruction
 		var instructionResolutionStatus enclave_structure.InstructionResolutionStatus
 		if builtin.instructionPlanMask.HasNext() {
-			_, scheduledInstructionPulledFromMaskMaybe = builtin.instructionPlanMask.Next()
-			if scheduledInstructionPulledFromMaskMaybe != nil {
-				instructionResolutionStatus = instructionWrapper.TryResolveWith(scheduledInstructionPulledFromMaskMaybe.GetInstruction(), builtin.enclaveComponents)
+			_, enclavePlanInstructionPulledFromMaskMaybe = builtin.instructionPlanMask.Next()
+			if enclavePlanInstructionPulledFromMaskMaybe != nil {
+				instructionResolutionStatus = instructionWrapper.TryResolveWith(enclavePlanInstructionPulledFromMaskMaybe, builtin.enclaveComponents)
 			} else {
 				instructionResolutionStatus = instructionWrapper.TryResolveWith(nil, builtin.enclaveComponents)
 			}
@@ -69,9 +74,27 @@ func (builtin *KurtosisPlanInstructionWrapper) CreateBuiltin() func(thread *star
 
 		switch instructionResolutionStatus {
 		case enclave_structure.InstructionIsEqual:
-			// add instruction from the mask and mark it as executed but not imported from the current enclave plan
-			builtin.instructionsPlan.AddScheduledInstruction(scheduledInstructionPulledFromMaskMaybe).Executed(true)
-			return scheduledInstructionPulledFromMaskMaybe.GetReturnedValue(), nil
+			// Build a scheduled instruction from a mix of the instruction and the EnclavePlanInstruction from the mask
+			// The important thing is to keep the returnedValue from the enclavePlan instruction such that runtime values
+			// IDs are kept. We also re-use the same UUID to make sure it's stable, but right now enclave plan instruction
+			// UUIDs are not used so it's not critical here
+			// Lastly, this scheduled instruction is marked as "EXECUTED" such that it will later be skipped by the
+			// executor
+			returnedValue, err := builtin.starlarkValueSerde.Deserialize(enclavePlanInstructionPulledFromMaskMaybe.ReturnedValue)
+			if err != nil {
+				return nil, startosis_errors.WrapWithInterpretationError(err, "The instruction was resolved with "+
+					"an instruction from the enclave plan, but the result of this instruction could not be"+
+					"deserialized. The instruction was: '%s' and its result was: '%s'",
+					enclavePlanInstructionPulledFromMaskMaybe.StarlarkCode,
+					enclavePlanInstructionPulledFromMaskMaybe.ReturnedValue)
+			}
+			scheduledInstruction := instructions_plan.NewScheduledInstruction(
+				instructions_plan.ScheduledInstructionUuid(enclavePlanInstructionPulledFromMaskMaybe.Uuid),
+				instructionWrapper,
+				returnedValue,
+			).Executed(true)
+			builtin.instructionsPlan.AddScheduledInstruction(scheduledInstruction).Executed(true)
+			return returnedValue, nil
 		case enclave_structure.InstructionIsUpdate:
 			// otherwise add the instruction as a new one to the plan and return its own returned value
 			if err := builtin.instructionsPlan.AddInstruction(instructionWrapper, returnedFutureValue); err != nil {
@@ -87,10 +110,10 @@ func (builtin *KurtosisPlanInstructionWrapper) CreateBuiltin() func(thread *star
 					instructionWrapper.String(),
 					instructionWrapper.GetPositionInOriginalScript().String())
 			}
-			if scheduledInstructionPulledFromMaskMaybe != nil {
+			if enclavePlanInstructionPulledFromMaskMaybe != nil {
 				builtin.instructionPlanMask.MarkAsInvalid()
 				logrus.Debugf("Marking the plan as invalid as instruction '%s' differs from '%s'",
-					instructionWrapper.String(), scheduledInstructionPulledFromMaskMaybe.GetInstruction().String())
+					instructionWrapper.String(), enclavePlanInstructionPulledFromMaskMaybe.StarlarkCode)
 			}
 			return returnedFutureValue, nil
 		case enclave_structure.InstructionIsNotResolvableAbort:
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_capabilities.go b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_capabilities.go
index 5ffda026b..a96e12a9f 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_capabilities.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_capabilities.go
@@ -2,6 +2,7 @@ package kurtosis_plan_instruction
 
 import (
 	"context"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_errors"
@@ -16,5 +17,9 @@ type KurtosisPlanInstructionCapabilities interface {
 
 	Execute(ctx context.Context, arguments *builtin_argument.ArgumentValuesSet) (string, error)
 
-	TryResolveWith(instructionsAreEqual bool, other KurtosisPlanInstructionCapabilities, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus
+	TryResolveWith(instructionsAreEqual bool, other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus
+
+	// FillPersistableAttributes adds to the builder the attributes of the instruction that needs to be persisted to the
+	// enclave database to power idempotent runs.
+	FillPersistableAttributes(builder *enclave_plan_persistence.EnclavePlanInstructionBuilder)
 }
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_internal.go b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_internal.go
index 1f7fa008f..adcc43f94 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_internal.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction_internal.go
@@ -4,8 +4,8 @@ import (
 	"context"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/kurtosis_core_rpc_api_bindings"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/lib/binding_constructors"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
-	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/builtin_argument"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_errors"
@@ -77,7 +77,7 @@ func (builtin *kurtosisPlanInstructionInternal) Execute(ctx context.Context) (*s
 	return &result, nil
 }
 
-func (builtin *kurtosisPlanInstructionInternal) TryResolveWith(other kurtosis_instruction.KurtosisInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *kurtosisPlanInstructionInternal) TryResolveWith(other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	isAnAbortAllInstruction := builtin.capabilities.TryResolveWith(false, nil, enclaveComponents) == enclave_structure.InstructionIsNotResolvableAbort
 	if isAnAbortAllInstruction {
 		return enclave_structure.InstructionIsNotResolvableAbort
@@ -87,13 +87,14 @@ func (builtin *kurtosisPlanInstructionInternal) TryResolveWith(other kurtosis_in
 		return enclave_structure.InstructionIsUnknown
 	}
 
-	otherPlanInstruction, ok := other.(*kurtosisPlanInstructionInternal)
-	if !ok {
-		return enclave_structure.InstructionIsUnknown
-	}
+	instructionsAreEqual := builtin.String() == other.StarlarkCode
+	return builtin.capabilities.TryResolveWith(instructionsAreEqual, other, enclaveComponents)
+}
 
-	instructionsAreEqual := builtin.String() == other.String()
-	return builtin.capabilities.TryResolveWith(instructionsAreEqual, otherPlanInstruction.capabilities, enclaveComponents)
+func (builtin *kurtosisPlanInstructionInternal) GetPersistableAttributes() *enclave_plan_persistence.EnclavePlanInstructionBuilder {
+	enclavePlaneInstructionBuilder := enclave_plan_persistence.NewEnclavePlanInstructionBuilder()
+	builtin.capabilities.FillPersistableAttributes(enclavePlaneInstructionBuilder)
+	return enclavePlaneInstructionBuilder.SetStarlarkCode(builtin.String())
 }
 
 func (builtin *kurtosisPlanInstructionInternal) interpret(locatorOfModuleInWhichThisBuiltInIsBeingCalled string) (starlark.Value, *startosis_errors.InterpretationError) {
diff --git a/core/server/api_container/server/startosis_engine/startosis_executor.go b/core/server/api_container/server/startosis_engine/startosis_executor.go
index d4f87c8f3..5d81678f2 100644
--- a/core/server/api_container/server/startosis_engine/startosis_executor.go
+++ b/core/server/api_container/server/startosis_engine/startosis_executor.go
@@ -5,8 +5,11 @@ import (
 	"fmt"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/kurtosis_core_rpc_api_bindings"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/lib/binding_constructors"
+	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/database_accessors/enclave_db"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers/magic_string_helper"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_types"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/runtime_value_store"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_constants"
 	"github.com/kurtosis-tech/stacktrace"
@@ -27,20 +30,24 @@ var (
 )
 
 type StartosisExecutor struct {
-	mutex             *sync.Mutex
-	enclavePlan       *instructions_plan.InstructionsPlan
-	runtimeValueStore *runtime_value_store.RuntimeValueStore
+	mutex              *sync.Mutex
+	starlarkValueSerde *kurtosis_types.StarlarkValueSerde
+	enclavePlan        *enclave_plan_persistence.EnclavePlan
+	enclaveDb          *enclave_db.EnclaveDB
+	runtimeValueStore  *runtime_value_store.RuntimeValueStore
 }
 
 type ExecutionError struct {
 	Error string
 }
 
-func NewStartosisExecutor(runtimeValueStore *runtime_value_store.RuntimeValueStore) *StartosisExecutor {
+func NewStartosisExecutor(starlarkValueSerde *kurtosis_types.StarlarkValueSerde, runtimeValueStore *runtime_value_store.RuntimeValueStore, enclavePlan *enclave_plan_persistence.EnclavePlan, enclaveDb *enclave_db.EnclaveDB) *StartosisExecutor {
 	return &StartosisExecutor{
-		mutex:             &sync.Mutex{},
-		enclavePlan:       instructions_plan.NewInstructionsPlan(),
-		runtimeValueStore: runtimeValueStore,
+		mutex:              &sync.Mutex{},
+		starlarkValueSerde: starlarkValueSerde,
+		enclaveDb:          enclaveDb,
+		enclavePlan:        enclavePlan,
+		runtimeValueStore:  runtimeValueStore,
 	}
 }
 
@@ -65,11 +72,20 @@ func (executor *StartosisExecutor) Execute(ctx context.Context, dryRun bool, par
 		logrus.Debugf("Current enclave plan contains %d instuctions. About to process a new plan with %d instructions starting at index %d (dry-run: %v)",
 			executor.enclavePlan.Size(), len(instructionsSequence), indexOfFirstInstructionInEnclavePlan, dryRun)
 
-		var err error
-		executor.enclavePlan, err = executor.enclavePlan.PartialClone(indexOfFirstInstructionInEnclavePlan)
-		if err != nil {
-			sendErrorAndFail(starlarkRunResponseLineStream, err, "An error occurred keeping the enclave plan up to date with the current enclave state")
-		}
+		executor.enclavePlan = executor.enclavePlan.PartialDeepClone(indexOfFirstInstructionInEnclavePlan)
+
+		defer func() {
+			// TODO: we now perist the plan at the end of the execution. We could persist it everytime an instruction
+			//  is executed, to be resilient to the APIC being stopped in the middle of a Starlark script execution
+			//  Or we could even persist it only when the APIC is stopped. This seems to be a good middle ground, but
+			//  can be tuned according the our future needs
+			logrus.Infof("Persisting enclave plan composed of %d instructions into enclave database", executor.enclavePlan.Size())
+			if err := executor.enclavePlan.Persist(executor.enclaveDb); err != nil {
+				logrus.Errorf("An error occurred persisting the enclave plan at the end of the execution of the" +
+					"package. The enclave will continue to run, but next runs of Starlark package might not be executed" +
+					"as expected.")
+			}
+		}()
 
 		logrus.Debugf("Transfered %d instructions from previous enclave plan to keep the enclave state consistent", executor.enclavePlan.Size())
 
@@ -104,8 +120,16 @@ func (executor *StartosisExecutor) Execute(ctx context.Context, dryRun bool, par
 					}
 					starlarkRunResponseLineStream <- binding_constructors.NewStarlarkRunResponseLineFromInstructionResult(instructionOutputStr)
 				}
-				// mark the instruction as executed and add it to the current instruction plan
-				executor.enclavePlan.AddScheduledInstruction(scheduledInstruction).Executed(true)
+				// add the instruction into the current enclave plan
+				enclavePlanInstruction, err := scheduledInstruction.GetInstruction().GetPersistableAttributes().SetUuid(
+					string(scheduledInstruction.GetUuid()),
+				).SetReturnedValue(
+					executor.starlarkValueSerde.Serialize(scheduledInstruction.GetReturnedValue()),
+				).Build()
+				if err != nil {
+					sendErrorAndFail(starlarkRunResponseLineStream, err, "An error occurred persisting instruction (number %d) at %v after it's been executed:\n%v", instructionNumber, instruction.GetPositionInOriginalScript().String(), instruction.String())
+				}
+				executor.enclavePlan.AppendInstruction(enclavePlanInstruction)
 			}
 		}
 
@@ -125,7 +149,7 @@ func (executor *StartosisExecutor) Execute(ctx context.Context, dryRun bool, par
 	return starlarkRunResponseLineStream
 }
 
-func (executor *StartosisExecutor) GetCurrentEnclavePLan() *instructions_plan.InstructionsPlan {
+func (executor *StartosisExecutor) GetCurrentEnclavePLan() *enclave_plan_persistence.EnclavePlan {
 	return executor.enclavePlan
 }
 
diff --git a/core/server/api_container/server/startosis_engine/startosis_interpreter.go b/core/server/api_container/server/startosis_engine/startosis_interpreter.go
index 9383822f1..393d43dd1 100644
--- a/core/server/api_container/server/startosis_engine/startosis_interpreter.go
+++ b/core/server/api_container/server/startosis_engine/startosis_interpreter.go
@@ -7,10 +7,12 @@ import (
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/builtins"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/builtins/print_builtin"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan/resolver"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/plan_module"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_types"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/package_io"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/runtime_value_store"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_constants"
@@ -54,6 +56,7 @@ type StartosisInterpreter struct {
 	serviceNetwork     service_network.ServiceNetwork
 	recipeExecutor     *runtime_value_store.RuntimeValueStore
 	moduleGlobalsCache map[string]*startosis_packages.ModuleCacheEntry
+	starlarkValueSerde *kurtosis_types.StarlarkValueSerde
 	// TODO AUTH there will be a leak here in case people with different repo visibility access a module
 	moduleContentProvider startosis_packages.PackageContentProvider
 	enclaveEnvVars        string
@@ -61,7 +64,7 @@ type StartosisInterpreter struct {
 
 type SerializedInterpretationOutput string
 
-func NewStartosisInterpreter(serviceNetwork service_network.ServiceNetwork, moduleContentProvider startosis_packages.PackageContentProvider, runtimeValueStore *runtime_value_store.RuntimeValueStore, enclaveVarEnvs string) *StartosisInterpreter {
+func NewStartosisInterpreter(serviceNetwork service_network.ServiceNetwork, moduleContentProvider startosis_packages.PackageContentProvider, runtimeValueStore *runtime_value_store.RuntimeValueStore, starlarkValueSerde *kurtosis_types.StarlarkValueSerde, enclaveVarEnvs string) *StartosisInterpreter {
 	return &StartosisInterpreter{
 		mutex:                 &sync.Mutex{},
 		serviceNetwork:        serviceNetwork,
@@ -69,6 +72,7 @@ func NewStartosisInterpreter(serviceNetwork service_network.ServiceNetwork, modu
 		moduleGlobalsCache:    make(map[string]*startosis_packages.ModuleCacheEntry),
 		moduleContentProvider: moduleContentProvider,
 		enclaveEnvVars:        enclaveVarEnvs,
+		starlarkValueSerde:    starlarkValueSerde,
 	}
 }
 
@@ -87,7 +91,7 @@ func (interpreter *StartosisInterpreter) InterpretAndOptimizePlan(
 	relativePathtoMainFile string,
 	serializedStarlark string,
 	serializedJsonParams string,
-	currentEnclavePlan *instructions_plan.InstructionsPlan,
+	currentEnclavePlan *enclave_plan_persistence.EnclavePlan,
 ) (string, *instructions_plan.InstructionsPlan, *kurtosis_core_rpc_api_bindings.StarlarkInterpretationError) {
 
 	// run interpretation with no mask at all to generate the list of instructions as if the enclave was empty
@@ -104,7 +108,7 @@ func (interpreter *StartosisInterpreter) InterpretAndOptimizePlan(
 	}
 	logrus.Debugf("First interpretation of package generated %d instructions", len(naiveInstructionsPlanSequence))
 
-	currentEnclavePlanSequence, interpretationErr := currentEnclavePlan.GeneratePlan()
+	currentEnclavePlanSequence := currentEnclavePlan.GeneratePlan()
 	if interpretationErr != nil {
 		return startosis_constants.NoOutputObject, nil, interpretationErr.ToAPIType()
 	}
@@ -258,7 +262,7 @@ func (interpreter *StartosisInterpreter) Interpret(
 		firstParamName, _ := mainFunction.Param(planParamIndex)
 		if firstParamName == planParamName {
 			kurtosisPlanInstructions := KurtosisPlanInstructions(interpreter.serviceNetwork, interpreter.recipeExecutor, interpreter.moduleContentProvider)
-			planModule := plan_module.PlanModule(newInstructionsPlan, enclaveComponents, instructionsPlanMask, kurtosisPlanInstructions)
+			planModule := plan_module.PlanModule(newInstructionsPlan, enclaveComponents, interpreter.starlarkValueSerde, instructionsPlanMask, kurtosisPlanInstructions)
 			argsTuple = append(argsTuple, planModule)
 		}
 
@@ -356,14 +360,14 @@ func (interpreter *StartosisInterpreter) buildBindings(thread *starlark.Thread,
 	return &predeclared, nil
 }
 
-func findFirstEqualInstructionPastIndex(currentEnclaveInstructionsList []*instructions_plan.ScheduledInstruction, naiveInstructionsList []*instructions_plan.ScheduledInstruction, minIndex int) int {
+func findFirstEqualInstructionPastIndex(currentEnclaveInstructionsList []*enclave_plan_persistence.EnclavePlanInstruction, naiveInstructionsList []*instructions_plan.ScheduledInstruction, minIndex int) int {
 	if len(naiveInstructionsList) == 0 {
 		return -1 // no result as the naiveInstructionsList is empty
 	}
 	for i := minIndex; i < len(currentEnclaveInstructionsList); i++ {
 		// We just need to compare instructions to see if they match, without needing any enclave specific context here
 		fakeEnclaveComponent := enclave_structure.NewEnclaveComponents()
-		instructionResolutionResult := naiveInstructionsList[0].GetInstruction().TryResolveWith(currentEnclaveInstructionsList[i].GetInstruction(), fakeEnclaveComponent)
+		instructionResolutionResult := naiveInstructionsList[0].GetInstruction().TryResolveWith(currentEnclaveInstructionsList[i], fakeEnclaveComponent)
 		if instructionResolutionResult == enclave_structure.InstructionIsEqual || instructionResolutionResult == enclave_structure.InstructionIsUpdate {
 			return i
 		}


Test patch:
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/suite_kurtosis_plan_instruction_test.go b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/suite_kurtosis_plan_instruction_test.go
index a387efa1a..c5996d62c 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/suite_kurtosis_plan_instruction_test.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/suite_kurtosis_plan_instruction_test.go
@@ -57,7 +57,7 @@ func (suite *KurtosisPlanInstructionTestSuite) run(builtin KurtosisPlanInstructi
 	instructionFromBuiltin := builtin.GetInstruction()
 	emptyEnclaveComponents := enclave_structure.NewEnclaveComponents()
 	emptyInstructionsPlanMask := resolver.NewInstructionsPlanMask(0)
-	instructionWrapper := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(instructionFromBuiltin, emptyEnclaveComponents, emptyInstructionsPlanMask, instructionsPlan)
+	instructionWrapper := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(instructionFromBuiltin, emptyEnclaveComponents, nil, emptyInstructionsPlanMask, instructionsPlan)
 	suite.starlarkEnv[instructionWrapper.GetName()] = starlark.NewBuiltin(instructionWrapper.GetName(), instructionWrapper.CreateBuiltin())
 
 	starlarkCode := builtin.GetStarlarkCode()
diff --git a/core/server/api_container/server/startosis_engine/startosis_executor_test.go b/core/server/api_container/server/startosis_engine/startosis_executor_test.go
index 2b2d3ee18..8da1fa06d 100644
--- a/core/server/api_container/server/startosis_engine/startosis_executor_test.go
+++ b/core/server/api_container/server/startosis_engine/startosis_executor_test.go
@@ -3,9 +3,11 @@ package startosis_engine
 import (
 	"context"
 	"errors"
+	"github.com/google/uuid"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/kurtosis_core_rpc_api_bindings"
 	"github.com/kurtosis-tech/kurtosis/api/golang/core/lib/binding_constructors"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/database_accessors/enclave_db"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/mock_instruction"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers"
@@ -47,7 +49,7 @@ func TestExecuteKurtosisInstructions_ExecuteForReal_Success(t *testing.T) {
 	runtimeValueStore, createRuntimeValueStoreErr := runtime_value_store.CreateRuntimeValueStore(dummySerde, enclaveDb)
 	require.NoError(t, createRuntimeValueStoreErr)
 
-	executor := NewStartosisExecutor(runtimeValueStore)
+	executor := NewStartosisExecutor(nil, runtimeValueStore, enclave_plan_persistence.NewEnclavePlan(), enclaveDb)
 
 	instructionsPlan := instructions_plan.NewInstructionsPlan()
 	instruction1 := createMockInstruction(t, "instruction1", executeSuccessfully)
@@ -91,7 +93,7 @@ func TestExecuteKurtosisInstructions_ExecuteForReal_FailureHalfWay(t *testing.T)
 	runtimeValueStore, err := runtime_value_store.CreateRuntimeValueStore(dummySerde, enclaveDb)
 	require.NoError(t, err)
 
-	executor := NewStartosisExecutor(runtimeValueStore)
+	executor := NewStartosisExecutor(nil, runtimeValueStore, enclave_plan_persistence.NewEnclavePlan(), enclaveDb)
 
 	instruction1 := createMockInstruction(t, "instruction1", executeSuccessfully)
 	instruction2 := createMockInstruction(t, "instruction2", throwOnExecute)
@@ -136,7 +138,7 @@ func TestExecuteKurtosisInstructions_DoDryRun(t *testing.T) {
 	runtimeValueStore, createRuntimeValueStoreErr := runtime_value_store.CreateRuntimeValueStore(dummySerde, enclaveDb)
 	require.NoError(t, createRuntimeValueStoreErr)
 
-	executor := NewStartosisExecutor(runtimeValueStore)
+	executor := NewStartosisExecutor(nil, runtimeValueStore, enclave_plan_persistence.NewEnclavePlan(), enclaveDb)
 
 	instruction1 := createMockInstruction(t, "instruction1", executeSuccessfully)
 	instruction2 := createMockInstruction(t, "instruction2", executeSuccessfully)
@@ -171,6 +173,10 @@ func createMockInstruction(t *testing.T, instructionName string, executeSuccessf
 	instruction.EXPECT().GetCanonicalInstruction(mock.Anything).Maybe().Return(canonicalInstruction)
 	instruction.EXPECT().GetPositionInOriginalScript().Maybe().Return(dummyPosition)
 	instruction.EXPECT().String().Maybe().Return(stringifiedInstruction)
+	instruction.EXPECT().GetPersistableAttributes().Maybe().Return(
+		enclave_plan_persistence.NewEnclavePlanInstructionBuilder().SetUuid(uuid.New().String()).SetType(instructionName).SetStarlarkCode(stringifiedInstruction).SetReturnedValue("None"),
+		nil,
+	)
 
 	if executeSuccessfully {
 		instruction.EXPECT().Execute(mock.Anything).Maybe().Return(nil, nil)
diff --git a/core/server/api_container/server/startosis_engine/startosis_interpreter_idempotent_test.go b/core/server/api_container/server/startosis_engine/startosis_interpreter_idempotent_test.go
index c8e878672..c78d3e5cf 100644
--- a/core/server/api_container/server/startosis_engine/startosis_interpreter_idempotent_test.go
+++ b/core/server/api_container/server/startosis_engine/startosis_interpreter_idempotent_test.go
@@ -2,21 +2,29 @@ package startosis_engine
 
 import (
 	"context"
+	"github.com/google/uuid"
 	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/backend_interface/objects/enclave"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/service_network"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_plan_persistence"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/enclave_structure"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan/resolver"
-	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/shared_helpers"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_types"
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_types/port_spec"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/runtime_value_store"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_constants"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_packages/mock_package_content_provider"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	"go.starlark.net/starlark"
+	"go.starlark.net/starlarkstruct"
 	"net"
 	"testing"
 )
 
 const (
+	starlarkValueSerdeThreadName = "test-serde-thread"
+
 	enclaveUuid = enclave.EnclaveUUID("enclave-uuid")
 
 	noInputParams = "{}"
@@ -32,9 +40,22 @@ func (suite *StartosisInterpreterIdempotentTestSuite) SetupTest() {
 	suite.packageContentProvider = mock_package_content_provider.NewMockPackageContentProvider()
 	enclaveDb := getEnclaveDBForTest(suite.T())
 
-	dummySerde := shared_helpers.NewDummyStarlarkValueSerDeForTest()
-
-	runtimeValueStore, err := runtime_value_store.CreateRuntimeValueStore(dummySerde, enclaveDb)
+	thread := &starlark.Thread{
+		Name:       starlarkValueSerdeThreadName,
+		Print:      nil,
+		Load:       nil,
+		OnMaxSteps: nil,
+		Steps:      0,
+	}
+	starlarkEnv := starlark.StringDict{
+		starlarkstruct.Default.GoString(): starlark.NewBuiltin(starlarkstruct.Default.GoString(), starlarkstruct.Make),
+
+		kurtosis_types.ServiceTypeName: starlark.NewBuiltin(kurtosis_types.ServiceTypeName, kurtosis_types.NewServiceType().CreateBuiltin()),
+		port_spec.PortSpecTypeName:     starlark.NewBuiltin(port_spec.PortSpecTypeName, port_spec.NewPortSpecType().CreateBuiltin()),
+	}
+	starlarkValueSerde := kurtosis_types.NewStarlarkValueSerde(thread, starlarkEnv)
+
+	runtimeValueStore, err := runtime_value_store.CreateRuntimeValueStore(starlarkValueSerde, enclaveDb)
 	require.NoError(suite.T(), err)
 
 	serviceNetwork := service_network.NewMockServiceNetwork(suite.T())
@@ -42,7 +63,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) SetupTest() {
 		service_network.NewApiContainerInfo(net.IPv4(0, 0, 0, 0), uint16(1234), "0.0.0"),
 	)
 	serviceNetwork.EXPECT().GetEnclaveUuid().Maybe().Return(enclaveUuid)
-	suite.interpreter = NewStartosisInterpreter(serviceNetwork, suite.packageContentProvider, runtimeValueStore, "")
+	suite.interpreter = NewStartosisInterpreter(serviceNetwork, suite.packageContentProvider, runtimeValueStore, starlarkValueSerde, "")
 }
 
 func TestRunStartosisInterpreterIdempotentTestSuite(t *testing.T) {
@@ -75,6 +96,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_I
 		resolver.NewInstructionsPlanMask(0))
 	require.Nil(suite.T(), interpretationApiErr)
 	require.Equal(suite.T(), 3, currentEnclavePlan.Size())
+	convertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(currentEnclavePlan)
 
 	// Interpret the updated script against the current enclave plan
 	_, instructionsPlan, interpretationError := suite.interpreter.InterpretAndOptimizePlan(
@@ -84,7 +106,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_I
 		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
 		script,
 		noInputParams,
-		currentEnclavePlan,
+		convertedEnclavePlan,
 	)
 	require.Nil(suite.T(), interpretationError)
 
@@ -127,6 +149,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_A
 		resolver.NewInstructionsPlanMask(0))
 	require.Nil(suite.T(), interpretationApiErr)
 	require.Equal(suite.T(), 2, currentEnclavePlan.Size())
+	convertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(currentEnclavePlan)
 
 	updatedScript := `def run(plan, args):
 	plan.print("instruction1")
@@ -141,7 +164,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_A
 		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
 		updatedScript,
 		noInputParams,
-		currentEnclavePlan,
+		convertedEnclavePlan,
 	)
 	require.Nil(suite.T(), interpretationError)
 
@@ -185,6 +208,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_D
 		resolver.NewInstructionsPlanMask(0))
 	require.Nil(suite.T(), interpretationApiErr)
 	require.Equal(suite.T(), 2, currentEnclavePlan.Size())
+	convertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(currentEnclavePlan)
 
 	updatedScript := `def run(plan, args):
 	plan.print("instruction3")
@@ -197,7 +221,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_D
 		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
 		updatedScript,
 		noInputParams,
-		currentEnclavePlan,
+		convertedEnclavePlan,
 	)
 	require.Nil(suite.T(), interpretationError)
 
@@ -235,6 +259,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_I
 		resolver.NewInstructionsPlanMask(0))
 	require.Nil(suite.T(), interpretationApiErr)
 	require.Equal(suite.T(), 3, currentEnclavePlan.Size())
+	convertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(currentEnclavePlan)
 
 	updatedScript := `def run(plan, args):
 	plan.print(msg="instruction1")
@@ -249,7 +274,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_I
 		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
 		updatedScript,
 		noInputParams,
-		currentEnclavePlan,
+		convertedEnclavePlan,
 	)
 	require.Nil(suite.T(), interpretationError)
 
@@ -293,6 +318,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_A
 		resolver.NewInstructionsPlanMask(0))
 	require.Nil(suite.T(), interpretationApiErr)
 	require.Equal(suite.T(), 4, currentEnclavePlan.Size())
+	convertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(currentEnclavePlan)
 
 	updatedScript := `def run(plan):
 	service_1 = plan.add_service(name="service_1", config=ServiceConfig(image="kurtosistech/image:1.5.0")) # <-- version updated
@@ -308,7 +334,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_A
 		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
 		updatedScript,
 		noInputParams,
-		currentEnclavePlan,
+		convertedEnclavePlan,
 	)
 	require.Nil(suite.T(), interpretationError)
 
@@ -361,6 +387,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_U
 		resolver.NewInstructionsPlanMask(0))
 	require.Nil(suite.T(), interpretationApiErr)
 	require.Equal(suite.T(), 4, currentEnclavePlan.Size())
+	convertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(currentEnclavePlan)
 
 	updatedScript := `def run(plan):
 	files_artifact = plan.render_templates(
@@ -381,7 +408,7 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_U
 		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
 		updatedScript,
 		noInputParams,
-		currentEnclavePlan,
+		convertedEnclavePlan,
 	)
 	require.Nil(suite.T(), interpretationError)
 
@@ -406,3 +433,19 @@ func (suite *StartosisInterpreterIdempotentTestSuite) TestInterpretAndOptimize_U
 	require.Regexp(suite.T(), `assert\(value="{{kurtosis:[a-z0-9]{32}:ip_address\.runtime_value}}", assertion="==", target_value="fake_ip"\)`, scheduledInstruction4.GetInstruction().String())
 	require.True(suite.T(), scheduledInstruction4.IsExecuted()) // this instruction is not affected, i.e. it won't be re-run
 }
+
+func (suite *StartosisInterpreterIdempotentTestSuite) convertInstructionPlanToEnclavePlan(instructionPlan *instructions_plan.InstructionsPlan) *enclave_plan_persistence.EnclavePlan {
+	enclavePlan := enclave_plan_persistence.NewEnclavePlan()
+	instructionPlanSequence, interpretationErr := instructionPlan.GeneratePlan()
+	suite.Require().Nil(interpretationErr)
+	for _, instruction := range instructionPlanSequence {
+		enclavePlanInstruction, err := instruction.GetInstruction().GetPersistableAttributes().SetUuid(
+			uuid.New().String(),
+		).SetReturnedValue(
+			"None", // the returnedValue does not matter for those tests as we're testing only the interpretation phase
+		).Build()
+		suite.Require().NoError(err)
+		enclavePlan.AppendInstruction(enclavePlanInstruction)
+	}
+	return enclavePlan
+}
diff --git a/core/server/api_container/server/startosis_engine/startosis_interpreter_test.go b/core/server/api_container/server/startosis_engine/startosis_interpreter_test.go
index f916cbcb4..68f5acc70 100644
--- a/core/server/api_container/server/startosis_engine/startosis_interpreter_test.go
+++ b/core/server/api_container/server/startosis_engine/startosis_interpreter_test.go
@@ -71,7 +71,7 @@ func (suite *StartosisInterpreterTestSuite) SetupTest() {
 	suite.runtimeValueStore = runtimeValueStore
 	suite.serviceNetwork = service_network.NewMockServiceNetwork(suite.T())
 
-	suite.interpreter = NewStartosisInterpreter(suite.serviceNetwork, suite.packageContentProvider, suite.runtimeValueStore, "")
+	suite.interpreter = NewStartosisInterpreter(suite.serviceNetwork, suite.packageContentProvider, suite.runtimeValueStore, nil, "")
 
 	service.NewServiceRegistration(
 		testServiceName,