diff --git a/cmd/ateapi/internal/controlapi/workflow_pause.go b/cmd/ateapi/internal/controlapi/workflow_pause.go
deleted file mode 100644
index 9949f76475..0000000000
--- a/cmd/ateapi/internal/controlapi/workflow_pause.go
+++ /dev/null
@@ -1,300 +0,0 @@
-// 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"
-	"errors"
-	"fmt"
-	"log/slog"
-	"time"
-
-	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
-	"github.com/agent-substrate/substrate/internal/apierror"
-	"github.com/agent-substrate/substrate/internal/ateattr"
-	"github.com/agent-substrate/substrate/internal/preview"
-	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
-	"github.com/agent-substrate/substrate/internal/resources"
-	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
-	"go.opentelemetry.io/otel/attribute"
-	"google.golang.org/protobuf/proto"
-)
-
-// PauseActor executes the workflow to pause a running actor. Idempotent:
-// a re-entered workflow fast-forwards past the steps a previous attempt
-// completed, deriving progress from the persisted actor alone.
-func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) {
-	ctx = preview.ContextWithOptIn(ctx)
-	start := time.Now()
-	var actor *ateapipb.Actor
-	var actorTemplate *ateapipb.ActorTemplate
-	var wireFidelity string
-	// Set just before finalize; nil until then, so earlier exits label
-	// themselves from the record they hold.
-	var finalAttrs []attribute.KeyValue
-
-	defer func() {
-		attrs := finalAttrs
-		if attrs == nil {
-			attrs = lifecycleOpAttrs(actor, actorTemplate, "", wireFidelity)
-		}
-		w.instruments.recordLifecycleOp(ctx, ateattr.OperationSuspend, start, err, attrs...)
-	}()
-
-	leaseCtx, lease, err := w.acquireActorLease(ctx, actorRef)
-	if err != nil {
-		return nil, err
-	}
-	defer lease.Close()
-
-	actor, actorTemplate, err = w.loadActorForPause(leaseCtx, actorRef)
-	if err != nil {
-		return nil, err
-	}
-	if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED && latestSnapshotIsLocalOnly(actor.GetStatus()) {
-		// Fully paused already: FinalizePaused commits SUSPENDED, the local
-		// snapshot's locality, and the cleared worker assignment in a single
-		// update, so there is nothing left to do.
-		// This success reports no pool, and cannot: the previous attempt
-		// released the worker, so the record names none (#957).
-		return actor, nil
-	}
-	var marked *ateapipb.Actor
-	if marked, err = w.ensureMarkedPausing(leaseCtx, actorRef, actor, actorTemplate); err != nil {
-		return nil, err
-	}
-	actor = marked
-	if wireFidelity, err = w.ensureAteletPaused(leaseCtx, actorRef, actor, actorTemplate); err != nil {
-		return nil, err
-	}
-	// TODO: There is no difference between suspend and pause for now, but we
-	// could optimize pause by not detaching. We would need to make sure Resume
-	// is idempotent.
-	if err = w.ensureVolumesDetached(leaseCtx, actor, actorTemplate, "DetachVolumesForPause", ateattr.OperationSuspend); err != nil {
-		return nil, err
-	}
-	// FinalizePaused clears the WorkerAssignment the labels read, so snapshot
-	// them here, as crash.go does for the crash counter.
-	finalAttrs = lifecycleOpAttrs(actor, actorTemplate, "", wireFidelity)
-	var finalized *ateapipb.Actor
-	if finalized, err = w.ensurePausedFinalized(leaseCtx, actorRef); err != nil {
-		return nil, err
-	}
-	actor = finalized
-	return actor, nil
-}
-
-// loadActorForPause fetches the current actor record and its template.
-func (w *ActorWorkflow) loadActorForPause(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, _ *ateapipb.ActorTemplate, err error) {
-	ctx, done := stepSpan(ctx, "LoadActorForPause")
-	defer func() { err = done(err) }()
-
-	actor, err := w.store.GetActor(ctx, actorRef)
-	if err != nil {
-		return nil, nil, err
-	}
-	actorTemplate, err := resolveActorTemplate(ctx, w.store, actor)
-	if err != nil {
-		return nil, nil, err
-	}
-	return actor, actorTemplate, nil
-}
-
-// ensureMarkedPausing transitions a RUNNING actor to SUSPENDING, minting the
-// local snapshot name. A pause in progress is a SUSPENDING actor whose latest
-// generation is a pending node-local snapshot; a suspend in progress has a
-// pending durable one instead. Skips when a previous attempt already marked
-// the actor; the persisted name then stays authoritative for the rest of the
-// workflow.
-func (w *ActorWorkflow) ensureMarkedPausing(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (_ *ateapipb.Actor, err error) {
-	ctx, done := stepSpan(ctx, "MarkPausing")
-	defer func() { err = done(err) }()
-
-	if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDING && latestSnapshotIsPendingLocal(actor.GetStatus()) {
-		markSkipped(ctx, "actor already SUSPENDING for a pause")
-		return actor, nil
-	}
-	// The pause edge only exists from RUNNING. A SUSPENDING actor that is not
-	// pausing is suspending, and that suspend owns it.
-	if actor.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_RUNNING {
-		return nil, apierror.FailedPrecondition("MarkPausing prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, actor.GetStatus().GetState(), ateapipb.ActorState_ACTOR_STATE_RUNNING)
-	}
-	// By design a golden actor cannot be paused — it can only be suspended
-	// (committed).
-	if actorRef.Atespace == resources.GoldenActorAtespace {
-		return nil, apierror.FailedPrecondition("actors in atespace %q are golden actors, which cannot be paused", actorRef.Atespace)
-	}
-
-	snapshotUUID := resources.NewSnapshotName()
-	storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error {
-		toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDING
-		// Increment last_assigned_generation for the new Pause request.
-		gen := toUpdate.Status.LastAssignedGeneration + 1
-		toUpdate.Status.LastAssignedGeneration = gen
-		snap := newLocalSnapshot(
-			gen,
-			actorTemplate.GetSnapshotConfig().GetPreferredFidelity(),
-			actorTemplate.GetMetadata().GetUid(),
-			snapshotUUID,
-			"",
-		)
-		snap.SandboxRuntime = proto.CloneOf(toUpdate.GetStatus().GetWorkerAssignment().GetSandboxRuntime())
-		toUpdate.Status.Snapshots = append(toUpdate.Status.Snapshots, snap)
-		return nil
-	})
-	if err != nil {
-		if errors.Is(err, store.ErrVersionConflict) {
-			return nil, apierror.Aborted("concurrent update conflict, please retry")
-		}
-		return nil, err
-	}
-	logActorStateChanged(ctx, storedActor, ateattr.OperationSuspend)
-	return storedActor, nil
-}
-
-// ensureAteletPaused checkpoints the workload locally on the worker node
-// under the actor's persisted snapshot name. This is the atelet reentrancy
-// seam (#372): the request is keyed by the actor UID, the worker pod UID, and
-// the once-minted snapshot name, so a re-entered workflow re-sends the same
-// semantic request; once atelet's Checkpoint is idempotent on those keys this
-// step becomes fully reentrant with no changes here.
-func (w *ActorWorkflow) ensureAteletPaused(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (wireFidelity string, err error) {
-	ctx, done := stepSpan(ctx, "CallAteletPause")
-	defer func() { err = done(err) }()
-
-	assignment := actor.GetStatus().GetWorkerAssignment()
-	if assignment == nil {
-		// Missing active worker pod reference mid-pause indicates corrupted store state.
-		return "", crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageWorkerAssignmentMissing)
-	}
-
-	ateletConn, err := w.dialer.DialForAteletOnNode(assignment.GetNodeName())
-	if err != nil {
-		return "", fmt.Errorf("while getting atelet conn for node %q: %w", assignment.GetNodeName(), err)
-	}
-	client := ateletpb.NewAteletClient(ateletConn)
-
-	workloadSpec, err := workloadSpecFromActorTemplate(actorTemplate, actor, nil)
-	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
-	// into the snapshot manifest.
-	inProgressSnap := snapshotAtLatestGeneration(actor.GetStatus())
-	req := &ateletpb.CheckpointRequest{
-		WorkerPodUid:          assignment.GetWorkerPodUid(),
-		Atespace:              actor.GetMetadata().GetAtespace(),
-		ActorName:             actor.GetMetadata().GetName(),
-		ActorTemplateAtespace: actor.GetActorTemplate().GetAtespace(),
-		ActorTemplateName:     actor.GetActorTemplate().GetName(),
-		Spec:                  workloadSpec,
-		Snapshot: &ateletpb.Snapshot{
-			SnapshotUuid: inProgressSnap.GetUuid(),
-			Generation:   inProgressSnap.GetGeneration(),
-			Fidelity:     fidelityToAtelet(actorTemplate.GetSnapshotConfig().GetPreferredFidelity()),
-		},
-		UploadConfig: &ateletpb.UploadConfig{
-			Mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED,
-		},
-		ActorUid: actor.GetMetadata().Uid,
-	}
-	wireFidelity = ateattr.SnapshotFidelityValue(req.GetSnapshot().GetFidelity())
-
-	if _, err = client.Checkpoint(ctx, req); err != nil {
-		return wireFidelity, handleAteletError(ctx, w.store, actorRef, ateattr.OperationSuspend, "Checkpoint", false, err)
-	}
-	return wireFidelity, nil
-}
-
-// ensurePausedFinalized releases the actor's worker (only when it is still
-// owned by this actor), records where the local snapshot lives, and commits
-// SUSPENDED with the assignment cleared in a single update. The actor keeps
-// its assigned node: a paused actor is a SUSPENDED actor whose latest
-// snapshot is node-local, and it resumes on that node. When the worker's
-// node name was lost or the latest local snapshot entry is missing, the actor
-// can never be resumed from local state, so it commits CRASHED instead and
-// fails the step with FailedPrecondition. It re-reads the actor first so an
-// out-of-band transition (e.g. the syncer crashing the actor after its worker
-// died) is not overwritten: an actor that is no longer SUSPENDING is left as
-// it is and the step fails with FailedPrecondition.
-func (w *ActorWorkflow) ensurePausedFinalized(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) {
-	ctx, done := stepSpan(ctx, "FinalizePaused")
-	defer func() { err = done(err) }()
-
-	latestActor, err := w.store.GetActor(ctx, actorRef)
-	if err != nil {
-		return nil, err
-	}
-	if state := latestActor.GetStatus().GetState(); state != ateapipb.ActorState_ACTOR_STATE_SUSPENDING {
-		return nil, apierror.FailedPrecondition("FinalizePaused prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_SUSPENDING)
-	}
-
-	// 1. Free the worker (if it hasn't been freed yet)
-	if assignment := latestActor.GetStatus().GetWorkerAssignment(); assignment != nil {
-		if latestActor.GetStatus().GetAssignedNode() == "" {
-			// Without a node name we cannot record where the local snapshot lives,
-			// so the actor can never be resumed (the scheduler would search for a
-			// worker on an unknown node forever). Crash it instead of committing
-			// SUSPENDED with a snapshot nothing can find.
-			return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotNodeUnknown)
-		}
-		if snapshotAtLatestGeneration(latestActor.GetStatus()) == nil {
-			return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotMissing)
-		}
-
-		worker, err := w.store.GetWorker(ctx, assignment.GetWorker().GetName())
-		if err != nil {
-			if !errors.Is(err, store.ErrNotFound) {
-				return nil, fmt.Errorf("while getting worker for release: %w", err)
-			}
-			slog.Warn("Worker already gone during finalize pause, skipping release", "worker", assignment.GetWorkerPod())
-		} else {
-			// Drop just this actor's assignment; any other actors the worker
-			// hosts keep theirs.
-			_, err := w.store.ReleaseActorFromWorker(ctx, worker.GetMetadata().GetName(), latestActor.GetMetadata().GetUid())
-			if err != nil {
-				if errors.Is(err, store.ErrVersionConflict) {
-					return nil, apierror.Aborted("concurrent update conflict, please retry")
-				}
-				return nil, err
-			}
-		}
-
-		// 2. Clear the actor's assignment, now that the worker is freed. The
-		// precondition pins the version checked above, so an out-of-band write
-		// since then fails the update instead of being overwritten.
-		storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(latestActor), func(toUpdate *ateapipb.Actor) error {
-			toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDED
-			// Record the node locality of the local checkpoint and clear older local checkpoints.
-			snap := snapshotAtLatestGeneration(toUpdate.Status)
-			snap.Locality = toUpdate.GetStatus().GetAssignedNode()
-			pruneSnapshots(toUpdate.Status, snap)
-			toUpdate.Status.WorkerAssignment = nil
-			return nil
-		})
-		if err != nil {
-			if errors.Is(err, store.ErrVersionConflict) {
-				return nil, apierror.Aborted("concurrent update conflict, please retry")
-			}
-			return nil, err
-		}
-		logActorStateChanged(ctx, storedActor, ateattr.OperationSuspend)
-		latestActor = storedActor
-	}
-
-	return latestActor, nil
-}
diff --git a/cmd/ateapi/internal/controlapi/workflow_pause_test.go b/cmd/ateapi/internal/controlapi/workflow_pause_test.go
index 89743a9fa5..5965d224b8 100644
--- a/cmd/ateapi/internal/controlapi/workflow_pause_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_pause_test.go
@@ -17,6 +17,7 @@ package controlapi
 import (
 	"context"
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
@@ -408,6 +409,10 @@ func TestPauseActor_MarkedPausing_StateMatrix(t *testing.T) {
 		if got := apierror.Code(err); got != codes.FailedPrecondition {
 			t.Errorf("state %v: code = %v, want FailedPrecondition (err: %v)", seedState, got, err)
 		}
+		// Only a sync suspend accepts a paused actor, so a pause must not offer it.
+		if strings.Contains(err.Error(), "node-local snapshot") {
+			t.Errorf("state %v: err = %v, want no mention of a node-local snapshot", seedState, err)
+		}
 		if got := stored.GetStatus().GetState(); got != seedState {
 			t.Errorf("state %v: stored state = %v, want it untouched", seedState, got)
 		}
diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go
index f8ae72fa36..4dfdb02d0b 100644
--- a/cmd/ateapi/internal/controlapi/workflow_suspend.go
+++ b/cmd/ateapi/internal/controlapi/workflow_suspend.go
@@ -32,13 +32,50 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+// suspendMode selects which of the two suspends the suspend workflow runs.
+// Both take the same steps; the mode only decides the shape of the snapshot
+// they take.
+type suspendMode int
+
+const (
+	// suspendSync makes the snapshot durable before the workflow returns:
+	// the actor ends SUSPENDED holding a durable snapshot (SuspendActor).
+	suspendSync suspendMode = iota
+	// suspendAsync leaves the snapshot on the worker's node, to be uploaded
+	// later: the actor ends SUSPENDED holding a node-local snapshot and
+	// pinned to that node (PauseActor). Nothing uploads it yet.
+	suspendAsync
+)
+
+// uploadMode is when atelet makes the checkpoint durable.
+func (m suspendMode) uploadMode() ateletpb.UploadMode {
+	if m == suspendAsync {
+		return ateletpb.UploadMode_UPLOAD_MODE_DEFERRED
+	}
+	return ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE
+}
+
 // SuspendActor executes the workflow to suspend a running or paused actor:
 // a running actor is checkpointed on its worker, a paused actor's node-local
 // snapshot is uploaded. Idempotent: a re-entered workflow fast-forwards past
 // the steps a previous attempt completed, deriving progress from the
 // persisted actor alone.
-func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) {
+func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error) {
+	ctx = preview.ContextWithOptIn(ctx)
+	return w.suspendActor(ctx, actorRef, suspendSync)
+}
+
+// PauseActor executes the workflow to pause a running actor: it is
+// checkpointed on its worker and the snapshot is kept on that node.
+// Idempotent: a re-entered workflow fast-forwards past the steps a previous
+// attempt completed, deriving progress from the persisted actor alone.
+func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error) {
 	ctx = preview.ContextWithOptIn(ctx)
+	return w.suspendActor(ctx, actorRef, suspendAsync)
+}
+
+// suspendActor runs the suspend workflow behind SuspendActor and PauseActor.
+func (w *ActorWorkflow) suspendActor(ctx context.Context, actorRef resources.ActorRef, mode suspendMode) (_ *ateapipb.Actor, err error) {
 	start := time.Now()
 	var actor *ateapipb.Actor
 	var actorTemplate *ateapipb.ActorTemplate
@@ -65,31 +102,39 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.Act
 	if err != nil {
 		return nil, err
 	}
-	if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED && !latestSnapshotIsLocalOnly(actor.GetStatus()) {
-		// Fully suspended already: FinalizeSuspended commits SUSPENDED and the
-		// cleared worker assignment in a single update, so there is nothing
-		// left to do. A SUSPENDED actor whose latest snapshot is node-local is
-		// paused, not suspended, and goes on to upload it. This success
-		// reports no pool, and cannot: the previous attempt released the
-		// worker, so the record names none (#957).
-		return actor, nil
+	// FinalizeSuspended commits SUSPENDED and the cleared worker assignment in
+	// a single update, so an actor already holding the snapshot this mode
+	// leaves behind has nothing left to do. These successes report no pool,
+	// and cannot: the previous attempt released the worker, so the record
+	// names none (#957).
+	if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED {
+		if mode == suspendAsync && latestSnapshotIsLocalOnly(actor.GetStatus()) {
+			return actor, nil // fully paused
+		}
+		if mode == suspendSync && !latestSnapshotIsLocalOnly(actor.GetStatus()) {
+			return actor, nil // fully suspended
+		}
+		// A sync suspend of a paused actor goes on to upload its node-local
+		// snapshot. An async suspend of a suspended actor is rejected by MarkSuspending.
 	}
 	// Decided before marking: once SUSPENDING is committed, the loaded status
 	// alone can no longer tell the two origins apart.
-	fromPaused := isPausedOriginSuspend(actor)
+	// TODO(paused-suspend): remove the paused -> suspended transition before GA.
+	fromPaused := mode == suspendSync && isPausedOriginSuspend(actor)
 	var marked *ateapipb.Actor
-	if marked, err = w.ensureMarkedSuspending(leaseCtx, actorRef, actor, actorTemplate); err != nil {
+	if marked, err = w.ensureMarkedSuspending(leaseCtx, actorRef, actor, actorTemplate, mode); err != nil {
 		return nil, err
 	}
 	actor = marked
 	if fromPaused {
 		wireFidelity, err = w.ensurePausedSnapshotUploaded(leaseCtx, actorRef, actor, actorTemplate)
 	} else {
-		wireFidelity, err = w.ensureAteletSuspended(leaseCtx, actorRef, actor, actorTemplate)
+		wireFidelity, err = w.ensureAteletSuspended(leaseCtx, actorRef, actor, actorTemplate, mode)
 	}
 	if err != nil {
 		return nil, err
 	}
+	// TODO(#2734): An async suspend should skip detaching here, since the actor can resume on the same node.
 	if err = w.ensureVolumesDetached(leaseCtx, actor, actorTemplate, "DetachVolumes", ateattr.OperationSuspend); err != nil {
 		return nil, err
 	}
@@ -97,7 +142,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.Act
 	// them here, as crash.go does for the crash counter.
 	finalAttrs = lifecycleOpAttrs(actor, actorTemplate, "", wireFidelity)
 	var finalized *ateapipb.Actor
-	if finalized, err = w.ensureSuspendedFinalized(leaseCtx, actorRef); err != nil {
+	if finalized, err = w.ensureSuspendedFinalized(leaseCtx, actorRef, mode); err != nil {
 		return nil, err
 	}
 	actor = finalized
@@ -121,28 +166,44 @@ func (w *ActorWorkflow) loadActorForSuspend(ctx context.Context, actorRef resour
 }
 
 // ensureMarkedSuspending transitions a RUNNING or paused actor to
-// SUSPENDING, minting the in-progress snapshot location and recording the
-// actor version the snapshot will capture. Skips when a previous attempt
-// already marked the actor; the persisted location and source version then
-// stay authoritative for the rest of the workflow.
-func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (_ *ateapipb.Actor, err error) {
+// SUSPENDING, minting the in-progress snapshot and recording the actor
+// version the snapshot will capture. A sync suspend mints a pending durable
+// snapshot; an async one mints a node-local snapshot with no locality yet.
+// Skips when a previous attempt of the same mode already marked the actor;
+// the persisted snapshot and source version then stay authoritative for the
+// rest of the workflow. A mark left by the other mode is rejected: that
+// workflow owns the actor until it finishes.
+func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, mode suspendMode) (_ *ateapipb.Actor, err error) {
 	ctx, done := stepSpan(ctx, "MarkSuspending")
 	defer func() { err = done(err) }()
 
 	if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDING {
-		if latestSnapshotIsPendingLocal(actor.GetStatus()) {
-			// The pause workflow owns the actor until it finishes; a suspend
-			// can run once the actor is paused.
-			return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (a pause is in progress)", actorRef)
+		// An async mark's latest snapshot is node-local with no locality yet,
+		// a sync mark's has a pending durable copy.
+		if latestSnapshotIsPendingLocal(actor.GetStatus()) != (mode == suspendAsync) {
+			inProgress := "Pause"
+			if mode == suspendAsync {
+				inProgress = "Suspend"
+			}
+			return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (a previous %s is in progress)", actorRef, inProgress)
 		}
 		markSkipped(ctx, "actor already SUSPENDING")
 		return actor, nil
 	}
 	state := actor.GetStatus().GetState()
-	paused := isPausedOriginSuspend(actor)
+	// Only a sync suspend takes a paused actor.
+	paused := mode == suspendSync && isPausedOriginSuspend(actor) // TODO(paused-suspend)
 	if state != ateapipb.ActorState_ACTOR_STATE_RUNNING && !paused {
+		if mode == suspendAsync {
+			return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_RUNNING)
+		}
 		return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (got: %v, want %s or %s with a node-local snapshot)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_RUNNING, ateapipb.ActorState_ACTOR_STATE_SUSPENDED)
 	}
+	// Golden actors are only suspended, never paused.
+	// TODO: Remove this check: the suspend workflow should not special-case golden actors.
+	if mode == suspendAsync && actorRef.Atespace == resources.GoldenActorAtespace {
+		return nil, apierror.FailedPrecondition("actors in atespace %q are golden actors, which cannot be paused", actorRef.Atespace)
+	}
 	// A paused actor's durable snapshot is an upload of its local checkpoint,
 	// so it keeps that checkpoint's name; a running actor's is a new one.
 	var snapshotName string
@@ -154,10 +215,14 @@ func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef res
 	}
 
 	// Fail here rather than at checkpoint time if the template's location
-	// cannot produce a usable URI: nothing has been written yet.
-	uri, err := newInProgressSnapshotURI(actorTemplate, actor, snapshotName)
-	if err != nil {
-		return nil, err
+	// cannot produce a usable URI: nothing has been written yet. An async
+	// suspend keeps its snapshot on the node, so it needs no URI.
+	var uri resources.SnapshotURI
+	if mode == suspendSync {
+		// TODO: Always populate the in-progress URI.
+		if uri, err = newInProgressSnapshotURI(actorTemplate, actor, snapshotName); err != nil {
+			return nil, err
+		}
 	}
 	storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error {
 		toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDING
@@ -176,18 +241,25 @@ func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef res
 			return nil
 		}
 		// Increment last_assigned_generation for the new Suspend request and create
-		// a pending snapshot for that generation.
+		// a pending snapshot for that generation: durable for a sync suspend,
+		// node-local with no locality until finalize for an async one.
 		gen := toUpdate.Status.LastAssignedGeneration + 1
 		toUpdate.Status.LastAssignedGeneration = gen
-		snap := newDurableSnapshot(
-			gen,
-			ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR,
-			preferredFidelity(actorRef.Atespace, actorTemplate),
-			actorTemplate.GetMetadata().GetUid(),
-			uri.Name(),
-			uri.String(),
-			ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING,
-		)
+
+		fidelity := preferredFidelity(actorRef.Atespace, actorTemplate)
+		templateUID := actorTemplate.GetMetadata().GetUid()
+		snap := newLocalSnapshot(gen, fidelity, templateUID, snapshotName, "")
+		if mode == suspendSync {
+			snap = newDurableSnapshot(
+				gen,
+				ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR,
+				fidelity,
+				templateUID,
+				uri.Name(),
+				uri.String(),
+				ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING,
+			)
+		}
 		snap.SandboxRuntime = proto.CloneOf(toUpdate.GetStatus().GetWorkerAssignment().GetSandboxRuntime())
 		toUpdate.Status.Snapshots = append(toUpdate.Status.Snapshots, snap)
 		return nil
@@ -213,6 +285,7 @@ func preferredFidelity(atespace string, tmpl *ateapipb.ActorTemplate) ateapipb.S
 	return tmpl.GetSnapshotConfig().GetPreferredFidelity()
 }
 
+// TODO(paused-suspend): remove with the paused -> suspended transition.
 // isPausedOriginSuspend reports whether the suspend must upload a paused
 // actor's node-local snapshot instead of checkpointing a running workload.
 // A local snapshot alone does not mean paused-origin: resume restores from it
@@ -232,12 +305,14 @@ func isPausedOriginSuspend(actor *ateapipb.Actor) bool {
 }
 
 // ensureAteletSuspended checkpoints the workload to the actor's persisted
-// in-progress snapshot location. This is the atelet reentrancy seam (#372):
+// in-progress snapshot location. A sync suspend has atelet upload it before
+// returning; an async suspend has atelet keep it on the node.
+// This is the atelet reentrancy seam (#372):
 // the request is keyed by the actor UID, the worker pod UID, and the
 // once-minted snapshot location, so a re-entered workflow re-sends the same
 // semantic request; once atelet's Checkpoint is idempotent on those keys this
 // step becomes fully reentrant with no changes here.
-func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (wireFidelity string, err error) {
+func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, mode suspendMode) (wireFidelity string, err error) {
 	ctx, done := stepSpan(ctx, "CallAteletSuspend")
 	defer func() { err = done(err) }()
 
@@ -263,6 +338,11 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso
 	// into the snapshot manifest.
 	inProgressSnap := snapshotAtLatestGeneration(actor.GetStatus())
 	fidelity := fidelityToAtelet(preferredFidelity(actor.GetMetadata().GetAtespace(), actorTemplate))
+	// An async suspend's snapshot has no durable copy, so it stays on the node.
+	var object *ateletpb.ObjectStorage
+	if uri := inProgressSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(); uri != "" {
+		object = &ateletpb.ObjectStorage{SnapshotUri: uri}
+	}
 	req := &ateletpb.CheckpointRequest{
 		WorkerPodUid:          assignment.GetWorkerPodUid(),
 		Atespace:              actor.GetMetadata().GetAtespace(),
@@ -274,16 +354,10 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso
 			SnapshotUuid: inProgressSnap.GetUuid(),
 			Generation:   inProgressSnap.GetGeneration(),
 			Fidelity:     fidelity,
-			Object: &ateletpb.ObjectStorage{
-				SnapshotUri: inProgressSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(),
-			},
+			Object:       object,
 		},
-		// FinalizeSuspended marks this snapshot COMPLETED, so it must be
-		// durable before Checkpoint returns. IMMEDIATE temporarily keeps no
-		// local copy either (see UPLOAD_MODE_IMMEDIATE), which matches
-		// FinalizeSuspended dropping the actor's local snapshots.
 		UploadConfig: &ateletpb.UploadConfig{
-			Mode: ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE,
+			Mode: mode.uploadMode(),
 		},
 		ActorUid: actor.GetMetadata().Uid,
 	}
@@ -295,6 +369,7 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso
 	return wireFidelity, nil
 }
 
+// TODO(paused-suspend): remove with the paused -> suspended transition.
 // ensurePausedSnapshotUploaded suspends a paused actor by telling the atelet
 // on the node holding the local pause snapshot to upload it to the actor's
 // persisted in-progress snapshot location; no workload runs, so there is no
@@ -311,8 +386,8 @@ func (w *ActorWorkflow) ensurePausedSnapshotUploaded(ctx context.Context, actorR
 	}
 	nodeName := actor.GetStatus().GetAssignedNode()
 	if nodeName == "" {
-		// Without the node the snapshot can never be found (mirrors
-		// FinalizePaused, which crashes rather than record an unknown node).
+		// Without the node the snapshot can never be found (mirrors an async
+		// FinalizeSuspended, which crashes rather than record an unknown node).
 		return "", crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotNodeUnknown)
 	}
 
@@ -363,7 +438,7 @@ func newInProgressSnapshotURI(actorTemplate *ateapipb.ActorTemplate, actor *atea
 
 // ensureVolumesDetached detaches the actor's mounted external volumes from
 // its worker node. Detachment is idempotent, so a re-entered workflow safely
-// runs it again. spanName distinguishes the suspend and pause steps in
+// runs it again. spanName distinguishes the suspend and revert steps in
 // traces; op labels the volume metrics.
 // TODO replace re-execution with a proper check on the volumes' attach state.
 func (w *ActorWorkflow) ensureVolumesDetached(ctx context.Context, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, spanName, op string) (err error) {
@@ -380,15 +455,16 @@ func (w *ActorWorkflow) ensureVolumesDetached(ctx context.Context, actor *ateapi
 
 // ensureSuspendedFinalized releases the actor's worker (only when it is still
 // owned by this actor), records the in-progress snapshot as the actor's
-// durable snapshot, and commits SUSPENDED with the assignment cleared in a
-// single update, then releases the durable snapshot that update replaced.
-// When the latest snapshot has no durable snapshot to record, it commits
-// CRASHED instead and fails the step.
+// durable snapshot (sync) or node-local snapshot (async), and commits
+// SUSPENDED with the assignment cleared in a single update, then releases the
+// durable snapshot that update replaced.
+// When the latest snapshot cannot be recorded, it commits CRASHED instead and
+// fails the step.
 // It re-reads the actor first so an out-of-band transition (e.g. the syncer
 // crashing the actor after its worker died) is not overwritten: an actor that
 // is no longer SUSPENDING is left as it is and the step fails with
 // FailedPrecondition.
-func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) {
+func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef resources.ActorRef, mode suspendMode) (_ *ateapipb.Actor, err error) {
 	ctx, done := stepSpan(ctx, "FinalizeSuspended")
 	defer func() { err = done(err) }()
 
@@ -417,8 +493,19 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r
 	if state := latestActor.GetStatus().GetState(); state != ateapipb.ActorState_ACTOR_STATE_SUSPENDING {
 		return nil, apierror.FailedPrecondition("FinalizeSuspended prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_SUSPENDING)
 	}
-	if snapshotAtLatestGeneration(latestActor.GetStatus()).GetDurableSnapshot() == nil {
-		return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageDurableSnapshotMissing)
+	switch mode {
+	case suspendSync:
+		if snapshotAtLatestGeneration(latestActor.GetStatus()).GetDurableSnapshot() == nil {
+			return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageDurableSnapshotMissing)
+		}
+	case suspendAsync:
+		// A local snapshot on an unknown node can never be found by the scheduler.
+		if latestActor.GetStatus().GetAssignedNode() == "" {
+			return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotNodeUnknown)
+		}
+		if snapshotAtLatestGeneration(latestActor.GetStatus()) == nil {
+			return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotMissing)
+		}
 	}
 
 	// 1. Free the worker (if it hasn't been freed yet)
@@ -431,7 +518,7 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r
 		}
 	}
 
-	// 2. Finalize the actor: record its new durable snapshot and mark it SUSPENDED. This
+	// 2. Finalize the actor: record its new snapshot and mark it SUSPENDED. This
 	// must run even with no worker assignment (nothing to free), or the actor
 	// would be left SUSPENDING forever with the workflow reporting success.
 	// The precondition pins the version checked above, so an out-of-band write
@@ -439,14 +526,22 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r
 	t = time.Now()
 	storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(latestActor), func(toUpdate *ateapipb.Actor) error {
 		toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDED
-		inProgressSt := snapshotAtLatestGeneration(toUpdate.Status).GetDurableSnapshot()
-		inProgressSt.Status = ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
-		// Remove all local snapshots since they are superseded by the new durable snapshot.
-		// TODO: This will change when we merge pause and suspend and allow for both local
-		// and durable snapshots to be kept.
-		pruneSnapshots(toUpdate.Status, nil)
+		snap := snapshotAtLatestGeneration(toUpdate.Status)
+		switch mode {
+		case suspendSync:
+			snap.GetDurableSnapshot().Status = ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
+			// Remove all local snapshots since they are superseded by the new durable snapshot.
+			// TODO: This will change when an async suspend uploads its snapshot
+			// and both local and durable snapshots are kept.
+			pruneSnapshots(toUpdate.Status, nil)
+			toUpdate.Status.AssignedNode = ""
+		case suspendAsync:
+			// Record the node locality of the local checkpoint and clear older local checkpoints.
+			// AssignedNode is kept: a paused actor resumes on that node.
+			snap.Locality = toUpdate.GetStatus().GetAssignedNode()
+			pruneSnapshots(toUpdate.Status, snap)
+		}
 		toUpdate.Status.WorkerAssignment = nil
-		toUpdate.Status.AssignedNode = ""
 		return nil
 	})
 	dUpdateActor = time.Since(t)
diff --git a/cmd/ateapi/internal/controlapi/workflow_test.go b/cmd/ateapi/internal/controlapi/workflow_test.go
index 3b42a501d2..f60920fc40 100644
--- a/cmd/ateapi/internal/controlapi/workflow_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_test.go
@@ -71,18 +71,13 @@ func TestActorStateChangeRecords(t *testing.T) {
 			seedOpts: []func(*ateapipb.Actor){
 				withPauseInProgress(),
 				func(a *ateapipb.Actor) {
-					a.Status.AssignedNode = "node1"
-					a.Status.WorkerAssignment = &ateapipb.WorkerAssignment{
-						Worker:          &ateapipb.ObjectRef{Name: "worker-gone"},
-						WorkerNamespace: "default",
-						WorkerPool:      "pool1",
-						WorkerPod:       "worker-pod-1",
-					}
+					a.Status.AssignedNode = "node-1"
+					a.Status.WorkerAssignment = wireTestAssignment()
 				},
 			},
 			transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, _ *ateapipb.Actor, _ *ateapipb.ActorTemplate) {
-				if _, err := w.ensurePausedFinalized(context.Background(), ref); err != nil {
-					t.Fatalf("ensurePausedFinalized: %v", err)
+				if _, err := w.PauseActor(context.Background(), ref); err != nil {
+					t.Fatalf("PauseActor: %v", err)
 				}
 			},
 			wantOp:     ateattr.OperationSuspend,
@@ -131,7 +126,7 @@ func TestActorStateChangeRecords(t *testing.T) {
 				t.Fatalf("get template: %v", err)
 			}
 
-			w := &ActorWorkflow{store: persistence}
+			w, _ := newWireCaptureWorkflow(t, persistence)
 			tt.transition(t, w, actorRef, actor, tmpl)
 
 			// The logs exporter is on, so the record goes to OTLP only.