Add basic per-actor external volume flow

cmd/ateapi/internal/controlapi/create_actor.go

@@ -33,12 +33,11 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
 	if err := validateCreateActorRequest(req); err != nil {
 		return nil, err
 	}
-
 	in := req.GetActor()
 	templateNamespace := in.GetActorTemplateNamespace()
 	templateName := in.GetActorTemplateName()
 
-	_, err := s.actorTemplateLister.ActorTemplates(templateNamespace).Get(templateName)
+	template, err := s.actorTemplateLister.ActorTemplates(templateNamespace).Get(templateName)
 	if err != nil {
 		if k8serrors.IsNotFound(err) {
 			return nil, status.Errorf(codes.FailedPrecondition, "ActorTemplate %s/%s not found", templateNamespace, templateName)
@@ -58,6 +57,16 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
 		return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", atespace)
 	}
 
+	actorRef := &ateapipb.ObjectRef{
+		Atespace: atespace,
+		Name:     name,
+	}
+
+	volumes, err := s.createActorVolumes(ctx, actorRef, template)
+	if err != nil {
+		return nil, err
+	}
+
 	actor := &ateapipb.Actor{
 		Metadata: &ateapipb.ResourceMetadata{
 			Atespace: atespace,
@@ -67,9 +76,12 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
 		ActorTemplateNamespace: templateNamespace,
 		ActorTemplateName:      templateName,
 		WorkerSelector:         in.GetWorkerSelector(),
+		ActorVolumes:           volumes,
 	}
 	stored, err := s.persistence.CreateActor(ctx, actor)
 	if err != nil {
+		// Cleanup created volumes if DB write fails
+		s.deleteActorVolumes(ctx, actorRef, volumes)
 		if errors.Is(err, store.ErrAlreadyExists) {
 			return nil, status.Errorf(codes.AlreadyExists, "Actor %s already exists", name)
 		}

cmd/ateapi/internal/controlapi/delete_actor.go

@@ -32,7 +32,21 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ
 		return nil, err
 	}
 
-	deleted, err := s.persistence.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())
+	atespace := req.GetActor().GetAtespace()
+	name := req.GetActor().GetName()
+
+	actor, err := s.persistence.GetActor(ctx, atespace, name)
+	if err != nil {
+		if errors.Is(err, store.ErrNotFound) {
+			return nil, status.Errorf(codes.NotFound, "Actor %s not found", name)
+		}
+		return nil, fmt.Errorf("while fetching actor: %w", err)
+	}
+
+	// Delete associated volumes (TODO: best effort?)
+	s.deleteActorVolumes(ctx, req.GetActor(), actor.GetActorVolumes())
+
+	deleted, err := s.persistence.DeleteActor(ctx, atespace, name)
 	if err != nil {
 		if errors.Is(err, store.ErrNotFound) {
 			return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActor().GetName())

cmd/ateapi/internal/controlapi/volumes.go

@@ -0,0 +1,111 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlapi
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/agent-substrate/substrate/internal/volume"
+	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
+	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var (
+	globalVolumePlugin = volume.NewMockVolumePlugin()
+)
+
+// TODO: Replace with actual volume plugin search
+func getVolumePlugin() volume.VolumePlugin {
+	return globalVolumePlugin
+}
+
+// TODO: we should persist creation first so that we can handle background cleanup.
+// this probably requires us to add a PROVISIONING actor state.
+
+// createActorVolumes provisions external volumes specified in the actor template.
+// It returns the list of created external volumes, or an error if any creation fails.
+// If any volume creation fails, it cleans up any volumes created in this call on a best-effort basis.
+func (s *Service) createActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, template *atev1alpha1.ActorTemplate) ([]*ateapipb.ExternalVolume, error) {
+	var volumes []*ateapipb.ExternalVolume
+	for _, vol := range template.Spec.Volumes {
+		if vol.ExternalVolumeTemplate != nil {
+			// Use a unique name for the volume to ensure idempotency
+			uniqueVolName := actorVolumeID(ref, vol.Name)
+			storageVolumeID, err := getVolumePlugin().CreateVolume(ctx, uniqueVolName, vol.ExternalVolumeTemplate.Capacity.String(), vol.ExternalVolumeTemplate.StorageClassName)
+			if err != nil {
+				// TODO: need better system - best effort cleanup of already created volumes
+				s.deleteActorVolumes(ctx, ref, volumes)
+				return nil, status.Errorf(codes.Internal, "failed to create volume %q: %v", vol.Name, err)
+			}
+			volumes = append(volumes, &ateapipb.ExternalVolume{
+				ActorVolumeId:   uniqueVolName,
+				StorageVolumeId: storageVolumeID,
+				VolumeType:      "mock", // TODO fix when we support multiple plugins
+				Status:          ateapipb.ExternalVolume_CREATED,
+			})
+		}
+	}
+	return volumes, nil
+}
+
+// deleteActorVolumes deletes all external volumes in the list on a best-effort basis.
+func (s *Service) deleteActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volumes []*ateapipb.ExternalVolume) {
+	for _, vol := range volumes {
+		if err := getVolumePlugin().DeleteVolume(ctx, vol.GetStorageVolumeId()); err != nil {
+			slog.ErrorContext(ctx, "failed to delete volume",
+				slog.String("atespace", ref.GetAtespace()),
+				slog.String("actor_id", ref.GetName()),
+				slog.String("volume_id", vol.GetStorageVolumeId()),
+				slog.Any("error", err))
+		}
+	}
+}
+
+// getMountedActorVolumes filters the actor's volumes and returns only those that are declared and mounted in the ActorTemplate.
+func getMountedActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volumes []*ateapipb.ExternalVolume, template *atev1alpha1.ActorTemplate) []*ateapipb.ExternalVolume {
+	var mounted []*ateapipb.ExternalVolume
+	for _, vol := range volumes {
+		// Find the corresponding volume in the ActorTemplate to check if it's mounted
+		var matchedTemplateVol *atev1alpha1.Volume
+		for _, tVol := range template.Spec.Volumes {
+			expectedID := actorVolumeID(ref, tVol.Name)
+			if vol.GetActorVolumeId() == expectedID {
+				matchedTemplateVol = &tVol
+				break
+			}
+		}
+
+		if matchedTemplateVol == nil {
+			slog.WarnContext(ctx, "Volume not found in template, skipping", slog.String("volume_id", vol.GetStorageVolumeId()))
+			continue
+		}
+
+		if !isVolumeMounted(matchedTemplateVol.Name, template) {
+			slog.InfoContext(ctx, "Volume not mounted in template, skipping", slog.String("volume_id", vol.GetStorageVolumeId()))
+			continue
+		}
+		mounted = append(mounted, vol)
+	}
+	return mounted
+}
+
+func actorVolumeID(ref *ateapipb.ObjectRef, volumeName string) string {
+	// TODO consider if this should be actor UUID
+	return fmt.Sprintf("%s-%s-%s", ref.GetAtespace(), ref.GetName(), volumeName)
+}

cmd/ateapi/internal/controlapi/workflow.go

@@ -167,6 +167,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, atespace, name string,
 	steps := []WorkflowStep[*ResumeInput, *ResumeState]{
 		&LoadActorForResumeStep{store: w.store, actorTemplateLister: w.actorTemplateLister},
 		&AssignWorkerStep{store: w.store, workerCache: w.workerCache},
+		&AttachVolumesStep{store: w.store},
 		&CallAteletRestoreStep{store: w.store, dialer: w.dialer, kubeClient: w.kubeClient, secretCache: w.secretCache, workerPoolLister: w.workerPoolLister, sandboxConfigLister: w.sandboxConfigLister},
 		&FinalizeRunningStep{store: w.store},
 	}
@@ -198,6 +199,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, atespace, name string)
 		&LoadActorForSuspendStep{store: w.store, actorTemplateLister: w.actorTemplateLister},
 		&MarkSuspendingStep{store: w.store},
 		&CallAteletSuspendStep{store: w.store, dialer: w.dialer},
+		&DetachVolumesStep{store: w.store},
 		&FinalizeSuspendedStep{store: w.store},
 	}
 
@@ -228,6 +230,7 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, atespace, name string) (
 		&LoadActorForPauseStep{store: w.store, actorTemplateLister: w.actorTemplateLister},
 		&MarkPausingStep{store: w.store},
 		&CallAteletPauseStep{store: w.store, dialer: w.dialer},
+		&DetachVolumesForPauseStep{store: w.store},
 		&FinalizePausedStep{store: w.store},
 	}
 

cmd/ateapi/internal/controlapi/workflow_pause.go

@@ -124,7 +124,10 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
 	}
 	client := ateletpb.NewAteomHerderClient(ateletConn)
 
-	workloadSpec := workloadSpecFromActorTemplate(state.ActorTemplate)
+	workloadSpec, err := workloadSpecFromActorTemplate(state.ActorTemplate, state.Actor)
+	if err != nil {
+		return err
+	}
 
 	// Checkpoint does not carry the sandbox config: atelet uses the version the
 	// actor is currently running (recorded on-node at Run/Restore) and pins it
@@ -151,6 +154,51 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
 
 func (s *CallAteletPauseStep) RetryBackoff() *wait.Backoff { return nil }
 
+type DetachVolumesForPauseStep struct {
+	store store.Interface
+}
+
+func (s *DetachVolumesForPauseStep) Name() string { return "DetachVolumesForPause" }
+
+func (s *DetachVolumesForPauseStep) IsComplete(ctx context.Context, input *PauseInput, state *PauseState) (bool, error) {
+	// TODO replace with a proper check on the volumes.
+	return state.Actor.GetStatus() == ateapipb.Actor_STATUS_PAUSED && state.Actor.GetAteomPodNamespace() == "", nil
+}
+
+func (s *DetachVolumesForPauseStep) Execute(ctx context.Context, input *PauseInput, state *PauseState) error {
+	if state.Actor.GetAteomPodNamespace() == "" {
+		slog.WarnContext(ctx, "Actor has no assigned worker pod during pause, skipping detach volumes", slog.String("actor_id", input.ActorName))
+		return nil
+	}
+
+	worker, err := s.store.GetWorker(ctx, state.Actor.GetAteomPodNamespace(), state.Actor.GetWorkerPoolName(), state.Actor.GetAteomPodName())
+	if err != nil {
+		if errors.Is(err, store.ErrNotFound) {
+			slog.WarnContext(ctx, "Worker not found in store during pause, skipping detach volumes", slog.String("actor_id", input.ActorName))
+			return nil
+		}
+		return fmt.Errorf("failed to get worker: %w", err)
+	}
+
+	node := worker.GetNodeName()
+	if node == "" {
+		slog.WarnContext(ctx, "Worker has no assigned node name during pause, skipping detach volumes", slog.String("actor_id", input.ActorName))
+		return nil
+	}
+
+	ref := &ateapipb.ObjectRef{Atespace: state.Actor.GetMetadata().GetAtespace(), Name: state.Actor.GetMetadata().GetName()}
+	for _, vol := range getMountedActorVolumes(ctx, ref, state.Actor.GetActorVolumes(), state.ActorTemplate) {
+		slog.InfoContext(ctx, "Detaching volume from node", slog.String("volume_id", vol.GetStorageVolumeId()), slog.String("node", node))
+		err := getVolumePlugin().DetachVolume(ctx, vol.GetStorageVolumeId(), node)
+		if err != nil {
+			return fmt.Errorf("failed to detach volume %q from node %q: %w", vol.GetStorageVolumeId(), node, err)
+		}
+	}
+	return nil
+}
+
+func (s *DetachVolumesForPauseStep) RetryBackoff() *wait.Backoff { return nil }
+
 type FinalizePausedStep struct {
 	store store.Interface
 }

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -208,6 +208,45 @@ func (s *AssignWorkerStep) RetryBackoff() *wait.Backoff {
 	}
 }
 
+type AttachVolumesStep struct {
+	store store.Interface
+}
+
+func (s *AttachVolumesStep) Name() string { return "AttachVolumes" }
+
+func (s *AttachVolumesStep) IsComplete(ctx context.Context, input *ResumeInput, state *ResumeState) (bool, error) {
+	// TODO replace with a proper check on the volumes.
+	return state.Actor.GetStatus() == ateapipb.Actor_STATUS_RUNNING, nil
+}
+
+func (s *AttachVolumesStep) Execute(ctx context.Context, input *ResumeInput, state *ResumeState) error {
+	if state.Actor.GetAteomPodNamespace() == "" {
+		return fmt.Errorf("actor has no assigned worker pod")
+	}
+
+	worker, err := s.store.GetWorker(ctx, state.Actor.GetAteomPodNamespace(), state.Actor.GetWorkerPoolName(), state.Actor.GetAteomPodName())
+	if err != nil {
+		return fmt.Errorf("failed to get worker for volume attachment: %w", err)
+	}
+
+	node := worker.GetNodeName()
+	if node == "" {
+		return fmt.Errorf("assigned worker has no node name")
+	}
+
+	ref := &ateapipb.ObjectRef{Atespace: state.Actor.GetMetadata().GetAtespace(), Name: state.Actor.GetMetadata().GetName()}
+	for _, vol := range getMountedActorVolumes(ctx, ref, state.Actor.GetActorVolumes(), state.ActorTemplate) {
+		slog.InfoContext(ctx, "Attaching volume to node", slog.String("volume_id", vol.GetStorageVolumeId()), slog.String("node", node))
+		err := getVolumePlugin().AttachVolume(ctx, vol.GetStorageVolumeId(), node)
+		if err != nil {
+			return fmt.Errorf("failed to attach volume %q to node %q: %w", vol.GetStorageVolumeId(), node, err)
+		}
+	}
+	return nil
+}
+
+func (s *AttachVolumesStep) RetryBackoff() *wait.Backoff { return nil }
+
 func (s *AssignWorkerStep) findFreeWorker(
 	workers []*ateapipb.Worker,
 	templateClass atev1alpha1.SandboxClass,
@@ -261,7 +300,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 	}
 	client := ateletpb.NewAteomHerderClient(ateletConn)
 
-	workloadSpec, err := workloadSpecFromActorTemplateWithEnv(ctx, s.kubeClient, s.secretCache, state.ActorTemplate)
+	workloadSpec, err := workloadSpecFromActorTemplateWithEnv(ctx, s.kubeClient, s.secretCache, state.ActorTemplate, state.Actor)
 	if err != nil {
 		return err
 	}

cmd/ateapi/internal/controlapi/workflow_suspend.go

@@ -126,7 +126,10 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
 	}
 	client := ateletpb.NewAteomHerderClient(ateletConn)
 
-	workloadSpec := workloadSpecFromActorTemplate(state.ActorTemplate)
+	workloadSpec, err := workloadSpecFromActorTemplate(state.ActorTemplate, state.Actor)
+	if err != nil {
+		return err
+	}
 
 	// Checkpoint does not carry the sandbox config: atelet uses the version the
 	// actor is currently running (recorded on-node at Run/Restore) and pins it
@@ -153,6 +156,51 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
 
 func (s *CallAteletSuspendStep) RetryBackoff() *wait.Backoff { return nil }
 
+type DetachVolumesStep struct {
+	store store.Interface
+}
+
+func (s *DetachVolumesStep) Name() string { return "DetachVolumes" }
+
+func (s *DetachVolumesStep) IsComplete(ctx context.Context, input *SuspendInput, state *SuspendState) (bool, error) {
+	// TODO replace with a proper check on the volumes.
+	return state.Actor.GetStatus() == ateapipb.Actor_STATUS_SUSPENDED && state.Actor.GetAteomPodNamespace() == "", nil
+}
+
+func (s *DetachVolumesStep) Execute(ctx context.Context, input *SuspendInput, state *SuspendState) error {
+	if state.Actor.GetAteomPodNamespace() == "" {
+		slog.WarnContext(ctx, "Actor has no assigned worker pod during suspend, skipping detach volumes", slog.String("actor_id", input.ActorName))
+		return nil
+	}
+
+	worker, err := s.store.GetWorker(ctx, state.Actor.GetAteomPodNamespace(), state.Actor.GetWorkerPoolName(), state.Actor.GetAteomPodName())
+	if err != nil {
+		if errors.Is(err, store.ErrNotFound) {
+			slog.WarnContext(ctx, "Worker not found in store during suspend, skipping detach volumes", slog.String("actor_id", input.ActorName))
+			return nil
+		}
+		return fmt.Errorf("failed to get worker: %w", err)
+	}
+
+	node := worker.GetNodeName()
+	if node == "" {
+		slog.WarnContext(ctx, "Worker has no assigned node name during suspend, skipping detach volumes", slog.String("actor_id", input.ActorName))
+		return nil
+	}
+
+	ref := &ateapipb.ObjectRef{Atespace: state.Actor.GetMetadata().GetAtespace(), Name: state.Actor.GetMetadata().GetName()}
+	for _, vol := range getMountedActorVolumes(ctx, ref, state.Actor.GetActorVolumes(), state.ActorTemplate) {
+		slog.InfoContext(ctx, "Detaching volume from node", slog.String("volume_id", vol.GetStorageVolumeId()), slog.String("node", node))
+		err := getVolumePlugin().DetachVolume(ctx, vol.GetStorageVolumeId(), node)
+		if err != nil {
+			return fmt.Errorf("failed to detach volume %q from node %q: %w", vol.GetStorageVolumeId(), node, err)
+		}
+	}
+	return nil
+}
+
+func (s *DetachVolumesStep) RetryBackoff() *wait.Backoff { return nil }
+
 type FinalizeSuspendedStep struct {
 	store store.Interface
 }