Repository: kurtosis-tech/kurtosis
Pull request #769 (2023-06-26 09:06:17)

feat: Implement idempotent run v0
## Description:
This feature saves the current state of the enclave after each starlark run and tries to optimize the next execution to execute only the instructions that changed compare to previous run. For this V0, the optimisation is "append-only". I.e. the resolution of instructions will succeed only if what changed is at the end of the instruction queue

## 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/server/startosis_engine/instructions_plan/instructions_plan.go b/core/server/api_container/server/startosis_engine/instructions_plan/instructions_plan.go
index 4036c3511..06b3d30c5 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
@@ -45,6 +45,7 @@ func (plan *InstructionsPlan) AddScheduledInstruction(scheduledInstruction *Sche
 	newScheduledInstructionUuid := scheduledInstruction.uuid
 	newScheduledInstruction := NewScheduledInstruction(newScheduledInstructionUuid, scheduledInstruction.kurtosisInstruction, scheduledInstruction.returnedValue)
 	newScheduledInstruction.Executed(scheduledInstruction.IsExecuted())
+	newScheduledInstruction.ImportedFromCurrentEnclavePlan(scheduledInstruction.IsImportedFromCurrentEnclavePlan())
 
 	plan.scheduledInstructionsIndex[newScheduledInstructionUuid] = newScheduledInstruction
 	plan.instructionsSequence = append(plan.instructionsSequence, newScheduledInstructionUuid)
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
new file mode 100644
index 000000000..8ee819f3b
--- /dev/null
+++ b/core/server/api_container/server/startosis_engine/instructions_plan/resolver/instructions_plan_mask.go
@@ -0,0 +1,43 @@
+package resolver
+
+import "github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/instructions_plan"
+
+type InstructionsPlanMask struct {
+	readIdx               int
+	scheduledInstructions []*instructions_plan.ScheduledInstruction
+	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
+	}
+}
+
+func (mask *InstructionsPlanMask) InsertAt(idx int, instruction *instructions_plan.ScheduledInstruction) {
+	mask.scheduledInstructions[idx] = instruction
+}
+
+func (mask *InstructionsPlanMask) HasNext() bool {
+	return mask.readIdx < len(mask.scheduledInstructions)
+}
+
+func (mask *InstructionsPlanMask) Next() *instructions_plan.ScheduledInstruction {
+	scheduledInstruction := mask.scheduledInstructions[mask.readIdx]
+	mask.readIdx += 1
+	return scheduledInstruction
+}
+
+func (mask *InstructionsPlanMask) Size() int {
+	return len(mask.scheduledInstructions)
+}
+
+func (mask *InstructionsPlanMask) MarkAsInvalid() {
+	mask.isValid = false
+}
+
+func (mask *InstructionsPlanMask) IsValid() bool {
+	return mask.isValid
+}
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..f233203f4 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
@@ -20,14 +20,17 @@ type ScheduledInstruction struct {
 	returnedValue starlark.Value
 
 	executed bool
+
+	importedFromCurrentEnclavePlan bool
 }
 
 func NewScheduledInstruction(uuid ScheduledInstructionUuid, kurtosisInstruction kurtosis_instruction.KurtosisInstruction, returnedValue starlark.Value) *ScheduledInstruction {
 	return &ScheduledInstruction{
-		uuid:                uuid,
-		kurtosisInstruction: kurtosisInstruction,
-		returnedValue:       returnedValue,
-		executed:            false,
+		uuid:                           uuid,
+		kurtosisInstruction:            kurtosisInstruction,
+		returnedValue:                  returnedValue,
+		executed:                       false,
+		importedFromCurrentEnclavePlan: false,
 	}
 }
 
@@ -47,3 +50,12 @@ func (instruction *ScheduledInstruction) Executed(isExecuted bool) *ScheduledIns
 func (instruction *ScheduledInstruction) IsExecuted() bool {
 	return instruction.executed
 }
+
+func (instruction *ScheduledInstruction) ImportedFromCurrentEnclavePlan(importedFromCurrentEnclavePlan bool) *ScheduledInstruction {
+	instruction.importedFromCurrentEnclavePlan = importedFromCurrentEnclavePlan
+	return instruction
+}
+
+func (instruction *ScheduledInstruction) IsImportedFromCurrentEnclavePlan() bool {
+	return instruction.importedFromCurrentEnclavePlan
+}
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 637287cbd..007a1d551 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
@@ -2,6 +2,7 @@ package plan_module
 
 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"
 	"go.starlark.net/starlark"
 	"go.starlark.net/starlarkstruct"
@@ -13,11 +14,12 @@ const (
 
 func PlanModule(
 	instructionsPlan *instructions_plan.InstructionsPlan,
+	instructionsPlanMask *resolver.InstructionsPlanMask,
 	kurtosisPlanInstructions []*kurtosis_plan_instruction.KurtosisPlanInstruction,
 ) *starlarkstruct.Module {
 	moduleBuiltins := starlark.StringDict{}
 	for _, planInstruction := range kurtosisPlanInstructions {
-		wrappedPlanInstruction := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(planInstruction, instructionsPlan)
+		wrappedPlanInstruction := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(planInstruction, instructionsPlanMask, instructionsPlan)
 		moduleBuiltins[planInstruction.GetName()] = starlark.NewBuiltin(planInstruction.GetName(), wrappedPlanInstruction.CreateBuiltin())
 	}
 
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 55942dd39..e5112e3e4 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
@@ -2,6 +2,7 @@ package kurtosis_plan_instruction
 
 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"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/startosis_errors"
 	"go.starlark.net/starlark"
@@ -20,13 +21,16 @@ type KurtosisPlanInstruction struct {
 type KurtosisPlanInstructionWrapper struct {
 	*KurtosisPlanInstruction
 
+	instructionPlanMask *resolver.InstructionsPlanMask
+
 	// TODO: This can be changed to KurtosisPlanInstructionInternal when we get rid of KurtosisInstruction
 	instructionsPlan *instructions_plan.InstructionsPlan
 }
 
-func NewKurtosisPlanInstructionWrapper(instruction *KurtosisPlanInstruction, instructionsPlan *instructions_plan.InstructionsPlan) *KurtosisPlanInstructionWrapper {
+func NewKurtosisPlanInstructionWrapper(instruction *KurtosisPlanInstruction, instructionPlanMask *resolver.InstructionsPlanMask, instructionsPlan *instructions_plan.InstructionsPlan) *KurtosisPlanInstructionWrapper {
 	return &KurtosisPlanInstructionWrapper{
 		KurtosisPlanInstruction: instruction,
+		instructionPlanMask:     instructionPlanMask,
 		instructionsPlan:        instructionsPlan,
 	}
 }
@@ -44,13 +48,34 @@ func (builtin *KurtosisPlanInstructionWrapper) CreateBuiltin() func(thread *star
 			return nil, interpretationErr
 		}
 
-		// before returning, automatically add instruction to queue
-		if err := builtin.instructionsPlan.AddInstruction(instructionWrapper, returnedFutureValue); err != nil {
-			return nil, startosis_errors.WrapWithInterpretationError(err,
-				"Unable to add Kurtosis instruction '%s' at position '%s' to the current plan being assembled. This is a Kurtosis internal bug",
-				instructionWrapper.String(),
-				instructionWrapper.GetPositionInOriginalScript().String())
+		var instructionPullFromMaskMaybe *instructions_plan.ScheduledInstruction
+		if builtin.instructionPlanMask.HasNext() {
+			instructionPullFromMaskMaybe = builtin.instructionPlanMask.Next()
+			if instructionPullFromMaskMaybe != nil && instructionPullFromMaskMaybe.GetInstruction().String() != instructionWrapper.String() {
+				// if the instructions differs, then the mask is invalid
+				builtin.instructionPlanMask.MarkAsInvalid()
+				// TODO: we could interrupt the interpretation here, because with an invalid mask the list of
+				//  instruction generated will be invalid anyway. Though we currently don't have a nive way to
+				//  interrupt an interpretation in progress (other than by throwing an error, which would be
+				//  misleading here)
+				//  To properly solve that, I think we should switch to an interactive interpretation where we
+				//  interpret each instruction one after the other, and evaluating the state after each step
+			}
+		}
+
+		if instructionPullFromMaskMaybe != nil {
+			// If there's a mask for this instruction, add the mask the plan and returned the mask's returned value
+			builtin.instructionsPlan.AddScheduledInstruction(instructionPullFromMaskMaybe).Executed(true).ImportedFromCurrentEnclavePlan(false)
+			return instructionPullFromMaskMaybe.GetReturnedValue(), nil
+		} else {
+			// 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 {
+				return nil, startosis_errors.WrapWithInterpretationError(err,
+					"Unable to add Kurtosis instruction '%s' at position '%s' to the current plan being assembled. This is a Kurtosis internal bug",
+					instructionWrapper.String(),
+					instructionWrapper.GetPositionInOriginalScript().String())
+			}
+			return returnedFutureValue, nil
 		}
-		return returnedFutureValue, nil
 	}
 }
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 2e0355e09..0e379b2f0 100644
--- a/core/server/api_container/server/startosis_engine/startosis_executor.go
+++ b/core/server/api_container/server/startosis_engine/startosis_executor.go
@@ -59,9 +59,16 @@ func (executor *StartosisExecutor) Execute(ctx context.Context, dryRun bool, par
 		// TODO: for now the plan is append only, as each Starlark run happens on top of whatever exists in the enclave
 		logrus.Debugf("Current enclave plan contains %d instuctions. About to process a new plan with %d instructions (dry-run: %v)",
 			executor.enclavePlan.Size(), len(instructionsSequence), dryRun)
+
+		executor.enclavePlan = instructions_plan.NewInstructionsPlan()
 		totalNumberOfInstructions := uint32(len(instructionsSequence))
 		for index, scheduledInstruction := range instructionsSequence {
 			instructionNumber := uint32(index + 1)
+
+			if scheduledInstruction.IsImportedFromCurrentEnclavePlan() {
+				executor.enclavePlan.AddScheduledInstruction(scheduledInstruction).Executed(true).ImportedFromCurrentEnclavePlan(true)
+				continue
+			}
 			progress := binding_constructors.NewStarlarkRunResponseLineFromSinglelineProgressInfo(
 				progressMsg, instructionNumber, totalNumberOfInstructions)
 			starlarkRunResponseLineStream <- progress
@@ -106,6 +113,10 @@ func (executor *StartosisExecutor) Execute(ctx context.Context, dryRun bool, par
 	return starlarkRunResponseLineStream
 }
 
+func (executor *StartosisExecutor) GetCurrentEnclavePLan() *instructions_plan.InstructionsPlan {
+	return executor.enclavePlan
+}
+
 func sendErrorAndFail(starlarkRunResponseLineStream chan<- *kurtosis_core_rpc_api_bindings.StarlarkRunResponseLine, err error, msg string, msgArgs ...interface{}) {
 	propagatedErr := stacktrace.Propagate(err, msg, msgArgs...)
 	serializedError := binding_constructors.NewStarlarkExecutionError(propagatedErr.Error())
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 74bcad6a8..1cd63b7a1 100644
--- a/core/server/api_container/server/startosis_engine/startosis_interpreter.go
+++ b/core/server/api_container/server/startosis_engine/startosis_interpreter.go
@@ -8,6 +8,7 @@ import (
 	"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/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/package_io"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/runtime_value_store"
@@ -73,6 +74,150 @@ func NewStartosisInterpreter(serviceNetwork service_network.ServiceNetwork, modu
 	}
 }
 
+// InterpretAndOptimizePlan is an evolution of the Interpret function which takes into account the current enclave
+// plan when  interpreting the Starlark package in order to not re-run the instructions that have alrady been run
+// inside the enclave.
+// What it does is that is calls the Interpret function with a mask that swap certain instructions with the ones from
+// the current enclave plan abd see what is affected by those swaps. If the new plan produced by the interpretation
+// with the swaps is correct, then it returns it. See detailed inline comments for more details on the implementation
+//
+// Note: the plan generated by this function is necessarily a SUBSET of the currentEnclavePlan passed as a parameter
+func (interpreter *StartosisInterpreter) InterpretAndOptimizePlan(
+	ctx context.Context,
+	packageId string,
+	mainFunctionName string,
+	serializedStarlark string,
+	serializedJsonParams string,
+	currentEnclavePlan *instructions_plan.InstructionsPlan,
+) (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
+	emptyPlanInstructionsMask := resolver.NewInstructionsPlanMask(0)
+	naiveInstructionsPlanSerializedScriptOutput, naiveInstructionsPlan, interpretationErrorApi := interpreter.Interpret(ctx, packageId, mainFunctionName, serializedStarlark, serializedJsonParams, emptyPlanInstructionsMask)
+	if interpretationErrorApi != nil {
+		return startosis_constants.NoOutputObject, nil, interpretationErrorApi
+	}
+
+	naiveInstructionsPlanSequence, interpretationErr := naiveInstructionsPlan.GeneratePlan()
+	if interpretationErr != nil {
+		return startosis_constants.NoOutputObject, nil, interpretationErr.ToAPIType()
+	}
+	logrus.Debugf("First interpretation of package generated %d instructions", len(naiveInstructionsPlanSequence))
+
+	currentEnclavePlanSequence, interpretationErr := currentEnclavePlan.GeneratePlan()
+	if interpretationErr != nil {
+		return startosis_constants.NoOutputObject, nil, interpretationErr.ToAPIType()
+	}
+	logrus.Debugf("Current enclave state contains %d instructions", len(currentEnclavePlanSequence))
+	logrus.Debugf("Starting iterations to find the best plan to execute given the current state of the enclave")
+
+	// We're going to iterate this way:
+	// 1. Find an instruction in the current enclave plan matching the first instruction of the new plan
+	// 2. Recopy all instructions prior to the match into the optimized plan
+	// 3. Recopy all following instruction from the current enclave plan into an Instructions Plan Mask -> the reason
+	//    we're naively recopying all the following instructions, not just the ones that depends on this instruction
+	//    is because right now, we don't have the ability to know which instructions depends on which. We assume that
+	//    all instructions executed AFTER this one will depend on it, to stay on the safe side
+	// 4. Run the interpretation with the mask.
+	//     - If it's successful, then we've found the optimized plan
+	//     - if it's not successful, then the mask is not compatible with the package. Go back to step 1
+	firstPossibleIndexForMatchingInstruction := 0
+	for {
+		// initialize an empty optimized plan and an empty the mask
+		potentialMask := resolver.NewInstructionsPlanMask(len(naiveInstructionsPlanSequence))
+		optimizedPlan := instructions_plan.NewInstructionsPlan()
+
+		// find the index of an instruction in the current enclave plan matching the FIRST instruction of our instructions plan generated by the first interpretation
+		matchingInstructionIdx := findFirstEqualInstructionPastIndex(currentEnclavePlanSequence, naiveInstructionsPlanSequence, firstPossibleIndexForMatchingInstruction)
+		if matchingInstructionIdx >= 0 {
+			logrus.Debugf("Found an instruction in enclave state at index %d which matches the first instruction of the new instructions plan", matchingInstructionIdx)
+			// we found a match
+			// -> First recopy all enclave state instructions prior to this match to the optimized plan. Those won't
+			// be executed, but they need to be part of the plan to keep the state of the enclave accurate
+			logrus.Debugf("Copying %d instructions from current enclave plan to new plan. Those instructions won't be executed but need to be kept in the enclave plan", matchingInstructionIdx)
+			for i := 0; i < matchingInstructionIdx; i++ {
+				optimizedPlan.AddScheduledInstruction(currentEnclavePlanSequence[i]).ImportedFromCurrentEnclavePlan(true).Executed(true)
+			}
+			// -> Then recopy all instructions past this match from the enclave state to the mask
+			// Those instructions are the instructions that will mask the instructions for the newly submitted plan
+			numberOfInstructionCopiedToMask := 0
+			for copyIdx := matchingInstructionIdx; copyIdx < len(currentEnclavePlanSequence); copyIdx++ {
+				if numberOfInstructionCopiedToMask >= potentialMask.Size() {
+					// the mask is already full, can't recopy more instructions, stop here
+					break
+				}
+				potentialMask.InsertAt(numberOfInstructionCopiedToMask, currentEnclavePlanSequence[copyIdx])
+				numberOfInstructionCopiedToMask += 1
+			}
+			logrus.Debugf("Writing %d instruction at the beginning of the plan mask, leaving %d empty at the end", numberOfInstructionCopiedToMask, potentialMask.Size()-numberOfInstructionCopiedToMask)
+		} else {
+			// We cannot find any more instructions inside the enclave state matching the first instruction of the plan
+			for _, currentPlanInstruction := range currentEnclavePlanSequence {
+				optimizedPlan.AddScheduledInstruction(currentPlanInstruction).ImportedFromCurrentEnclavePlan(true).Executed(true)
+			}
+			for _, newPlanInstruction := range naiveInstructionsPlanSequence {
+				optimizedPlan.AddScheduledInstruction(newPlanInstruction)
+			}
+			logrus.Debugf("Exhausted all possibilities. Concatenated the previous enclave plan with the new plan to obtain a %d instructions plan", optimizedPlan.Size())
+			return naiveInstructionsPlanSerializedScriptOutput, optimizedPlan, nil
+		}
+
+		// Now that we have a potential plan mask, try running interpretation again using this plan mask
+		attemptSerializedScriptOutput, attemptInstructionsPlan, interpretationErrorApi := interpreter.Interpret(ctx, packageId, mainFunctionName, serializedStarlark, serializedJsonParams, potentialMask)
+		if interpretationErrorApi != nil {
+			// Note: there's no real reason why this interpretation would fail with an error, given that the package
+			// has been interpreted once already (right above). But to be on the safe side, check the error
+			logrus.Warnf("Interpreting the package again with the plan mask failed, this is an unexpected error. " +
+				"Ignoring this mask")
+			firstPossibleIndexForMatchingInstruction += 1
+			continue
+		}
+		if !potentialMask.IsValid() {
+			// mask has been marks as invalid by the interpreter, we need to find another one
+			firstPossibleIndexForMatchingInstruction += 1
+			continue
+		}
+
+		// no error happened, it seems we found a good mask
+		// -> recopy all instructions from the interpretation to the optimized plan
+		attemptInstructionsPlanSequence, interpretationErr := attemptInstructionsPlan.GeneratePlan()
+		if interpretationErr != nil {
+			return startosis_constants.NoOutputObject, nil, interpretationErr.ToAPIType()
+		}
+
+		logrus.Debugf("Interpreting the package again with the plan mask succeeded and generated %d new instructions. Adding them to the new optimized plan", attemptInstructionsPlan.Size())
+		for _, scheduledInstruction := range attemptInstructionsPlanSequence {
+			optimizedPlan.AddScheduledInstruction(scheduledInstruction)
+		}
+
+		// there might be still be instructions in the current enclave plan that have not been imported to the
+		// optimized plan
+		// for now, we support this only if no new instructions will be executed for this run. If that's not the case
+		// continue the loop in the hope of finding another mask
+		if len(currentEnclavePlanSequence) > matchingInstructionIdx+optimizedPlan.Size() {
+			logrus.Debugf("There are %d instructions remaining in the current state that have not been transferred to the new plan. Transferring them now", len(currentEnclavePlanSequence)-matchingInstructionIdx+optimizedPlan.Size())
+			atLeastOneInstructionWillBeExecuted := false
+			for _, instructionThatWillPotentiallyBeRun := range attemptInstructionsPlanSequence {
+				if !instructionThatWillPotentiallyBeRun.IsExecuted() {
+					atLeastOneInstructionWillBeExecuted = true
+				}
+			}
+			if atLeastOneInstructionWillBeExecuted {
+				logrus.Debugf("The remaining instructions in the current enclave plan cannot be transferred to the new plan because this plan contains instructions that will be executed." +
+					"The remaining instructions might depend on those and Kurtosis cannot re-run them (this is unsupported for now)")
+				continue
+			}
+			// recopy all remaining instructions into the optimized plan
+			for _, remainingInstructionFromCurrentEnclaveState := range currentEnclavePlanSequence[matchingInstructionIdx+optimizedPlan.Size():] {
+				optimizedPlan.AddScheduledInstruction(remainingInstructionFromCurrentEnclaveState).ImportedFromCurrentEnclavePlan(true).Executed(true)
+			}
+		}
+		// finally we can return the optimized plan as well as the serialized script output returned by the last
+		// interpretation attempt
+		return attemptSerializedScriptOutput, optimizedPlan, nil
+	}
+}
+
 // Interpret interprets the Starlark script and produce different outputs:
 //   - A potential interpretation error that the writer of the script should be aware of (syntax error in the Startosis
 //     code, inconsistent). Can be nil if the script was successfully interpreted
@@ -84,6 +229,7 @@ func (interpreter *StartosisInterpreter) Interpret(
 	mainFunctionName string,
 	serializedStarlark string,
 	serializedJsonParams string,
+	instructionsPlanMask *resolver.InstructionsPlanMask,
 ) (string, *instructions_plan.InstructionsPlan, *kurtosis_core_rpc_api_bindings.StarlarkInterpretationError) {
 	interpreter.mutex.Lock()
 	defer interpreter.mutex.Unlock()
@@ -110,13 +256,13 @@ func (interpreter *StartosisInterpreter) Interpret(
 	mainFunction, ok := globalVariables[mainFunctionName].(*starlark.Function)
 	// if there is an element with the `mainFunctionName` but it isn't a function we have to error as well
 	if !ok {
-		return "", nil, missingMainFunctionError(packageId, mainFunctionName)
+		return startosis_constants.NoOutputObject, nil, missingMainFunctionError(packageId, mainFunctionName)
 	}
 
 	runFunctionExecutionThread := newStarlarkThread(starlarkGoThreadName)
 
 	if isUsingDefaultMainFunction && mainFunction.NumParams() > maximumParamsAllowedForRunFunction {
-		return "", nil, startosis_errors.NewInterpretationError("The 'run' entrypoint function can have at most '%v' argument got '%v'", maximumParamsAllowedForRunFunction, mainFunction.NumParams()).ToAPIType()
+		return startosis_constants.NoOutputObject, nil, startosis_errors.NewInterpretationError("The 'run' entrypoint function can have at most '%v' argument got '%v'", maximumParamsAllowedForRunFunction, mainFunction.NumParams()).ToAPIType()
 	}
 
 	var argsTuple starlark.Tuple
@@ -129,12 +275,12 @@ func (interpreter *StartosisInterpreter) Interpret(
 		firstParamName, _ := mainFunction.Param(planParamIndex)
 		if firstParamName == planParamName {
 			kurtosisPlanInstructions := KurtosisPlanInstructions(interpreter.serviceNetwork, interpreter.recipeExecutor, interpreter.moduleContentProvider)
-			planModule := plan_module.PlanModule(newInstructionsPlan, kurtosisPlanInstructions)
+			planModule := plan_module.PlanModule(newInstructionsPlan, instructionsPlanMask, kurtosisPlanInstructions)
 			argsTuple = append(argsTuple, planModule)
 		}
 
 		if firstParamName != planParamName && isUsingDefaultMainFunction {
-			return "", nil, startosis_errors.NewInterpretationError(unexpectedArgNameError, planParamIndex, planParamName, firstParamName).ToAPIType()
+			return startosis_constants.NoOutputObject, nil, startosis_errors.NewInterpretationError(unexpectedArgNameError, planParamIndex, planParamName, firstParamName).ToAPIType()
 		}
 	}
 
@@ -142,13 +288,13 @@ func (interpreter *StartosisInterpreter) Interpret(
 		(!isUsingDefaultMainFunction && mainFuncParamsNum > 0) {
 		if isUsingDefaultMainFunction {
 			if paramName, _ := mainFunction.Param(argsParamIndex); paramName != argsParamName {
-				return "", nil, startosis_errors.NewInterpretationError(unexpectedArgNameError, argsParamIndex, argsParamName, paramName).ToAPIType()
+				return startosis_constants.NoOutputObject, nil, startosis_errors.NewInterpretationError(unexpectedArgNameError, argsParamIndex, argsParamName, paramName).ToAPIType()
 			}
 		}
 		// run function has an argument so we parse input args
 		inputArgs, interpretationError := interpreter.parseInputArgs(runFunctionExecutionThread, serializedJsonParams)
 		if interpretationError != nil {
-			return "", nil, interpretationError.ToAPIType()
+			return startosis_constants.NoOutputObject, nil, interpretationError.ToAPIType()
 		}
 		if isUsingDefaultMainFunction {
 			argsTuple = append(argsTuple, inputArgs)
@@ -156,7 +302,7 @@ func (interpreter *StartosisInterpreter) Interpret(
 		} else {
 			argsDict, ok := inputArgs.(*starlark.Dict)
 			if !ok {
-				return "", nil, startosis_errors.NewInterpretationError("An error occurred casting input args '%s' to Starlark Dict", inputArgs).ToAPIType()
+				return startosis_constants.NoOutputObject, nil, startosis_errors.NewInterpretationError("An error occurred casting input args '%s' to Starlark Dict", inputArgs).ToAPIType()
 			}
 			kwArgs = append(kwArgs, argsDict.Items()...)
 		}
@@ -164,7 +310,7 @@ func (interpreter *StartosisInterpreter) Interpret(
 
 	outputObject, err := starlark.Call(runFunctionExecutionThread, mainFunction, argsTuple, kwArgs)
 	if err != nil {
-		return "", nil, generateInterpretationError(err).ToAPIType()
+		return startosis_constants.NoOutputObject, nil, generateInterpretationError(err).ToAPIType()
 	}
 
 	// Serialize and return the output object. It might contain magic strings that should be resolved post-execution
@@ -233,6 +379,18 @@ func (interpreter *StartosisInterpreter) buildBindings(instructionPlan *instruct
 	return &predeclared, nil
 }
 
+func findFirstEqualInstructionPastIndex(currentEnclaveInstructionsList []*instructions_plan.ScheduledInstruction, 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++ {
+		if currentEnclaveInstructionsList[i].GetInstruction().String() == naiveInstructionsList[0].GetInstruction().String() {
+			return i
+		}
+	}
+	return -1 // no match
+}
+
 // This method handles the different cases a Startosis module can be executed.
 // - If input args are empty it uses empty JSON ({}) as the input args
 // - If input args aren't empty it tries to deserialize them
diff --git a/core/server/api_container/server/startosis_engine/startosis_runner.go b/core/server/api_container/server/startosis_engine/startosis_runner.go
index 43ef36722..7fa3ea566 100644
--- a/core/server/api_container/server/startosis_engine/startosis_runner.go
+++ b/core/server/api_container/server/startosis_engine/startosis_runner.go
@@ -4,6 +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/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/starlark_warning"
 	"github.com/sirupsen/logrus"
 	"strings"
@@ -87,7 +89,31 @@ func (runner *StartosisRunner) Run(
 			startingInterpretationMsg, defaultCurrentStepNumber, defaultTotalStepsNumber)
 		starlarkRunResponseLines <- progressInfo
 
-		serializedScriptOutput, instructionsPlan, interpretationError := runner.startosisInterpreter.Interpret(ctx, packageId, mainFunctionName, serializedStartosis, serializedParams)
+		// TODO: once we have feature flags, add a switch here to call InterpretAndOptimizePlan if the feature flag is
+		//  turned on
+		var serializedScriptOutput string
+		var instructionsPlan *instructions_plan.InstructionsPlan
+		var interpretationError *kurtosis_core_rpc_api_bindings.StarlarkInterpretationError
+		if doesFeatureFlagsContain(experimentalFeatures, kurtosis_core_rpc_api_bindings.KurtosisFeatureFlag_USE_INSTRUCTIONS_CACHING) {
+			serializedScriptOutput, instructionsPlan, interpretationError = runner.startosisInterpreter.InterpretAndOptimizePlan(
+				ctx,
+				packageId,
+				mainFunctionName,
+				serializedStartosis,
+				serializedParams,
+				runner.startosisExecutor.enclavePlan,
+			)
+		} else {
+			serializedScriptOutput, instructionsPlan, interpretationError = runner.startosisInterpreter.Interpret(
+				ctx,
+				packageId,
+				mainFunctionName,
+				serializedStartosis,
+				serializedParams,
+				resolver.NewInstructionsPlanMask(0),
+			)
+		}
+
 		if interpretationError != nil {
 			starlarkRunResponseLines <- binding_constructors.NewStarlarkRunResponseLineFromInterpretationError(interpretationError)
 			starlarkRunResponseLines <- binding_constructors.NewStarlarkRunResponseLineFromRunFailureEvent()
@@ -142,3 +168,12 @@ func forwardKurtosisResponseLineChannelUntilSourceIsClosed(sourceChan <-chan *ku
 	logrus.Debugf("Kurtosis instructions stream was closed. Exiting execution loop. Run finishedL '%v'", isStarlarkRunFinished)
 	return isStarlarkRunFinished
 }
+
+func doesFeatureFlagsContain(featureFlags []kurtosis_core_rpc_api_bindings.KurtosisFeatureFlag, requestedFeatureFlag kurtosis_core_rpc_api_bindings.KurtosisFeatureFlag) bool {
+	for _, featureFlag := range featureFlags {
+		if featureFlag == requestedFeatureFlag {
+			return true
+		}
+	}
+	return false
+}
diff --git a/core/server/api_container/server/startosis_engine/startosis_validator.go b/core/server/api_container/server/startosis_engine/startosis_validator.go
index 16d6a8033..da92de7c8 100644
--- a/core/server/api_container/server/startosis_engine/startosis_validator.go
+++ b/core/server/api_container/server/startosis_engine/startosis_validator.go
@@ -80,7 +80,7 @@ func (validator *StartosisValidator) Validate(ctx context.Context, instructionsS
 func (validator *StartosisValidator) validateAndUpdateEnvironment(instructionsSequence []*instructions_plan.ScheduledInstruction, environment *startosis_validator.ValidatorEnvironment, starlarkRunResponseLineStream chan *kurtosis_core_rpc_api_bindings.StarlarkRunResponseLine) bool {
 	isValidationFailure := false
 	for _, scheduledInstruction := range instructionsSequence {
-		if scheduledInstruction.IsExecuted() {
+		if scheduledInstruction.IsImportedFromCurrentEnclavePlan() || scheduledInstruction.IsExecuted() {
 			// no need to validate the instruction as it won't be executed in this round
 			continue
 		}


Test patch:
diff --git a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/starlark_framework_engine_test.go b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/starlark_framework_engine_test.go
index 2f11b25d7..da7c1ac11 100644
--- a/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/starlark_framework_engine_test.go
+++ b/core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/starlark_framework_engine_test.go
@@ -6,6 +6,7 @@ import (
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine"
 	"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/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/builtin_argument"
 	"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/startosis_constants"
@@ -79,7 +80,8 @@ func testKurtosisPlanInstruction(t *testing.T, builtin KurtosisPlanInstructionBa
 	predeclared := getBasePredeclaredDict(t)
 	// Add the KurtosisPlanInstruction that is being tested
 	instructionFromBuiltin := builtin.GetInstruction()
-	instructionWrapper := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(instructionFromBuiltin, instructionsPlan)
+	emptyInstructionsPlanMask := resolver.NewInstructionsPlanMask(0)
+	instructionWrapper := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(instructionFromBuiltin, emptyInstructionsPlanMask, instructionsPlan)
 	predeclared[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 aacd279c3..cb2213da4 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
@@ -38,33 +38,39 @@ func TestExecuteKurtosisInstructions_ExecuteForReal_Success(t *testing.T) {
 
 	instructionsPlan := instructions_plan.NewInstructionsPlan()
 	instruction0 := createMockInstruction(t, "instruction0", executeSuccessfully)
-	scheduledInstruction0 := instructions_plan.NewScheduledInstruction("instruction0", instruction0, starlark.None).Executed(true)
+	scheduledInstruction0 := instructions_plan.NewScheduledInstruction("instruction0", instruction0, starlark.None).Executed(true).ImportedFromCurrentEnclavePlan(true)
 	instructionsPlan.AddScheduledInstruction(scheduledInstruction0)
 
 	instruction1 := createMockInstruction(t, "instruction1", executeSuccessfully)
+	scheduledInstruction1 := instructions_plan.NewScheduledInstruction("instruction1", instruction1, starlark.None).Executed(true)
+	instructionsPlan.AddScheduledInstruction(scheduledInstruction1)
+
 	instruction2 := createMockInstruction(t, "instruction2", executeSuccessfully)
-	require.NoError(t, instructionsPlan.AddInstruction(instruction1, starlark.None))
+	instruction3 := createMockInstruction(t, "instruction3", executeSuccessfully)
 	require.NoError(t, instructionsPlan.AddInstruction(instruction2, starlark.None))
+	require.NoError(t, instructionsPlan.AddInstruction(instruction3, starlark.None))
 
 	require.Equal(t, executor.enclavePlan.Size(), 0) // check that the enclave plan is empty prior to execution
 
 	_, serializedInstruction, err := executeSynchronously(t, executor, executeForReal, instructionsPlan)
-	instruction0.AssertNumberOfCalls(t, "GetCanonicalInstruction", 1)
-	instruction0.AssertNumberOfCalls(t, "Execute", 0) // not executed as it was already executed
+	instruction0.AssertNumberOfCalls(t, "GetCanonicalInstruction", 0) // skipped directly
+	instruction0.AssertNumberOfCalls(t, "Execute", 0)
 	instruction1.AssertNumberOfCalls(t, "GetCanonicalInstruction", 1)
-	instruction1.AssertNumberOfCalls(t, "Execute", 1)
+	instruction1.AssertNumberOfCalls(t, "Execute", 0) // not executed as it was already executed
 	instruction2.AssertNumberOfCalls(t, "GetCanonicalInstruction", 1)
 	instruction2.AssertNumberOfCalls(t, "Execute", 1)
+	instruction3.AssertNumberOfCalls(t, "GetCanonicalInstruction", 1)
+	instruction3.AssertNumberOfCalls(t, "Execute", 1)
 
 	require.Nil(t, err)
 
 	expectedSerializedInstructions := []*kurtosis_core_rpc_api_bindings.StarlarkInstruction{
-		binding_constructors.NewStarlarkInstruction(dummyPosition.ToAPIType(), "instruction0", "instruction0()", noInstructionArgsForTesting),
 		binding_constructors.NewStarlarkInstruction(dummyPosition.ToAPIType(), "instruction1", "instruction1()", noInstructionArgsForTesting),
 		binding_constructors.NewStarlarkInstruction(dummyPosition.ToAPIType(), "instruction2", "instruction2()", noInstructionArgsForTesting),
+		binding_constructors.NewStarlarkInstruction(dummyPosition.ToAPIType(), "instruction3", "instruction3()", noInstructionArgsForTesting),
 	}
 	require.Equal(t, expectedSerializedInstructions, serializedInstruction)
-	require.Equal(t, executor.enclavePlan.Size(), 3) // check that the enclave plan now contains the 3 instructions
+	require.Equal(t, executor.enclavePlan.Size(), 4) // check that the enclave plan now contains the 4 instructions
 }
 
 func TestExecuteKurtosisInstructions_ExecuteForReal_FailureHalfWay(t *testing.T) {
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
new file mode 100644
index 000000000..a55ab8284
--- /dev/null
+++ b/core/server/api_container/server/startosis_engine/startosis_interpreter_idempotent_test.go
@@ -0,0 +1,326 @@
+package startosis_engine
+
+import (
+	"context"
+	"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"
+	"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/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"
+	"go.starlark.net/starlark"
+	"testing"
+)
+
+const (
+	noInputParams = "{}"
+	noReturnValue = starlark.None
+)
+
+// Most simple case - replay the same package twice
+// Current plan ->     [`print("instruction1")`  `print("instruction2")`  `print("instruction3")`]
+// Package to run ->   [`print("instruction1")`  `print("instruction2")`  `print("instruction3")`]
+// Check that this results in the entire set of instruction being skipped
+func TestInterpretAndOptimize_IdenticalPackage(t *testing.T) {
+	packageContentProvider := mock_package_content_provider.NewMockPackageContentProvider()
+	defer packageContentProvider.RemoveAll()
+	runtimeValueStore := runtime_value_store.NewRuntimeValueStore()
+	interpreter := NewStartosisInterpreter(testServiceNetwork, packageContentProvider, runtimeValueStore)
+	script := `
+def run(plan, args):
+	plan.print("instruction1")
+	plan.print("instruction2")
+	plan.print("instruction3")
+`
+
+	instruction1Str := `print(msg="instruction1")`
+	instruction1 := newMockInstruction(t, instruction1Str)
+	instruction2Str := `print(msg="instruction2")`
+	instruction2 := newMockInstruction(t, instruction2Str)
+	instruction3Str := `print(msg="instruction3")`
+	instruction3 := newMockInstruction(t, instruction3Str)
+
+	currentEnclavePlan := instructions_plan.NewInstructionsPlan()
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction1, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction2, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction3, noReturnValue))
+
+	_, instructionsPlan, interpretationError := interpreter.InterpretAndOptimizePlan(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		useDefaultMainFunctionName,
+		script,
+		noInputParams,
+		currentEnclavePlan,
+	)
+	require.Nil(t, interpretationError)
+
+	instructionSequence, err := instructionsPlan.GeneratePlan()
+	require.Nil(t, err)
+	require.Equal(t, 3, len(instructionSequence))
+
+	scheduledInstruction1 := instructionSequence[0]
+	require.Equal(t, instruction1Str, scheduledInstruction1.GetInstruction().String())
+	require.False(t, scheduledInstruction1.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction1.IsExecuted())
+
+	scheduledInstruction2 := instructionSequence[1]
+	require.Equal(t, instruction2Str, scheduledInstruction2.GetInstruction().String())
+	require.False(t, scheduledInstruction2.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction2.IsExecuted())
+
+	scheduledInstruction3 := instructionSequence[2]
+	require.Equal(t, instruction3Str, scheduledInstruction3.GetInstruction().String())
+	require.False(t, scheduledInstruction3.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction3.IsExecuted())
+}
+
+// Add an instruction at the end of a package that was already run
+// Current plan ->     [`print("instruction1")`  `print("instruction2")`                         ]
+// Package to run ->   [`print("instruction1")`  `print("instruction2")`  `print("instruction3")`]
+// Check that the first two instructions are a;ready executed, and the last one is in the new plan marked as not
+// executed
+func TestInterpretAndOptimize_AppendNewInstruction(t *testing.T) {
+	packageContentProvider := mock_package_content_provider.NewMockPackageContentProvider()
+	defer packageContentProvider.RemoveAll()
+	runtimeValueStore := runtime_value_store.NewRuntimeValueStore()
+	interpreter := NewStartosisInterpreter(testServiceNetwork, packageContentProvider, runtimeValueStore)
+	script := `
+def run(plan, args):
+	plan.print("instruction1")
+	plan.print("instruction2")
+	plan.print("instruction3")
+`
+
+	instruction1Str := `print(msg="instruction1")`
+	instruction1 := newMockInstruction(t, instruction1Str)
+	instruction2Str := `print(msg="instruction2")`
+	instruction2 := newMockInstruction(t, instruction2Str)
+
+	currentEnclavePlan := instructions_plan.NewInstructionsPlan()
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction1, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction2, noReturnValue))
+
+	_, instructionsPlan, interpretationError := interpreter.InterpretAndOptimizePlan(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		useDefaultMainFunctionName,
+		script,
+		noInputParams,
+		currentEnclavePlan,
+	)
+	require.Nil(t, interpretationError)
+
+	instructionSequence, err := instructionsPlan.GeneratePlan()
+	require.Nil(t, err)
+	require.Equal(t, 3, len(instructionSequence))
+
+	scheduledInstruction1 := instructionSequence[0]
+	require.Equal(t, instruction1Str, scheduledInstruction1.GetInstruction().String())
+	require.False(t, scheduledInstruction1.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction1.IsExecuted())
+
+	scheduledInstruction2 := instructionSequence[1]
+	require.Equal(t, instruction2Str, scheduledInstruction2.GetInstruction().String())
+	require.False(t, scheduledInstruction2.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction2.IsExecuted())
+
+	scheduledInstruction3 := instructionSequence[2]
+	require.Equal(t, `print(msg="instruction3")`, scheduledInstruction3.GetInstruction().String())
+	require.False(t, scheduledInstruction3.IsImportedFromCurrentEnclavePlan())
+	require.False(t, scheduledInstruction3.IsExecuted())
+}
+
+// Run an instruction inside an enclave that is not empty (other non-related package were run in the past)
+// Current plan ->     [`print("instruction1")`  `print("instruction2")`                         ]
+// Package to run ->   [                                                  `print("instruction3")`]
+// Check that the first two instructions are marked as imported from a previous plan, already executed, and the last
+// one is in the new plan marked as not executed
+func TestInterpretAndOptimize_DisjointInstructionSet(t *testing.T) {
+	packageContentProvider := mock_package_content_provider.NewMockPackageContentProvider()
+	defer packageContentProvider.RemoveAll()
+	runtimeValueStore := runtime_value_store.NewRuntimeValueStore()
+	interpreter := NewStartosisInterpreter(testServiceNetwork, packageContentProvider, runtimeValueStore)
+	script := `
+def run(plan, args):
+	plan.print("instruction3")
+`
+
+	instruction1Str := `print(msg="instruction1")`
+	instruction1 := newMockInstruction(t, instruction1Str)
+	instruction2Str := `print(msg="instruction2")`
+	instruction2 := newMockInstruction(t, instruction2Str)
+
+	currentEnclavePlan := instructions_plan.NewInstructionsPlan()
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction1, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction2, noReturnValue))
+
+	_, instructionsPlan, interpretationError := interpreter.InterpretAndOptimizePlan(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		useDefaultMainFunctionName,
+		script,
+		noInputParams,
+		currentEnclavePlan,
+	)
+	require.Nil(t, interpretationError)
+
+	instructionSequence, err := instructionsPlan.GeneratePlan()
+	require.Nil(t, err)
+	require.Equal(t, 3, len(instructionSequence))
+
+	scheduledInstruction1 := instructionSequence[0]
+	require.Equal(t, instruction1Str, scheduledInstruction1.GetInstruction().String())
+	require.True(t, scheduledInstruction1.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction1.IsExecuted())
+
+	scheduledInstruction2 := instructionSequence[1]
+	require.Equal(t, instruction2Str, scheduledInstruction2.GetInstruction().String())
+	require.True(t, scheduledInstruction2.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction2.IsExecuted())
+
+	scheduledInstruction3 := instructionSequence[2]
+	require.Equal(t, `print(msg="instruction3")`, scheduledInstruction3.GetInstruction().String())
+	require.False(t, scheduledInstruction3.IsImportedFromCurrentEnclavePlan())
+	require.False(t, scheduledInstruction3.IsExecuted())
+}
+
+// This is a bit of an edge case here, we run only part of the instruction that are identical to what was already run
+// Current plan ->     [`print("instruction1")`  `print("instruction2")`  `print("instruction3")`]
+// Package to run ->   [`print("instruction1")`  `print("instruction2")`                         ]
+// Check that instruction 1 and 2 are part of the new plan, already executed, and instruction3 is ALSO part of the
+// plan, but marked as imported from a previous plan
+func TestInterpretAndOptimize_ReplacePartOfInstructionWithIdenticalInstruction(t *testing.T) {
+	packageContentProvider := mock_package_content_provider.NewMockPackageContentProvider()
+	defer packageContentProvider.RemoveAll()
+	runtimeValueStore := runtime_value_store.NewRuntimeValueStore()
+	interpreter := NewStartosisInterpreter(testServiceNetwork, packageContentProvider, runtimeValueStore)
+	script := `
+def run(plan, args):
+	plan.print(msg="instruction1")
+	plan.print(msg="instruction2")
+`
+
+	instruction1Str := `print(msg="instruction1")`
+	instruction1 := newMockInstruction(t, instruction1Str)
+	instruction2Str := `print(msg="instruction2")`
+	instruction2 := newMockInstruction(t, instruction2Str)
+	instruction3Str := `print(msg="instruction3")`
+	instruction3 := newMockInstruction(t, instruction3Str)
+
+	currentEnclavePlan := instructions_plan.NewInstructionsPlan()
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction1, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction2, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction3, noReturnValue))
+
+	_, instructionsPlan, interpretationError := interpreter.InterpretAndOptimizePlan(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		useDefaultMainFunctionName,
+		script,
+		noInputParams,
+		currentEnclavePlan,
+	)
+	require.Nil(t, interpretationError)
+
+	instructionSequence, err := instructionsPlan.GeneratePlan()
+	require.Nil(t, err)
+	require.Equal(t, 3, len(instructionSequence))
+
+	scheduledInstruction1 := instructionSequence[0]
+	require.Equal(t, instruction1Str, scheduledInstruction1.GetInstruction().String())
+	require.False(t, scheduledInstruction1.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction1.IsExecuted())
+
+	scheduledInstruction2 := instructionSequence[1]
+	require.Equal(t, instruction2Str, scheduledInstruction2.GetInstruction().String())
+	require.False(t, scheduledInstruction2.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction2.IsExecuted())
+
+	scheduledInstruction3 := instructionSequence[2]
+	require.Equal(t, `print(msg="instruction3")`, scheduledInstruction3.GetInstruction().String())
+	require.True(t, scheduledInstruction3.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction3.IsExecuted())
+}
+
+// Submit a package with a update on an instruction located "in the middle" of the package
+// Current plan ->     [`print("instruction1")`  `print("instruction2")`      `print("instruction3")`]
+// Package to run ->   [`print("instruction1")`  `print("instruction2_NEW")`  `print("instruction3")`]
+// That will result in the concatenation of the two plans because we're not able to properly resolve dependencies yet
+// [`print("instruction1")`  `print("instruction2")`  `print("instruction3")`  `print("instruction1")`  `print("instruction2_NEW")`  `print("instruction3")`]
+// The first three are imported from a previous plan, already executed, while the last three are from all new
+func TestInterpretAndOptimize_InvalidNewVersionOfThePackage(t *testing.T) {
+	packageContentProvider := mock_package_content_provider.NewMockPackageContentProvider()
+	defer packageContentProvider.RemoveAll()
+	runtimeValueStore := runtime_value_store.NewRuntimeValueStore()
+	interpreter := NewStartosisInterpreter(testServiceNetwork, packageContentProvider, runtimeValueStore)
+	script := `
+def run(plan, args):
+	plan.print(msg="instruction1")
+	plan.print(msg="instruction2_NEW")
+	plan.print(msg="instruction3")
+`
+
+	instruction1Str := `print(msg="instruction1")`
+	instruction1 := newMockInstruction(t, instruction1Str)
+	instruction2Str := `print(msg="instruction2")`
+	instruction2 := newMockInstruction(t, instruction2Str)
+	instruction3Str := `print(msg="instruction3")`
+	instruction3 := newMockInstruction(t, instruction3Str)
+
+	currentEnclavePlan := instructions_plan.NewInstructionsPlan()
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction1, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction2, noReturnValue))
+	require.Nil(t, currentEnclavePlan.AddInstruction(instruction3, noReturnValue))
+
+	_, instructionsPlan, interpretationError := interpreter.InterpretAndOptimizePlan(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		useDefaultMainFunctionName,
+		script,
+		noInputParams,
+		currentEnclavePlan,
+	)
+	require.Nil(t, interpretationError)
+
+	instructionSequence, err := instructionsPlan.GeneratePlan()
+	require.Nil(t, err)
+	require.Equal(t, 6, len(instructionSequence))
+
+	scheduledInstruction1 := instructionSequence[0]
+	require.Equal(t, instruction1Str, scheduledInstruction1.GetInstruction().String())
+	require.True(t, scheduledInstruction1.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction1.IsExecuted())
+
+	scheduledInstruction2 := instructionSequence[1]
+	require.Equal(t, instruction2Str, scheduledInstruction2.GetInstruction().String())
+	require.True(t, scheduledInstruction2.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction2.IsExecuted())
+
+	scheduledInstruction3 := instructionSequence[2]
+	require.Equal(t, instruction3Str, scheduledInstruction3.GetInstruction().String())
+	require.True(t, scheduledInstruction3.IsImportedFromCurrentEnclavePlan())
+	require.True(t, scheduledInstruction3.IsExecuted())
+
+	scheduledInstruction4 := instructionSequence[3]
+	require.Equal(t, instruction1Str, scheduledInstruction4.GetInstruction().String())
+	require.False(t, scheduledInstruction4.IsImportedFromCurrentEnclavePlan())
+	require.False(t, scheduledInstruction4.IsExecuted())
+
+	scheduledInstruction5 := instructionSequence[4]
+	require.Equal(t, `print(msg="instruction2_NEW")`, scheduledInstruction5.GetInstruction().String())
+	require.False(t, scheduledInstruction5.IsImportedFromCurrentEnclavePlan())
+	require.False(t, scheduledInstruction5.IsExecuted())
+
+	scheduledInstruction6 := instructionSequence[5]
+	require.Equal(t, instruction3Str, scheduledInstruction6.GetInstruction().String())
+	require.False(t, scheduledInstruction6.IsImportedFromCurrentEnclavePlan())
+	require.False(t, scheduledInstruction6.IsExecuted())
+}
+
+func newMockInstruction(t *testing.T, instructionName string) kurtosis_instruction.KurtosisInstruction {
+	mockInstruction := mock_instruction.NewMockKurtosisInstruction(t)
+	mockInstruction.EXPECT().String().Maybe().Return(instructionName)
+	return mockInstruction
+}
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 f64193ce8..baaaa4da7 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
@@ -9,6 +9,7 @@ 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/print_builtin"
 	"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/add_service"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/kurtosis_print"
 	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/remove_service"
@@ -29,6 +30,10 @@ var (
 	testServiceIpAddress = net.ParseIP("127.0.0.1")
 )
 
+var (
+	emptyInstructionsPlanMask = resolver.NewInstructionsPlanMask(0)
+)
+
 const (
 	testServiceName        = service.ServiceName("example-datastore-server")
 	testContainerImageName = "kurtosistech/example-datastore-server"
@@ -49,7 +54,7 @@ def run(plan):
 	plan.print("` + testString + `")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size()) // Only the print statement
 
@@ -76,7 +81,7 @@ def deploy_contract(plan,service_name,contract_name,init_message,args):
 	mainFunctionName := "deploy_contract"
 	inputArgs := `{"service_name": "my-service", "contract_name": "my-contract", "init_message": "Init message", "args": {"arg1": "arg1-value", "arg2": "arg2-value"}}`
 
-	result, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, mainFunctionName, script, inputArgs)
+	result, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, mainFunctionName, script, inputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 3, instructionsPlan.Size()) // The three print functions
 	require.NotNil(t, result)
@@ -103,7 +108,7 @@ def my_func(my_arg1, my_arg2, args):
 	mainFunctionName := "my_func"
 	inputArgs := `{"my_arg1": "foo", "my_arg2": "bar", "args": {"arg1": "arg1-value", "arg2": "arg2-value"}}`
 
-	result, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, mainFunctionName, script, inputArgs)
+	result, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, mainFunctionName, script, inputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 0, instructionsPlan.Size()) // There are no instructions to execute
 	require.NotNil(t, result)
@@ -125,7 +130,7 @@ def run(plan):
 	plan.print(my_dict)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 2, instructionsPlan.Size())
 
@@ -147,7 +152,7 @@ def run(plan):
 unknownInstruction()
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 
 	expectedError := startosis_errors.NewInterpretationErrorWithCustomMsg(
 		[]startosis_errors.CallFrame{
@@ -172,7 +177,7 @@ unknownVariable
 unknownInstruction2()
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 
 	expectedError := startosis_errors.NewInterpretationErrorWithCustomMsg(
 		[]startosis_errors.CallFrame{
@@ -198,7 +203,7 @@ def run():
 load("otherScript.start") # fails b/c load takes in at least 2 args
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 
 	expectedError := startosis_errors.NewInterpretationErrorFromStacktrace(
 		[]startosis_errors.CallFrame{
@@ -234,7 +239,7 @@ def run(plan):
 	plan.print("The grpc transport protocol is " + datastore_service.ports["grpc"].transport_protocol)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, fmt.Sprintf(script, testServiceName), startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, fmt.Sprintf(script, testServiceName), startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 5, instructionsPlan.Size())
 
@@ -273,7 +278,7 @@ def run(plan):
 	plan.print("The transport protocol is " + datastore_service.ports["grpc"].transport_protocol)
 	plan.print("The application protocol is " + datastore_service.ports["grpc"].application_protocol)
 `
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, fmt.Sprintf(script, testServiceName), startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, fmt.Sprintf(script, testServiceName), startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 6, instructionsPlan.Size())
 
@@ -307,7 +312,7 @@ def run(plan):
 	plan.add_service(name = service_name, config = config)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 
 	expectedError := startosis_errors.NewInterpretationErrorWithCauseAndCustomMsg(
 		errors.New("ServiceConfig: missing argument for image"),
@@ -342,7 +347,7 @@ def run(plan):
 	plan.add_service(name = service_name, config = config)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	expectedError := startosis_errors.NewInterpretationErrorWithCauseAndCustomMsg(
 		startosis_errors.NewInterpretationError(`The following argument(s) could not be parsed or did not pass validation: {"transport_protocol":"Invalid argument value for 'transport_protocol': 'TCPK'. Valid values are TCP, SCTP, UDP"}`),
 		[]startosis_errors.CallFrame{
@@ -376,7 +381,7 @@ def run(plan):
 	plan.add_service(name = service_name, config = config)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	expectedError := startosis_errors.NewInterpretationErrorWithCauseAndCustomMsg(
 		startosis_errors.NewInterpretationError(`The following argument(s) could not be parsed or did not pass validation: {"number":"Value for 'number' was expected to be an integer between 1 and 65535, but it was 'starlark.String'"}`),
 		[]startosis_errors.CallFrame{
@@ -420,7 +425,7 @@ def run(plan):
 	plan.print("Done!")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 8, instructionsPlan.Size())
 
@@ -452,7 +457,7 @@ load("` + barModulePath + `", "a")
 def run(plan):
 	plan.print("Hello " + a)
 `
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	expectedError := startosis_errors.NewInterpretationErrorWithCustomMsg(
 		[]startosis_errors.CallFrame{
 			*startosis_errors.NewCallFrame("<toplevel>", startosis_errors.NewScriptPosition(startosis_constants.PackageIdPlaceholderForStandaloneScript, 2, 1)),
@@ -479,7 +484,7 @@ my_module = import_module("` + barModulePath + `")
 def run(plan):
 	plan.print("Hello " + my_module.a)
 `
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size()) // Only the print statement
 
@@ -508,7 +513,7 @@ def run(plan):
 
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size())
 
@@ -536,7 +541,7 @@ def run(plan):
 	plan.print(module_doo.b)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	expectedError := startosis_errors.NewInterpretationErrorWithCustomMsg(
 		[]startosis_errors.CallFrame{
 			*startosis_errors.NewCallFrame("<toplevel>", startosis_errors.NewScriptPosition(moduleBarLoadsModuleDoo, 1, 27)),
@@ -561,7 +566,7 @@ def run(plan):
 	plan.print(my_module.b)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 
 	errorMsg := `Evaluation error: An error occurred while loading the module '` + nonExistentModule + `'
 	Caused by: Package '` + nonExistentModule + `' not found`
@@ -588,7 +593,7 @@ def run(plan):
 `
 
 	// assert that first load fails
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.NotNil(t, interpretationError)
 	require.Nil(t, instructionsPlan)
 
@@ -597,7 +602,7 @@ def run(plan):
 	expectedOutput := `Hello World!
 `
 	// assert that second load succeeds
-	_, instructionsPlan, interpretationError = interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError = interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size())
 	validateScriptOutputFromPrintInstructions(t, instructionsPlan, expectedOutput)
@@ -628,7 +633,7 @@ def run(plan):
 	plan.add_service(name = module_bar.service_name, config = module_bar.config)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 3, instructionsPlan.Size())
 
@@ -676,7 +681,7 @@ def run(plan):
 	plan.print("Done!")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 8, instructionsPlan.Size())
 
@@ -709,7 +714,7 @@ def run(plan):
 	plan.print("World!")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size())
 
@@ -751,7 +756,7 @@ Adding service example-datastore-server
 Starting Startosis script!
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, scriptA, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, scriptA, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 4, instructionsPlan.Size())
 	assertInstructionTypeAndPosition(t, instructionsPlan, 2, add_service.AddServiceBuiltinName, moduleBar, 12, 18)
@@ -776,7 +781,7 @@ def run(plan):
 Adding service example-datastore-server
 `
 
-	_, instructionsPlan, interpretationError = interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, scriptB, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError = interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, scriptB, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 3, instructionsPlan.Size())
 	assertInstructionTypeAndPosition(t, instructionsPlan, 2, add_service.AddServiceBuiltinName, startosis_constants.PackageIdPlaceholderForStandaloneScript, 14, 18)
@@ -800,7 +805,7 @@ def run(plan):
 	plan.print(file_contents)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 2, instructionsPlan.Size())
 
@@ -838,7 +843,7 @@ def run(plan):
 	plan.print(artifact_name)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 3, instructionsPlan.Size())
 
@@ -884,7 +889,7 @@ def run(plan):
 	plan.print(uuid)
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 4, instructionsPlan.Size())
 
@@ -910,7 +915,7 @@ def run(plan):
 	plan.print("The service example-datastore-server has been removed")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 3, instructionsPlan.Size())
 
@@ -932,7 +937,7 @@ func TestStartosisInterpreter_NoPanicIfUploadIsPassedAPathNotOnDisk(t *testing.T
 def run(plan):
 	plan.upload_files("` + filePath + `")
 `
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.NotNil(t, interpretationError)
 	require.Nil(t, instructionsPlan)
 }
@@ -947,7 +952,7 @@ def run(plan):
 	plan.print("Hello World!")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size())
 
@@ -966,7 +971,7 @@ def run(plan):
 	plan.print("Hello World!")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, `{"number": 4}`)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, `{"number": 4}`, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size())
 
@@ -985,7 +990,7 @@ def run(plan, args):
 	plan.print("My favorite number is {0}".format(args["number"]))
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, `{"number": 4}`)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, `{"number": 4}`, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size())
 
@@ -1007,7 +1012,7 @@ def run(plan, args):
 		plan.print("Sorry no args!")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.Nil(t, interpretationError)
 	require.Equal(t, 1, instructionsPlan.Size())
 
@@ -1026,7 +1031,7 @@ def run(plan, args, invalid_arg):
 	plan.print("this wouldn't interpret so the text here doesnt matter")
 `
 
-	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, instructionsPlan, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.NotNil(t, interpretationError)
 	expectedError := fmt.Sprintf("The 'run' entrypoint function can have at most '%v' argument got '%v'", maximumParamsAllowedForRunFunction, 3)
 	require.Equal(t, expectedError, interpretationError.GetErrorMessage())
@@ -1043,7 +1048,7 @@ def run(plan):
 	print("this doesnt matter")
 `
 
-	_, _, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs)
+	_, _, interpretationError := interpreter.Interpret(context.Background(), startosis_constants.PackageIdPlaceholderForStandaloneScript, useDefaultMainFunctionName, script, startosis_constants.EmptyInputArgs, emptyInstructionsPlanMask)
 	require.NotNil(t, interpretationError)
 	require.Equal(t, fmt.Sprintf("Evaluation error: %v\n\tat [3:7]: run\n\tat [0:0]: print", print_builtin.UsePlanFromKurtosisInstructionError), interpretationError.GetErrorMessage())
 }
diff --git a/internal_testsuites/golang/testsuite/startosis_package_test/startosis_valid_package_no_input_test.go b/internal_testsuites/golang/testsuite/startosis_package_test/startosis_valid_package_no_input_test.go
index ae23c8ed0..d3bddad17 100644
--- a/internal_testsuites/golang/testsuite/startosis_package_test/startosis_valid_package_no_input_test.go
+++ b/internal_testsuites/golang/testsuite/startosis_package_test/startosis_valid_package_no_input_test.go
@@ -20,9 +20,9 @@ func (suite *StartosisPackageTestSuite) TestStartosisPackage_ValidPackageNoInput
 	require.Empty(t, runResult.ValidationErrors)
 	require.Nil(t, runResult.ExecutionError)
 
-	expectedScriptOutput := `Hello world!
+	expectedScriptOutput := `package with no input
 {
-	"message": "Hello world!"
+	"message": "package with no input"
 }
 `
 	require.Equal(t, expectedScriptOutput, string(runResult.RunOutput))
@@ -40,9 +40,9 @@ func (suite *StartosisPackageTestSuite) TestStartosisPackage_ValidPackageNoInput
 	require.Empty(t, runResult.ValidationErrors)
 	require.Nil(t, runResult.ExecutionError)
 
-	expectedScriptOutput := `Hello world!
+	expectedScriptOutput := `package with no input
 {
-	"message": "Hello world!"
+	"message": "package with no input"
 }
 `
 	require.Equal(t, expectedScriptOutput, string(runResult.RunOutput))
diff --git a/internal_testsuites/golang/testsuite/startosis_start_service_test/startosis_start_service_test.go b/internal_testsuites/golang/testsuite/startosis_start_service_test/startosis_start_service_test.go
index 2132e3dff..b00c4b0d4 100644
--- a/internal_testsuites/golang/testsuite/startosis_start_service_test/startosis_start_service_test.go
+++ b/internal_testsuites/golang/testsuite/startosis_start_service_test/startosis_start_service_test.go
@@ -14,7 +14,7 @@ const (
 	testName              = "startosis_start_service_test"
 	isPartitioningEnabled = false
 
-	serviceName = "example-datastore-server-1"
+	serviceName = "example-datastore-start-test"
 	portId      = "grpc"
 
 	starlarkScript = `
@@ -78,9 +78,9 @@ func TestStartosis(t *testing.T) {
 	require.Empty(t, runResult.ValidationErrors, "Unexpected validation error")
 	require.Nil(t, runResult.ExecutionError, "Unexpected execution error")
 
-	expectedScriptOutput := `Adding service example-datastore-server-1.
-Service 'example-datastore-server-1' added with service UUID '[a-z-0-9]+'
-Service example-datastore-server-1 deployed successfully.
+	expectedScriptOutput := `Adding service ` + serviceName + `.
+Service '` + serviceName + `' added with service UUID '[a-z-0-9]+'
+Service ` + serviceName + ` deployed successfully.
 `
 	require.Regexp(t, expectedScriptOutput, string(runResult.RunOutput))
 	logrus.Infof("Successfully ran Starlark script to add datastore service")
@@ -111,7 +111,7 @@ Service example-datastore-server-1 deployed successfully.
 	require.Empty(t, runResult.ValidationErrors, "Unexpected validation error")
 	require.Nil(t, runResult.ExecutionError, "Unexpected execution error")
 
-	expectedScriptOutput = `Service 'example-datastore-server-1' stopped
+	expectedScriptOutput = `Service '` + serviceName + `' stopped
 `
 	require.Regexp(t, expectedScriptOutput, string(runResult.RunOutput))
 
diff --git a/internal_testsuites/golang/testsuite/startosis_stop_service_test/startosis_stop_service_test.go b/internal_testsuites/golang/testsuite/startosis_stop_service_test/startosis_stop_service_test.go
index eb00d9c30..4be8e442c 100644
--- a/internal_testsuites/golang/testsuite/startosis_stop_service_test/startosis_stop_service_test.go
+++ b/internal_testsuites/golang/testsuite/startosis_stop_service_test/startosis_stop_service_test.go
@@ -14,7 +14,7 @@ const (
 	testName              = "startosis_stop_service_test"
 	isPartitioningEnabled = false
 
-	serviceName = "example-datastore-server-1"
+	serviceName = "example-datastore-stop-service"
 	portId      = "grpc"
 
 	starlarkScript = `
@@ -70,9 +70,9 @@ func TestStartosis(t *testing.T) {
 	require.Empty(t, runResult.ValidationErrors, "Unexpected validation error")
 	require.Nil(t, runResult.ExecutionError, "Unexpected execution error")
 
-	expectedScriptOutput := `Adding service example-datastore-server-1.
-Service 'example-datastore-server-1' added with service UUID '[a-z-0-9]+'
-Service example-datastore-server-1 deployed successfully.
+	expectedScriptOutput := `Adding service ` + serviceName + `.
+Service '` + serviceName + `' added with service UUID '[a-z-0-9]+'
+Service ` + serviceName + ` deployed successfully.
 `
 	require.Regexp(t, expectedScriptOutput, string(runResult.RunOutput))
 	logrus.Infof("Successfully ran Starlark script to add datastore service")
@@ -96,7 +96,7 @@ Service example-datastore-server-1 deployed successfully.
 	require.Empty(t, runResult.ValidationErrors, "Unexpected validation error")
 	require.Nil(t, runResult.ExecutionError, "Unexpected execution error")
 
-	expectedScriptOutput = `Service 'example-datastore-server-1' stopped
+	expectedScriptOutput = `Service '` + serviceName + `' stopped
 `
 	require.Regexp(t, expectedScriptOutput, string(runResult.RunOutput))
 
diff --git a/internal_testsuites/starlark/valid-kurtosis-package-no-input/main.star b/internal_testsuites/starlark/valid-kurtosis-package-no-input/main.star
index 8a2941c3c..ad9b99a1e 100644
--- a/internal_testsuites/starlark/valid-kurtosis-package-no-input/main.star
+++ b/internal_testsuites/starlark/valid-kurtosis-package-no-input/main.star
@@ -1,4 +1,4 @@
 def run(plan):
-    output = struct(message="Hello world!")
+    output = struct(message="package with no input")
     plan.print(output.message)
     return output