feat: Persist enclave plan in the Starlark executor memory (#757)

## Description:
The Starlark executor now keeps track of all instructions that were
executed within this enclave. It doesn't take into account the manual
modification that might have been made, such as manual file uploads via
the CLI.

## 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. -->

core/server/api_container/server/startosis_engine/instructions_plan/instructions_plan.go

@@ -0,0 +1,69 @@
+package instructions_plan
+
+import (
+	"github.com/kurtosis-tech/kurtosis/container-engine-lib/lib/uuid_generator"
+	"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/startosis_errors"
+	"github.com/kurtosis-tech/stacktrace"
+	"go.starlark.net/starlark"
+)
+
+// InstructionsPlan is the object to store a sequence of instructions which forms a "plan" for the enclave.
+// Right now, the object is fairly simple in the sense of it just stores literally the sequence of instructions, and
+// a bit of metadata about each instruction (i.e. whether it has been executed of not, for example)
+// The plan is "append-only", i.e. when an instruction is added, it cannot be removed.
+// The only read method is GeneratePlan unwraps the plan into an actual list of instructions that can be submitted to
+// the executor.
+type InstructionsPlan struct {
+	scheduledInstructionsIndex map[ScheduledInstructionUuid]*ScheduledInstruction
+
+	instructionsSequence []ScheduledInstructionUuid
+}
+
+func NewInstructionsPlan() *InstructionsPlan {
+	return &InstructionsPlan{
+		scheduledInstructionsIndex: map[ScheduledInstructionUuid]*ScheduledInstruction{},
+		instructionsSequence:       []ScheduledInstructionUuid{},
+	}
+}
+
+func (plan *InstructionsPlan) AddInstruction(instruction kurtosis_instruction.KurtosisInstruction, returnedValue starlark.Value) error {
+	generatedUuid, err := uuid_generator.GenerateUUIDString()
+	if err != nil {
+		return stacktrace.Propagate(err, "Unable to generate a random UUID for instruction '%s' to add it to the plan", instruction.String())
+	}
+
+	scheduledInstructionUuid := ScheduledInstructionUuid(generatedUuid)
+	scheduledInstruction := NewScheduledInstruction(scheduledInstructionUuid, instruction, returnedValue)
+
+	plan.scheduledInstructionsIndex[scheduledInstructionUuid] = scheduledInstruction
+	plan.instructionsSequence = append(plan.instructionsSequence, scheduledInstructionUuid)
+	return nil
+}
+
+func (plan *InstructionsPlan) AddScheduledInstruction(scheduledInstruction *ScheduledInstruction) *ScheduledInstruction {
+	newScheduledInstructionUuid := scheduledInstruction.uuid
+	newScheduledInstruction := NewScheduledInstruction(newScheduledInstructionUuid, scheduledInstruction.kurtosisInstruction, scheduledInstruction.returnedValue)
+	newScheduledInstruction.Executed(scheduledInstruction.IsExecuted())
+
+	plan.scheduledInstructionsIndex[newScheduledInstructionUuid] = newScheduledInstruction
+	plan.instructionsSequence = append(plan.instructionsSequence, newScheduledInstructionUuid)
+	return newScheduledInstruction
+}
+
+// GeneratePlan unwraps the plan into a list of instructions
+func (plan *InstructionsPlan) GeneratePlan() ([]*ScheduledInstruction, *startosis_errors.InterpretationError) {
+	var generatedPlan []*ScheduledInstruction
+	for _, instructionUuid := range plan.instructionsSequence {
+		instruction, found := plan.scheduledInstructionsIndex[instructionUuid]
+		if !found {
+			return nil, startosis_errors.NewInterpretationError("Unexpected error generating the Kurtosis Instructions plan. Instruction with UUID '%s' was scheduled but could not be found in Kurtosis instruction index", instructionUuid)
+		}
+		generatedPlan = append(generatedPlan, instruction)
+	}
+	return generatedPlan, nil
+}
+
+func (plan *InstructionsPlan) Size() int {
+	return len(plan.instructionsSequence)
+}

core/server/api_container/server/startosis_engine/instructions_plan/instructions_plan_test.go

@@ -0,0 +1,86 @@
+package instructions_plan
+
+import (
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/mock_instruction"
+	"github.com/stretchr/testify/require"
+	"go.starlark.net/starlark"
+	"testing"
+)
+
+func TestAddInstruction(t *testing.T) {
+	plan := NewInstructionsPlan()
+
+	instruction1 := mock_instruction.NewMockKurtosisInstruction(t)
+	instruction1ReturnedValue := starlark.None
+	require.NoError(t, plan.AddInstruction(instruction1, instruction1ReturnedValue))
+
+	require.Len(t, plan.instructionsSequence, 1)
+	scheduledInstructionUuid := plan.instructionsSequence[0]
+	require.Contains(t, plan.scheduledInstructionsIndex, scheduledInstructionUuid)
+	scheduledInstruction, found := plan.scheduledInstructionsIndex[scheduledInstructionUuid]
+	require.True(t, found)
+	require.Equal(t, scheduledInstruction.GetInstruction(), instruction1)
+	require.False(t, scheduledInstruction.IsExecuted())
+	require.Equal(t, scheduledInstruction.GetReturnedValue(), instruction1ReturnedValue)
+}
+
+func TestAddScheduledInstruction(t *testing.T) {
+	plan := NewInstructionsPlan()
+
+	instruction1Uuid := ScheduledInstructionUuid("instruction1")
+	instruction1 := mock_instruction.NewMockKurtosisInstruction(t)
+	instruction1ReturnedValue := starlark.MakeInt(1)
+	scheduleInstruction := NewScheduledInstruction(instruction1Uuid, instruction1, instruction1ReturnedValue)
+	scheduleInstruction.executed = true
+
+	plan.AddScheduledInstruction(scheduleInstruction)
+
+	require.Len(t, plan.instructionsSequence, 1)
+	scheduledInstructionUuid := plan.instructionsSequence[0]
+	require.Equal(t, instruction1Uuid, scheduledInstructionUuid)
+	require.Contains(t, plan.scheduledInstructionsIndex, scheduledInstructionUuid)
+	addedScheduledInstruction, found := plan.scheduledInstructionsIndex[scheduledInstructionUuid]
+	require.True(t, found)
+	require.NotSame(t, scheduleInstruction, addedScheduledInstruction) // validate the instruction was cloned
+	require.Equal(t, addedScheduledInstruction.GetInstruction(), instruction1)
+	require.True(t, addedScheduledInstruction.IsExecuted())
+	require.Equal(t, addedScheduledInstruction.GetReturnedValue(), instruction1ReturnedValue)
+}
+
+func TestGeneratePlan(t *testing.T) {
+	plan := NewInstructionsPlan()
+
+	// add instruction1 which is marked as executed
+	instruction1Uuid := ScheduledInstructionUuid("instruction1")
+	instruction1 := mock_instruction.NewMockKurtosisInstruction(t)
+	instruction1ReturnedValue := starlark.None
+	scheduleInstruction1 := NewScheduledInstruction(instruction1Uuid, instruction1, instruction1ReturnedValue)
+	scheduleInstruction1.Executed(true)
+
+	plan.scheduledInstructionsIndex[instruction1Uuid] = scheduleInstruction1
+	plan.instructionsSequence = append(plan.instructionsSequence, instruction1Uuid)
+
+	// add instruction2 which by default is not executed
+	instruction2Uuid := ScheduledInstructionUuid("instruction2")
+	instruction2 := mock_instruction.NewMockKurtosisInstruction(t)
+	instruction2ReturnedValue := starlark.MakeInt(1)
+	scheduleInstruction2 := NewScheduledInstruction(instruction2Uuid, instruction2, instruction2ReturnedValue)
+
+	plan.scheduledInstructionsIndex[instruction2Uuid] = scheduleInstruction2
+	plan.instructionsSequence = append(plan.instructionsSequence, instruction2Uuid)
+
+	// generate plan and validate it
+	instructionsSequence, err := plan.GeneratePlan()
+	require.Nil(t, err)
+
+	require.Len(t, instructionsSequence, 2)
+	scheduledInstruction1 := instructionsSequence[0]
+	require.Equal(t, scheduledInstruction1.GetInstruction(), instruction1)
+	require.True(t, scheduledInstruction1.IsExecuted())
+	require.Equal(t, scheduledInstruction1.GetReturnedValue(), instruction1ReturnedValue)
+
+	scheduledInstruction2 := instructionsSequence[1]
+	require.Equal(t, scheduledInstruction2.GetInstruction(), instruction2)
+	require.False(t, scheduledInstruction2.IsExecuted())
+	require.Equal(t, scheduledInstruction2.GetReturnedValue(), instruction2ReturnedValue)
+}

core/server/api_container/server/startosis_engine/instructions_plan/scheduled_instruction.go

@@ -0,0 +1,49 @@
+package instructions_plan
+
+import (
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction"
+	"go.starlark.net/starlark"
+)
+
+type ScheduledInstructionUuid string
+
+// ScheduledInstruction is a wrapper around a KurtosisInstruction to specify that the instruction is part of an
+// InstructionPlan. The instruction plan can either be the current enclave plan (which has been executed) or a newly
+// generated plan from the latest interpretation.
+// In any case, the ScheduledInstructionUuid stores the result object from the interpretation of the instruction,
+// as well as a flag to track whether this instruction was already executed or not.
+type ScheduledInstruction struct {
+	uuid ScheduledInstructionUuid
+
+	kurtosisInstruction kurtosis_instruction.KurtosisInstruction
+
+	returnedValue starlark.Value
+
+	executed bool
+}
+
+func NewScheduledInstruction(uuid ScheduledInstructionUuid, kurtosisInstruction kurtosis_instruction.KurtosisInstruction, returnedValue starlark.Value) *ScheduledInstruction {
+	return &ScheduledInstruction{
+		uuid:                uuid,
+		kurtosisInstruction: kurtosisInstruction,
+		returnedValue:       returnedValue,
+		executed:            false,
+	}
+}
+
+func (instruction *ScheduledInstruction) GetInstruction() kurtosis_instruction.KurtosisInstruction {
+	return instruction.kurtosisInstruction
+}
+
+func (instruction *ScheduledInstruction) GetReturnedValue() starlark.Value {
+	return instruction.returnedValue
+}
+
+func (instruction *ScheduledInstruction) Executed(isExecuted bool) *ScheduledInstruction {
+	instruction.executed = isExecuted
+	return instruction
+}
+
+func (instruction *ScheduledInstruction) IsExecuted() bool {
+	return instruction.executed
+}

core/server/api_container/server/startosis_engine/instructions_plan/scheduled_instruction_test.go

@@ -0,0 +1,19 @@
+package instructions_plan
+
+import (
+	"github.com/kurtosis-tech/kurtosis/core/server/api_container/server/startosis_engine/kurtosis_instruction/mock_instruction"
+	"github.com/stretchr/testify/require"
+	"go.starlark.net/starlark"
+	"testing"
+)
+
+func TestExecutedDefaultAndUpdate(t *testing.T) {
+	instruction1Uuid := ScheduledInstructionUuid("instruction1")
+	instruction1 := mock_instruction.NewMockKurtosisInstruction(t)
+	instruction1ReturnedValue := starlark.MakeInt(1)
+	scheduleInstruction := NewScheduledInstruction(instruction1Uuid, instruction1, instruction1ReturnedValue)
+
+	require.False(t, scheduleInstruction.executed)
+	scheduleInstruction.Executed(true)
+	require.True(t, scheduleInstruction.executed)
+}

core/server/api_container/server/startosis_engine/kurtosis_instruction/plan_module/plan_module.go

@@ -1,7 +1,7 @@
 package plan_module
 
 import (
-	"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/instructions_plan"
 	"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"
@@ -12,12 +12,12 @@ const (
 )
 
 func PlanModule(
-	instructionsQueue *[]kurtosis_instruction.KurtosisInstruction,
+	instructionsPlan *instructions_plan.InstructionsPlan,
 	kurtosisPlanInstructions []*kurtosis_plan_instruction.KurtosisPlanInstruction,
 ) *starlarkstruct.Module {
 	moduleBuiltins := starlark.StringDict{}
 	for _, planInstruction := range kurtosisPlanInstructions {
-		wrappedPlanInstruction := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(planInstruction, instructionsQueue)
+		wrappedPlanInstruction := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(planInstruction, instructionsPlan)
 		moduleBuiltins[planInstruction.GetName()] = starlark.NewBuiltin(planInstruction.GetName(), wrappedPlanInstruction.CreateBuiltin())
 	}
 

core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/kurtosis_plan_instruction/kurtosis_plan_instruction.go

@@ -1,8 +1,9 @@
 package kurtosis_plan_instruction
 
 import (
-	"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/instructions_plan"
 	"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,13 @@ type KurtosisPlanInstructionWrapper struct {
 	*KurtosisPlanInstruction
 
 	// TODO: This can be changed to KurtosisPlanInstructionInternal when we get rid of KurtosisInstruction
-	instructionQueue *[]kurtosis_instruction.KurtosisInstruction
+	instructionsPlan *instructions_plan.InstructionsPlan
 }
 
-func NewKurtosisPlanInstructionWrapper(instruction *KurtosisPlanInstruction, instructionQueue *[]kurtosis_instruction.KurtosisInstruction) *KurtosisPlanInstructionWrapper {
+func NewKurtosisPlanInstructionWrapper(instruction *KurtosisPlanInstruction, instructionsPlan *instructions_plan.InstructionsPlan) *KurtosisPlanInstructionWrapper {
 	return &KurtosisPlanInstructionWrapper{
 		KurtosisPlanInstruction: instruction,
-		instructionQueue:        instructionQueue,
+		instructionsPlan:        instructionsPlan,
 	}
 }
 
@@ -44,8 +45,12 @@ func (builtin *KurtosisPlanInstructionWrapper) CreateBuiltin() func(thread *star
 		}
 
 		// before returning, automatically add instruction to queue
-		*builtin.instructionQueue = append(*builtin.instructionQueue, instructionWrapper)
-
+		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
 	}
 }

core/server/api_container/server/startosis_engine/kurtosis_starlark_framework/test_engine/starlark_framework_engine_test.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"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/kurtosis_instruction"
+	"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_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"
@@ -73,22 +73,24 @@ func TestAllRegisteredBuiltins(t *testing.T) {
 
 func testKurtosisPlanInstruction(t *testing.T, builtin KurtosisPlanInstructionBaseTest) {
 	testId := builtin.GetId()
-	var instructionQueue []kurtosis_instruction.KurtosisInstruction
+	instructionsPlan := instructions_plan.NewInstructionsPlan()
 	thread := newStarlarkThread("framework-testing-engine")
 
 	predeclared := getBasePredeclaredDict(t)
 	// Add the KurtosisPlanInstruction that is being tested
 	instructionFromBuiltin := builtin.GetInstruction()
-	instructionWrapper := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(instructionFromBuiltin, &instructionQueue)
+	instructionWrapper := kurtosis_plan_instruction.NewKurtosisPlanInstructionWrapper(instructionFromBuiltin, instructionsPlan)
 	predeclared[instructionWrapper.GetName()] = starlark.NewBuiltin(instructionWrapper.GetName(), instructionWrapper.CreateBuiltin())
 
 	starlarkCode := builtin.GetStarlarkCode()
 	globals, err := starlark.ExecFile(thread, startosis_constants.PackageIdPlaceholderForStandaloneScript, codeToExecute(starlarkCode), predeclared)
 	require.Nil(t, err, "Error interpreting Starlark code for instruction '%s'", testId)
 	interpretationResult := extractResultValue(t, globals)
 
-	require.Len(t, instructionQueue, 1)
-	instructionToExecute := instructionQueue[0]
+	require.Equal(t, 1, instructionsPlan.Size())
+	instructionsSequence, err := instructionsPlan.GeneratePlan()
+	require.Nil(t, err)
+	instructionToExecute := instructionsSequence[0].GetInstruction()
 
 	// execute the instruction and run custom builtin assertions
 	executionResult, err := instructionToExecute.Execute(context.WithValue(context.Background(), "PARALLELISM", 1))