From 749f49464b04dfd1efe057970fc608e3a2cb8c22 Mon Sep 17 00:00:00 2001
From: zoezhao <zoezhao@google.com>
Date: Wed, 21 Oct 2026 11:53:51 -0700
Subject: [PATCH 1/7] ateapi, atelet: keep an upload Terminate reports on
 revert

atelet can upload a snapshot before its report reaches ateapi, and
ateapi refuses reports for a CRASHED actor. Revert then deleted the
upload as in progress.

Terminate now returns the actor's snapshot state as atelet forgets it,
and revert records its latest durable snapshot COMPLETED before it
discards in-progress snapshots.
---
 .../internal/controlapi/snapshot_report.go    |   6 +-
 .../internal/controlapi/workflow_delete.go    |  30 +--
 .../controlapi/workflow_delete_test.go        |   2 +-
 .../controlapi/workflow_resume_test.go        |   8 +-
 .../internal/controlapi/workflow_revert.go    |  63 +++++--
 .../controlapi/workflow_revert_test.go        | 131 ++++++++-----
 .../snapshottracker/background_test.go        |   6 +-
 .../internal/snapshottracker/detach_test.go   |   3 +-
 .../internal/snapshottracker/lifecycle.go     |  14 +-
 .../internal/snapshottracker/metrics_test.go  |   6 +-
 .../internal/snapshottracker/report_test.go   |   2 +-
 .../internal/snapshottracker/tracker_test.go  |   6 +-
 .../internal/snapshottracker/upload_test.go   |  26 ++-
 cmd/atelet/lifecycle_test.go                  |  12 +-
 cmd/atelet/main.go                            |   5 +-
 internal/proto/ateletpb/atelet.pb.go          | 176 ++++++++++--------
 internal/proto/ateletpb/atelet.proto          |   9 +-
 internal/proto/ateletpb/atelet_grpc.pb.go     |   8 +-
 18 files changed, 327 insertions(+), 186 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/snapshot_report.go b/cmd/ateapi/internal/controlapi/snapshot_report.go
index 4b5ef70f28..549dc63aa3 100644
--- a/cmd/ateapi/internal/controlapi/snapshot_report.go
+++ b/cmd/ateapi/internal/controlapi/snapshot_report.go
@@ -154,9 +154,9 @@ func applyActorSnapshotState(ctx context.Context, status *ateapipb.ActorStatus,
 }
 
 // applyAteletResponseState records an upload that atelet's state, as returned
-// by Checkpoint or PromoteSnapshot, says finished, with the same rules as a
-// report. It records the fidelity atelet reports, which for both RPCs is the
-// one requested. Unlike a report, a state with no local snapshot is not a
+// by Checkpoint, PromoteSnapshot, or Terminate, says finished, with the same
+// rules as a report. It records the fidelity atelet reports, which is the one
+// the upload was requested at. Unlike a report, a state with no local snapshot is not a
 // detach. The workflow that made the call records where its snapshot lives.
 func applyAteletResponseState(ctx context.Context, status *ateapipb.ActorStatus, state *ateletpb.ActorSnapshotState) error {
 	return applyLatestDurable(ctx, status, snapshotReportFromAtelet(state.GetLatestDurable()))
diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go
index 070f1a2966..33f7afc041 100644
--- a/cmd/ateapi/internal/controlapi/workflow_delete.go
+++ b/cmd/ateapi/internal/controlapi/workflow_delete.go
@@ -61,7 +61,7 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto
 	}
 
 	var atletTerminatedErr, volumesDetachedErr error
-	if err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete); err != nil {
+	if _, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete); err != nil {
 		atletTerminatedErr = fmt.Errorf("while terminating atelet: %w", err)
 		errs = append(errs, atletTerminatedErr)
 	}
@@ -96,15 +96,17 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto
 	return w.finalizeDeleted(ctx, actor)
 }
 
-// ensureAteletTerminated calls atelet to terminate the workload for opName.
-func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, opName string) (err error) {
+// ensureAteletTerminated calls atelet to terminate the workload for opName. It
+// returns the actor's state as atelet forgot it, or nil when it skipped the
+// call.
+func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, opName string) (_ *ateletpb.ActorSnapshotState, err error) {
 	ctx, done := stepSpan(ctx, "CallAteletTerminate")
 	defer func() { err = done(err) }()
 
 	nodeName := actor.GetStatus().GetAssignedNode()
 	if nodeName == "" {
 		slog.InfoContext(ctx, "actor has no assigned node, skipping atelet terminate request", slog.Any("actor", actorRef))
-		return nil
+		return nil, nil
 	}
 
 	workerPodUID := ""
@@ -116,7 +118,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 			// ordinary case, and the others are none of this delete's business.
 			hosted, err := workerHostsActor(ctx, w.store, workerName, actor.GetMetadata().GetUid())
 			if err != nil {
-				return err
+				return nil, err
 			}
 			if !hosted {
 				slog.InfoContext(ctx, "worker is no longer assigned to this actor, skipping worker workload termination",
@@ -131,18 +133,18 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 	if errors.Is(err, ErrNoAteletOnNode) {
 		gone, nodeErr := w.dialer.NodeGone(ctx, nodeName)
 		if nodeErr != nil {
-			return fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr)
+			return nil, fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr)
 		}
 		if gone {
 			// A deleted Node object counts as the node being gone for good. If the
 			// machine still runs, whatever atelet holds for the actor stays there.
 			slog.WarnContext(ctx, "assigned node no longer exists, skipping atelet terminate request",
 				slog.String("node", nodeName), slog.Any("actor", actorRef))
-			return nil
+			return nil, nil
 		}
 	}
 	if err != nil {
-		return fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err)
+		return nil, fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err)
 	}
 
 	client := ateletpb.NewAteletClient(conn)
@@ -151,7 +153,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 	if actorTemplate != nil {
 		spec, err := workloadSpecFromActorTemplate(actorTemplate, actor, nil)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		workloadSpec = spec
 	} else {
@@ -174,15 +176,15 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 		Spec:                  workloadSpec,
 	}
 
-	if _, err := client.Terminate(ctx, req); err != nil {
+	resp, err := client.Terminate(ctx, req)
+	if err != nil {
 		if status.Code(err) == codes.NotFound {
 			slog.InfoContext(ctx, "workload already terminated on atelet", slog.Any("actor", actorRef))
-			return nil
+			return nil, nil
 		}
-		return handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err)
+		return nil, handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err)
 	}
-
-	return nil
+	return resp.GetState(), nil
 }
 
 // fallbackWorkloadSpec is the spec to terminate an actor with when its template
diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go
index 76bb5ff02f..31d6a11cca 100644
--- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go
@@ -124,7 +124,7 @@ func TestEnsureAteletTerminated_NodeGone(t *testing.T) {
 			w := newTestActorWorkflow(t, newTestPersistence(t), "team-a", "tmpl")
 			w.dialer = NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "",
 				WithNodeGetter(nodeGetter(tc.lookupErr, tc.nodes...)))
-			err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert")
+			_, err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert")
 			if tc.wantErr != (err != nil) {
 				t.Fatalf("ensureAteletTerminated = %v, want error %v", err, tc.wantErr)
 			}
diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go
index 21a5c5e64c..531345864c 100644
--- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go
@@ -1151,6 +1151,9 @@ type capturingAtelet struct {
 	onDetach     func()
 	terminateErr error
 	onTerminate  func()
+	// terminateState, when set, is the snapshot state Terminate returns.
+	// Terminate is unimplemented otherwise.
+	terminateState *ateletpb.ActorSnapshotState
 }
 
 func (f *capturingAtelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (*ateletpb.RestoreResponse, error) {
@@ -1218,11 +1221,14 @@ func (f *capturingAtelet) Terminate(ctx context.Context, req *ateletpb.Terminate
 		hook()
 	}
 	f.mu.Lock()
-	err := f.terminateErr
+	err, state := f.terminateErr, proto.CloneOf(f.terminateState)
 	f.mu.Unlock()
 	if err != nil {
 		return nil, err
 	}
+	if state != nil {
+		return &ateletpb.TerminateResponse{State: state}, nil
+	}
 	return f.UnimplementedAteletServer.Terminate(ctx, req)
 }
 
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go
index 55c239fe25..8c55251226 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert.go
@@ -26,16 +26,18 @@ import (
 	"go.ate.dev/substrate/internal/apierror"
 	"go.ate.dev/substrate/internal/ateattr"
 	"go.ate.dev/substrate/internal/preview"
+	"go.ate.dev/substrate/internal/proto/ateletpb"
 	"go.ate.dev/substrate/internal/resources"
 	"go.ate.dev/substrate/pkg/proto/ateapipb"
 	"go.opentelemetry.io/otel/attribute"
+	"google.golang.org/protobuf/proto"
 )
 
 // revertableStates are the states a revert is accepted from. A SUSPENDED
 // actor is also accepted when its latest snapshot is node-local: the revert
 // discards the node-local copy and returns the actor to its latest durable
-// snapshot, which is that same snapshot once atelet's upload of it is
-// recorded. See isRevertable.
+// snapshot, which is that same snapshot once atelet has uploaded it. See
+// isRevertable.
 var revertableStates = []ateapipb.ActorState{
 	ateapipb.ActorState_ACTOR_STATE_RUNNING,
 	ateapipb.ActorState_ACTOR_STATE_CRASHED,
@@ -91,9 +93,11 @@ func (w *ActorWorkflow) RevertActor(ctx context.Context, actorRef resources.Acto
 	}
 	actor = marked
 
-	if err = w.ensureWorkerDiscarded(leaseCtx, actorRef, actor, actorTemplate); err != nil {
+	var discarded *ateapipb.Actor
+	if discarded, err = w.ensureWorkerDiscarded(leaseCtx, actorRef, actor, actorTemplate); err != nil {
 		return nil, err
 	}
+	actor = discarded
 
 	if err = w.ensureInProgressSnapshotDiscarded(leaseCtx, actor); err != nil {
 		return nil, err
@@ -181,34 +185,69 @@ func (w *ActorWorkflow) ensureMarkedReverting(ctx context.Context, actorRef reso
 // worker. An assignment commits before the actor is updated to point at it, so
 // a crash in between leaves a row nothing references — and a CRASHED actor is
 // the likeliest holder of one. Absence is the ordinary case, not an error.
-func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (err error) {
+//
+// It returns the actor as stored after the upload Terminate reports is
+// recorded.
+func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (_ *ateapipb.Actor, err error) {
 	ctx, done := stepSpan(ctx, "DiscardWorker")
 	defer func() { err = done(err) }()
 
-	if actor.GetStatus().GetAssignedNode() != "" {
+	if node := actor.GetStatus().GetAssignedNode(); node != "" {
 		forgetNode := false
-		if err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert); err != nil {
+		state, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert)
+		if err != nil {
 			if !canDetachWithoutAtelet(ctx, actor, err) {
-				return err
+				return nil, err
 			}
 			forgetNode = true
 		}
+		if actor, err = w.recordTerminatedUpload(ctx, actorRef, actor, node, state); err != nil {
+			return nil, err
+		}
 		if err := w.ensureVolumesDetached(ctx, actor, actorTemplate, "DetachVolumesForRevert", ateattr.OperationRevert); err != nil {
-			return err
+			return nil, err
 		}
 		if forgetNode {
-			if _, err := w.recordDetach(ctx, actorRef, actor, actor.GetStatus().GetAssignedNode(), nil); err != nil {
-				return err
+			if actor, err = w.recordDetach(ctx, actorRef, actor, node, nil); err != nil {
+				return nil, err
 			}
 		}
 		if actor.GetStatus().GetWorkerAssignment() != nil {
 			if _, err := releaseWorker(ctx, w.store, actor); err != nil {
-				return fmt.Errorf("while releasing worker: %w", err)
+				return nil, fmt.Errorf("while releasing worker: %w", err)
 			}
 		}
 	}
 
-	return w.releaseAssignmentWithoutBacklink(ctx, actor)
+	return actor, w.releaseAssignmentWithoutBacklink(ctx, actor)
+}
+
+// recordTerminatedUpload marks COMPLETED the snapshot that the atelet on node
+// reported durable in state as it terminated the actor, with the same rules
+// as a report. atelet may upload a snapshot without its report landing, and a
+// CRASHED actor refuses reports, so without this the revert would delete the
+// upload as in progress. It writes at once because atelet forgot the actor, so
+// a retried Terminate returns nothing.
+func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, node string, state *ateletpb.ActorSnapshotState) (*ateapipb.Actor, error) {
+	applied := proto.CloneOf(actor.GetStatus())
+	if err := applyAteletResponseState(ctx, applied, state); err != nil {
+		return nil, apierror.Internal("atelet on node %q reported a durable snapshot for Actor %s that conflicts with its record: %w", node, actorRef, err)
+	}
+	if proto.Equal(applied, actor.GetStatus()) {
+		return actor, nil
+	}
+	stored, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error {
+		// The precondition makes toUpdate the version applied was built from.
+		toUpdate.Status.Snapshots = applied.GetSnapshots()
+		return nil
+	})
+	if err != nil {
+		if errors.Is(err, store.ErrVersionConflict) {
+			return nil, apierror.Aborted("concurrent update conflict, please retry")
+		}
+		return nil, fmt.Errorf("while recording the upload atelet on node %q reported for Actor %s: %w", node, actorRef, err)
+	}
+	return stored, nil
 }
 
 // ensureInProgressSnapshotDiscarded removes pending uploads of the discarded execution.
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
index 31635b8518..de2644ffc9 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
@@ -24,6 +24,7 @@ import (
 	"go.ate.dev/substrate/internal/apierror"
 	"go.ate.dev/substrate/internal/objectstore/objectstoretest"
 	"go.ate.dev/substrate/internal/objectstoreplugin/objectstoreplugintest"
+	"go.ate.dev/substrate/internal/proto/ateletpb"
 	"go.ate.dev/substrate/internal/resources"
 	"go.ate.dev/substrate/pkg/proto/ateapipb"
 	"google.golang.org/grpc/codes"
@@ -336,56 +337,96 @@ func TestRevertActor_UnreachableNode(t *testing.T) {
 	}
 }
 
-// TestRevertActor_PausedSnapshotAlreadyUploaded verifies a revert of a
-// suspended actor whose local snapshot atelet uploaded returns it to that
-// upload, its latest durable snapshot, and collects the durable snapshot the
-// upload replaced. The actor has no assigned node, so there is nothing to
-// terminate, as in TestRevertActor_ReturnsActorToItsSnapshot.
+// TestRevertActor_PausedSnapshotAlreadyUploaded verifies a revert returns the
+// actor to a local snapshot atelet uploaded, its latest durable snapshot, and
+// collects the durable snapshot the upload replaced. The revert learns of the
+// upload from the record, or from Terminate when atelet's report has not
+// landed. An upload atelet does not report is deleted as in progress.
 func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 	memory := ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY
 	completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
-	ctx := context.Background()
-	st, cleanup := storetest.SetupTestStore(t)
-	defer cleanup()
-	w := newTestActorWorkflow(t, st, "ns", "tmpl1")
-	objects := objectstoretest.New()
-	w.objectStorePlugin = objectstoreplugintest.ControlClient(objects)
+	pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING
+	for _, tc := range []struct {
+		name  string
+		state ateapipb.ActorState
+		// recorded is the upload's status on the record.
+		recorded ateapipb.SnapshotStorageStatus
+		// node is the actor's assigned node, which the revert terminates on.
+		node string
+		// terminateReports makes Terminate return the upload as durable.
+		terminateReports bool
+		wantUploadKept   bool
+	}{
+		{name: "upload recorded", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: completed, wantUploadKept: true},
+		{name: "paused, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1", terminateReports: true, wantUploadKept: true},
+		{name: "crashed, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, wantUploadKept: true},
+		{name: "upload not reported", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			st, cleanup := storetest.SetupTestStore(t)
+			defer cleanup()
+			w := newTestActorWorkflow(t, st, "ns", "tmpl1")
+			objects := objectstoretest.New()
+			w.objectStorePlugin = objectstoreplugintest.ControlClient(objects)
+			wire, atelet := newWireCaptureWorkflow(t, st)
+			w.dialer = wire.dialer
 
-	actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"}
-	seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDED)
-	actor, err := st.GetActor(ctx, actorRef)
-	if err != nil {
-		t.Fatalf("get seeded actor: %v", err)
-	}
-	// newTestActorWorkflow's template stores snapshots under gs://snapshots.
-	tmpl := &ateapipb.ActorTemplate{SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}}
-	old := mustActorSnapshotURI(t, tmpl, actor, "old")
-	uploaded := mustActorSnapshotURI(t, tmpl, actor, "pause-snap")
-	objects.PutSnapshot(t, old, "manifest.json")
-	objects.PutSnapshot(t, uploaded, "manifest.json")
-	mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapipb.ActorStatus) {
-		paused := newPausedSnapshot(t, 2, memory, "", "pause-snap", "node-1")
-		paused.DurableSnapshot = &ateapipb.SnapshotStorage{Status: completed, Object: &ateapipb.ObjectSnapshot{SnapshotUri: uploaded.String()}}
-		s.LastAssignedGeneration = 2
-		s.Snapshots = []*ateapipb.Snapshot{
-			newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, memory, "", old.Name(), old.String(), completed),
-			paused,
-		}
-	})
+			actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"}
+			seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", tc.state)
+			actor, err := st.GetActor(ctx, actorRef)
+			if err != nil {
+				t.Fatalf("get seeded actor: %v", err)
+			}
+			// newTestActorWorkflow's template stores snapshots under gs://snapshots.
+			tmpl := &ateapipb.ActorTemplate{SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}}
+			old := mustActorSnapshotURI(t, tmpl, actor, "old")
+			uploaded := mustActorSnapshotURI(t, tmpl, actor, "pause-snap")
+			objects.PutSnapshot(t, old, "manifest.json")
+			objects.PutSnapshot(t, uploaded, "manifest.json")
+			mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapipb.ActorStatus) {
+				paused := newPausedSnapshot(t, 2, memory, "", "pause-snap", "node-1")
+				paused.DurableSnapshot = &ateapipb.SnapshotStorage{Status: tc.recorded, Object: &ateapipb.ObjectSnapshot{SnapshotUri: uploaded.String()}}
+				s.AssignedNode = tc.node
+				s.LastAssignedGeneration = 2
+				s.Snapshots = []*ateapipb.Snapshot{
+					newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, memory, "", old.Name(), old.String(), completed),
+					paused,
+				}
+				if tc.state == ateapipb.ActorState_ACTOR_STATE_CRASHED {
+					s.Crash = newActorCrash("ResumeActor", "interrupted execution")
+				}
+			})
+			atelet.terminateErr = apierror.NotFound("atelet holds nothing for the actor")
+			if tc.terminateReports {
+				atelet.terminateErr = nil
+				atelet.terminateState = &ateletpb.ActorSnapshotState{LatestDurable: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{
+					SnapshotUuid: "pause-snap",
+					Generation:   2,
+					Fidelity:     ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY,
+					Object:       &ateletpb.ObjectStorage{SnapshotUri: uploaded.String()},
+				}}}
+			}
 
-	reverted, err := w.RevertActor(ctx, actorRef)
-	if err != nil {
-		t.Fatalf("RevertActor: %v", err)
-	}
-	want := []snapshotOutcome{{generation: 2, fidelity: memory, durable: completed}}
-	if diff := cmp.Diff(want, describeSnapshots(reverted.GetStatus()), cmp.AllowUnexported(snapshotOutcome{})); diff != "" {
-		t.Errorf("snapshots mismatch (-want +got):\n%s", diff)
-	}
-	if len(objects.Snapshot(t, old)) != 0 {
-		t.Error("the durable snapshot the upload replaced was not released")
-	}
-	if len(objects.Snapshot(t, uploaded)) == 0 {
-		t.Error("the upload the revert returned the actor to was released")
+			reverted, err := w.RevertActor(ctx, actorRef)
+			if err != nil {
+				t.Fatalf("RevertActor: %v", err)
+			}
+			want := []snapshotOutcome{{generation: 1, fidelity: memory, durable: completed}}
+			if tc.wantUploadKept {
+				want = []snapshotOutcome{{generation: 2, fidelity: memory, durable: completed}}
+			}
+			if diff := cmp.Diff(want, describeSnapshots(reverted.GetStatus()), cmp.AllowUnexported(snapshotOutcome{})); diff != "" {
+				t.Errorf("snapshots mismatch (-want +got):\n%s", diff)
+			}
+			// A kept upload replaces the old snapshot, which the revert collects.
+			if got, want := len(objects.Snapshot(t, old)) != 0, !tc.wantUploadKept; got != want {
+				t.Errorf("old snapshot kept = %v, want %v", got, want)
+			}
+			if got := len(objects.Snapshot(t, uploaded)) != 0; got != tc.wantUploadKept {
+				t.Errorf("uploaded snapshot kept = %v, want %v", got, tc.wantUploadKept)
+			}
+		})
 	}
 }
 
diff --git a/cmd/atelet/internal/snapshottracker/background_test.go b/cmd/atelet/internal/snapshottracker/background_test.go
index 9eb2c356c4..b8591f1eb3 100644
--- a/cmd/atelet/internal/snapshottracker/background_test.go
+++ b/cmd/atelet/internal/snapshottracker/background_test.go
@@ -481,7 +481,7 @@ func TestForgetWaitsForInFlightOp(t *testing.T) {
 	opCtx, op := beginOp(t, tr, s1)
 
 	forgetDone := make(chan error, 1)
-	go func() { forgetDone <- tr.Forget(trackedActorUID) }()
+	go func() { _, err := tr.Forget(trackedActorUID); forgetDone <- err }()
 	select {
 	case <-opCtx.Done():
 	case <-time.After(10 * time.Second):
@@ -557,7 +557,7 @@ func TestTrackerTerminateHoldsOffBackgroundWork(t *testing.T) {
 					<-opCtx.Done()
 					endOp(tr, op)
 				}()
-				if err := tr.Forget(trackedActorUID); err != nil {
+				if _, err := tr.Forget(trackedActorUID); err != nil {
 					t.Fatalf("Forget: %v", err)
 				}
 			}
@@ -665,7 +665,7 @@ func TestTrackerBackoff(t *testing.T) {
 		{"CommitDurable", func() { tr.CommitDurable(trackedActorUID, testRef("s2", 2, clk.Now())) }},
 		{"CommitPromoted", func() { tr.CommitPromoted(trackedActorUID, testRef("s3", 3, clk.Now())) }},
 		{"Forget", func() {
-			if err := tr.Forget(trackedActorUID); err != nil {
+			if _, err := tr.Forget(trackedActorUID); err != nil {
 				t.Fatalf("Forget: %v", err)
 			}
 		}},
diff --git a/cmd/atelet/internal/snapshottracker/detach_test.go b/cmd/atelet/internal/snapshottracker/detach_test.go
index 5b073f86d5..c4ce8e8cae 100644
--- a/cmd/atelet/internal/snapshottracker/detach_test.go
+++ b/cmd/atelet/internal/snapshottracker/detach_test.go
@@ -769,7 +769,8 @@ func TestTrackerForgetCancelsDetach(t *testing.T) {
 
 	forgetDone := make(chan error, 1)
 	go func() {
-		forgetDone <- tr.Forget(trackedActorUID)
+		_, err := tr.Forget(trackedActorUID)
+		forgetDone <- err
 	}()
 	if got := waitCanceled(t, store); got != "s1" {
 		t.Fatalf("canceled detach upload = %q, want s1", got)
diff --git a/cmd/atelet/internal/snapshottracker/lifecycle.go b/cmd/atelet/internal/snapshottracker/lifecycle.go
index 8e9284002b..2412130628 100644
--- a/cmd/atelet/internal/snapshottracker/lifecycle.go
+++ b/cmd/atelet/internal/snapshottracker/lifecycle.go
@@ -346,12 +346,15 @@ func (t *Tracker) holdLocked(actorUID string, a *actorSnapshots, kind holdKind)
 	}
 }
 
-// Forget drops the actor's state, in memory and on disk. It first cancels any
-// background upload or detach of the actor in progress and waits for it to
-// end, releasing the actor's lock meanwhile.
-func (t *Tracker) Forget(actorUID string) error {
+// Forget drops the actor's state, in memory and on disk, and returns the state
+// it dropped with only its latest durable snapshot, since the caller deletes
+// the actor's local snapshots. It first cancels any background upload or
+// detach of the actor in progress and waits for it to end, releasing the
+// actor's lock meanwhile.
+func (t *Tracker) Forget(actorUID string) (*ateletpb.ActorSnapshotState, error) {
 	t.lock(actorUID)
 	defer t.unlock(actorUID)
+	state := &ateletpb.ActorSnapshotState{}
 	if a := t.findLocked(actorUID); a != nil {
 		t.addLocked(a)
 		a.hold = holdTerminate
@@ -362,12 +365,13 @@ func (t *Tracker) Forget(actorUID string) error {
 			<-op.done
 			t.lock(actorUID)
 		}
+		state.LatestDurable = a.LatestDurable.report()
 	}
 	t.retries.Forget(actorUID)
 	t.reporting.retries.Forget(actorUID)
 	t.dropLocked(actorUID)
 	t.undurableActorsTally.forget(actorUID)
-	return removeSnapshotState(actorUID)
+	return state, removeSnapshotState(actorUID)
 }
 
 // setLatestLocalLocked records snap as the actor's latest local snapshot and
diff --git a/cmd/atelet/internal/snapshottracker/metrics_test.go b/cmd/atelet/internal/snapshottracker/metrics_test.go
index e0b65d303e..c377607ef0 100644
--- a/cmd/atelet/internal/snapshottracker/metrics_test.go
+++ b/cmd/atelet/internal/snapshottracker/metrics_test.go
@@ -441,7 +441,7 @@ func TestDetachDurationOutcomes(t *testing.T) {
 		// Forget cancels the detach as its upload finishes, and waits for it
 		// to end.
 		forgetDone := make(chan error, 1)
-		go func() { forgetDone <- tr.Forget(trackedActorUID) }()
+		go func() { _, err := tr.Forget(trackedActorUID); forgetDone <- err }()
 		waitCanceled(t, store)
 		store.release <- nil
 		returnsPromptly(t, "Forget", func() {
@@ -698,7 +698,7 @@ func TestUndurableActorsInstrumentsDropActor(t *testing.T) {
 		if got, want := readUndurable(t, reader), (undurableReading{age: 60}); got != want {
 			t.Errorf("before Forget: %+v, want %+v", got, want)
 		}
-		if err := tr.Forget(trackedActorUID); err != nil {
+		if _, err := tr.Forget(trackedActorUID); err != nil {
 			t.Fatalf("Forget: %v", err)
 		}
 		if got, want := readUndurable(t, reader), (undurableReading{}); got != want {
@@ -799,7 +799,7 @@ func TestUndurableActorsInstrumentsMultipleActors(t *testing.T) {
 		t.Errorf("split across bands: %+v, want %+v", got, want)
 	}
 
-	if err := tr.Forget(trackedActorUID); err != nil {
+	if _, err := tr.Forget(trackedActorUID); err != nil {
 		t.Fatalf("Forget: %v", err)
 	}
 	if got, want := readUndurable(t, reader), (undurableReading{
diff --git a/cmd/atelet/internal/snapshottracker/report_test.go b/cmd/atelet/internal/snapshottracker/report_test.go
index 65d378e181..6d0117357e 100644
--- a/cmd/atelet/internal/snapshottracker/report_test.go
+++ b/cmd/atelet/internal/snapshottracker/report_test.go
@@ -1213,7 +1213,7 @@ func TestTrackerForgetDuringReport(t *testing.T) {
 	waitReported(t, reporter)
 
 	returnsPromptly(t, "Forget during a report", func() {
-		if err := tr.Forget(trackedActorUID); err != nil {
+		if _, err := tr.Forget(trackedActorUID); err != nil {
 			t.Errorf("Forget: %v", err)
 		}
 	})
diff --git a/cmd/atelet/internal/snapshottracker/tracker_test.go b/cmd/atelet/internal/snapshottracker/tracker_test.go
index 30114f1e20..88ebe31e39 100644
--- a/cmd/atelet/internal/snapshottracker/tracker_test.go
+++ b/cmd/atelet/internal/snapshottracker/tracker_test.go
@@ -474,9 +474,13 @@ func TestTrackerState(t *testing.T) {
 		t.Errorf("State() on a fresh tracker = %v, want %v from the state file", got, st)
 	}
 
-	if err := tr.Forget(trackedActorUID); err != nil {
+	forgotten, err := tr.Forget(trackedActorUID)
+	if err != nil {
 		t.Fatalf("forget: %v", err)
 	}
+	if want := (&ateletpb.ActorSnapshotState{LatestDurable: st.GetLatestDurable()}); !proto.Equal(forgotten, want) {
+		t.Errorf("Forget returned %v, want %v: s2 durable and nothing local", forgotten, want)
+	}
 	if _, err := os.Stat(ateletpath.ActorSnapshotStateFile(trackedActorUID)); !os.IsNotExist(err) {
 		t.Errorf("state file after forget: stat = %v, want it deleted", err)
 	}
diff --git a/cmd/atelet/internal/snapshottracker/upload_test.go b/cmd/atelet/internal/snapshottracker/upload_test.go
index 863dd8dcfd..9347bf6e1b 100644
--- a/cmd/atelet/internal/snapshottracker/upload_test.go
+++ b/cmd/atelet/internal/snapshottracker/upload_test.go
@@ -346,21 +346,31 @@ func TestTrackerForgetCancelsUpload(t *testing.T) {
 	done := startUpload(t, tr, store)
 	tr.CommitLocal(trackedActorUID, testRef("s2", 2, tr.clock.Now()))
 
-	forgetDone := make(chan error, 1)
-	go func() { forgetDone <- tr.Forget(trackedActorUID) }()
+	type forgetResult struct {
+		state *ateletpb.ActorSnapshotState
+		err   error
+	}
+	forgetDone := make(chan forgetResult, 1)
+	go func() {
+		state, err := tr.Forget(trackedActorUID)
+		forgetDone <- forgetResult{state, err}
+	}()
 	if got := waitCanceled(t, store); got != "s1" {
 		t.Fatalf("canceled upload = %q, want s1", got)
 	}
 	select {
-	case err := <-forgetDone:
-		t.Fatalf("Forget returned (%v) before the canceled upload ended", err)
+	case res := <-forgetDone:
+		t.Fatalf("Forget returned (%v) before the canceled upload ended", res.err)
 	case <-time.After(50 * time.Millisecond):
 	}
 	store.release <- context.Canceled
 	select {
-	case err := <-forgetDone:
-		if err != nil {
-			t.Errorf("Forget: %v", err)
+	case res := <-forgetDone:
+		if res.err != nil {
+			t.Errorf("Forget: %v", res.err)
+		}
+		if got := res.state.GetLatestDurable(); got != nil {
+			t.Errorf("Forget returned latest durable %v, want nil: the canceled upload did not finish", got)
 		}
 	case <-time.After(10 * time.Second):
 		t.Fatal("timed out waiting for Forget to return")
@@ -1048,7 +1058,7 @@ func TestTrackerTerminateHoldsOffSupersededUploadingCleanup(t *testing.T) {
 			}
 
 			if tc.forget {
-				if err := tr.Forget(trackedActorUID); err != nil {
+				if _, err := tr.Forget(trackedActorUID); err != nil {
 					t.Fatalf("Forget: %v", err)
 				}
 				end()
diff --git a/cmd/atelet/lifecycle_test.go b/cmd/atelet/lifecycle_test.go
index 7bc276da1e..fc0af091fd 100644
--- a/cmd/atelet/lifecycle_test.go
+++ b/cmd/atelet/lifecycle_test.go
@@ -1143,7 +1143,11 @@ func TestTerminateWithoutWorkerPodUID(t *testing.T) {
 		Containers: []*ateletpb.Container{{Name: "app", Image: "example.com/app@sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"}},
 	}
 
-	if _, err := s.Terminate(ctx, &ateletpb.TerminateRequest{
+	// atelet uploaded the snapshot, and the control plane has not recorded it.
+	uploaded := lifecycleSnapshot(actorUID, snapshotName, 1, ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY)
+	s.snapshots.CommitDurable(actorUID, snapshottracker.SnapshotRef{Snapshot: uploaded})
+
+	resp, err := s.Terminate(ctx, &ateletpb.TerminateRequest{
 		Atespace:              atespace,
 		ActorName:             actorName,
 		ActorUid:              actorUID,
@@ -1151,10 +1155,14 @@ func TestTerminateWithoutWorkerPodUID(t *testing.T) {
 		ActorTemplateName:     "counter",
 		WorkerPodUid:          "",
 		Spec:                  spec,
-	}); err != nil {
+	})
+	if err != nil {
 		t.Fatalf("Terminate: %v", err)
 	}
 
+	if got := resp.GetState().GetLatestDurable().GetSnapshot(); !proto.Equal(got, uploaded) {
+		t.Errorf("TerminateResponse.state.latest_durable = %v, want the uploaded snapshot %v", got, uploaded)
+	}
 	if got := worker.actorDirs["TerminateWorkload"]; got != nil {
 		t.Errorf("TerminateWorkload was called on worker (%v), want skipped when WorkerPodUid is empty", got)
 	}
diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go
index d935c6df09..d6464032fa 100644
--- a/cmd/atelet/main.go
+++ b/cmd/atelet/main.go
@@ -2005,7 +2005,8 @@ func (s *Atelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest)
 	// up the copies on any other NodeVmsWithLocalSnapshots. This is fine *as of
 	// the day this was written* because today NodeVmsWithLocalSnapshots has at
 	// most one item.
-	if err := s.snapshots.Forget(actorUID); err != nil {
+	state, err := s.snapshots.Forget(actorUID)
+	if err != nil {
 		return nil, fmt.Errorf("failed to delete snapshot state during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err)
 	}
 	if err := snapshottracker.PruneLocalCheckpoints(ctx, actorUID); err != nil {
@@ -2018,7 +2019,7 @@ func (s *Atelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest)
 	}
 	s.restoreFailures.reset(actorUID)
 
-	return &ateletpb.TerminateResponse{}, nil
+	return &ateletpb.TerminateResponse{State: state}, nil
 }
 
 // terminateWorkloadOnWorker tears the workload down on the worker hosting it.
diff --git a/internal/proto/ateletpb/atelet.pb.go b/internal/proto/ateletpb/atelet.pb.go
index 754fb1b093..d9c6387d43 100644
--- a/internal/proto/ateletpb/atelet.pb.go
+++ b/internal/proto/ateletpb/atelet.pb.go
@@ -1918,7 +1918,12 @@ func (x *TerminateRequest) GetSpec() *WorkloadSpec {
 }
 
 type TerminateResponse struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// The actor's state as atelet forgets it: latest_local and uploading are
+	// unset, and latest_durable is its latest snapshot atelet knew was durable,
+	// if any. Terminate cancels an upload in progress, so that upload's snapshot
+	// is latest_durable only if the upload finished first.
+	State         *ActorSnapshotState `protobuf:"bytes,1,opt,name=state,proto3" json:"state,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1953,6 +1958,13 @@ func (*TerminateResponse) Descriptor() ([]byte, []int) {
 	return file_atelet_proto_rawDescGZIP(), []int{28}
 }
 
+func (x *TerminateResponse) GetState() *ActorSnapshotState {
+	if x != nil {
+		return x.State
+	}
+	return nil
+}
+
 type RunRequest struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// The UID of the worker pod whose the worker hosts the actor.
@@ -4437,8 +4449,9 @@ const file_atelet_proto_rawDesc = "" +
 	"\tactor_uid\x18\x04 \x01(\tR\bactorUid\x126\n" +
 	"\x17actor_template_atespace\x18\x05 \x01(\tR\x15actorTemplateAtespace\x12.\n" +
 	"\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12(\n" +
-	"\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\"\x13\n" +
-	"\x11TerminateResponse\"\xf4\x03\n" +
+	"\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\"E\n" +
+	"\x11TerminateResponse\x120\n" +
+	"\x05state\x18\x01 \x01(\v2\x1a.atelet.ActorSnapshotStateR\x05state\"\xf4\x03\n" +
 	"\n" +
 	"RunRequest\x12$\n" +
 	"\x0eworker_pod_uid\x18\x01 \x01(\tR\fworkerPodUid\x12\x1a\n" +
@@ -4733,84 +4746,85 @@ var file_atelet_proto_depIdxs = []int32{
 	27, // 13: atelet.EgressDestinations.ports:type_name -> atelet.EgressPorts
 	28, // 14: atelet.EgressPorts.all:type_name -> atelet.EgressAllPorts
 	35, // 15: atelet.TerminateRequest.spec:type_name -> atelet.WorkloadSpec
-	35, // 16: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec
-	34, // 17: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway
-	15, // 18: atelet.RunRequest.resources:type_name -> atelet.ResourceRequirements
-	11, // 19: atelet.RunRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime
-	46, // 20: atelet.WorkloadSpec.containers:type_name -> atelet.Container
-	44, // 21: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume
-	66, // 22: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry
-	67, // 23: atelet.ExternalVolumeSource.publish_context:type_name -> atelet.ExternalVolumeSource.PublishContextEntry
-	0,  // 24: atelet.ActorMetadataItem.field:type_name -> atelet.ActorMetadataField
-	39, // 25: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem
-	40, // 26: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource
-	41, // 27: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource
-	42, // 28: atelet.SystemInfoVolumeSource.data_sources:type_name -> atelet.SystemInfoDataSource
-	36, // 29: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolumeSource
-	37, // 30: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource
-	43, // 31: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolumeSource
-	38, // 32: atelet.Volume.image:type_name -> atelet.ImageVolumeSource
-	49, // 33: atelet.Container.env:type_name -> atelet.EnvVar
-	50, // 34: atelet.Container.wakeup_probe:type_name -> atelet.WakeupProbe
-	45, // 35: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount
-	47, // 36: atelet.Container.security_context:type_name -> atelet.SecurityContext
-	15, // 37: atelet.Container.resources:type_name -> atelet.ResourceRequirements
-	48, // 38: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities
-	51, // 39: atelet.WakeupProbe.http_get:type_name -> atelet.HttpGetAction
-	2,  // 40: atelet.ObjectStorage.sharing:type_name -> atelet.SnapshotSharing
-	1,  // 41: atelet.Snapshot.fidelity:type_name -> atelet.SnapshotFidelity
-	53, // 42: atelet.Snapshot.object:type_name -> atelet.ObjectStorage
-	54, // 43: atelet.SnapshotReport.snapshot:type_name -> atelet.Snapshot
-	68, // 44: atelet.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp
-	68, // 45: atelet.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp
-	55, // 46: atelet.ActorSnapshotState.latest_local:type_name -> atelet.SnapshotReport
-	55, // 47: atelet.ActorSnapshotState.uploading:type_name -> atelet.SnapshotReport
-	55, // 48: atelet.ActorSnapshotState.latest_durable:type_name -> atelet.SnapshotReport
-	68, // 49: atelet.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp
-	3,  // 50: atelet.UploadConfig.mode:type_name -> atelet.UploadMode
-	35, // 51: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec
-	54, // 52: atelet.CheckpointRequest.snapshot:type_name -> atelet.Snapshot
-	57, // 53: atelet.CheckpointRequest.upload_config:type_name -> atelet.UploadConfig
-	56, // 54: atelet.CheckpointResponse.state:type_name -> atelet.ActorSnapshotState
-	35, // 55: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec
-	54, // 56: atelet.RestoreRequest.snapshot:type_name -> atelet.Snapshot
-	34, // 57: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway
-	15, // 58: atelet.RestoreRequest.resources:type_name -> atelet.ResourceRequirements
-	11, // 59: atelet.RestoreRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime
-	54, // 60: atelet.PromoteSnapshotRequest.snapshot:type_name -> atelet.Snapshot
-	56, // 61: atelet.PromoteSnapshotResponse.state:type_name -> atelet.ActorSnapshotState
-	56, // 62: atelet.DetachActorResponse.state:type_name -> atelet.ActorSnapshotState
-	29, // 63: atelet.WorkerSupport.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest
-	4,  // 64: atelet.WorkerSupport.AcquireWorkerEpoch:input_type -> atelet.AcquireWorkerEpochRequest
-	6,  // 65: atelet.WorkerSupport.RegisterWorker:input_type -> atelet.RegisterWorkerRequest
-	7,  // 66: atelet.WorkerSupport.SetWorkerSandboxRuntimes:input_type -> atelet.SetWorkerSandboxRuntimesRequest
-	18, // 67: atelet.WorkerSupport.RequestActorSuspend:input_type -> atelet.RequestActorSuspendRequest
-	22, // 68: atelet.WorkerSupport.GetActorEgressPolicy:input_type -> atelet.GetActorEgressPolicyRequest
-	20, // 69: atelet.WorkerSupport.ReportActorCrashed:input_type -> atelet.ReportActorCrashedRequest
-	33, // 70: atelet.Atelet.Run:input_type -> atelet.RunRequest
-	58, // 71: atelet.Atelet.Checkpoint:input_type -> atelet.CheckpointRequest
-	60, // 72: atelet.Atelet.Restore:input_type -> atelet.RestoreRequest
-	62, // 73: atelet.Atelet.PromoteSnapshot:input_type -> atelet.PromoteSnapshotRequest
-	64, // 74: atelet.Atelet.DetachActor:input_type -> atelet.DetachActorRequest
-	31, // 75: atelet.Atelet.Terminate:input_type -> atelet.TerminateRequest
-	30, // 76: atelet.WorkerSupport.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse
-	5,  // 77: atelet.WorkerSupport.AcquireWorkerEpoch:output_type -> atelet.AcquireWorkerEpochResponse
-	17, // 78: atelet.WorkerSupport.RegisterWorker:output_type -> atelet.RegisterWorkerResponse
-	8,  // 79: atelet.WorkerSupport.SetWorkerSandboxRuntimes:output_type -> atelet.SetWorkerSandboxRuntimesResponse
-	19, // 80: atelet.WorkerSupport.RequestActorSuspend:output_type -> atelet.RequestActorSuspendResponse
-	23, // 81: atelet.WorkerSupport.GetActorEgressPolicy:output_type -> atelet.GetActorEgressPolicyResponse
-	21, // 82: atelet.WorkerSupport.ReportActorCrashed:output_type -> atelet.ReportActorCrashedResponse
-	52, // 83: atelet.Atelet.Run:output_type -> atelet.RunResponse
-	59, // 84: atelet.Atelet.Checkpoint:output_type -> atelet.CheckpointResponse
-	61, // 85: atelet.Atelet.Restore:output_type -> atelet.RestoreResponse
-	63, // 86: atelet.Atelet.PromoteSnapshot:output_type -> atelet.PromoteSnapshotResponse
-	65, // 87: atelet.Atelet.DetachActor:output_type -> atelet.DetachActorResponse
-	32, // 88: atelet.Atelet.Terminate:output_type -> atelet.TerminateResponse
-	76, // [76:89] is the sub-list for method output_type
-	63, // [63:76] is the sub-list for method input_type
-	63, // [63:63] is the sub-list for extension type_name
-	63, // [63:63] is the sub-list for extension extendee
-	0,  // [0:63] is the sub-list for field type_name
+	56, // 16: atelet.TerminateResponse.state:type_name -> atelet.ActorSnapshotState
+	35, // 17: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec
+	34, // 18: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway
+	15, // 19: atelet.RunRequest.resources:type_name -> atelet.ResourceRequirements
+	11, // 20: atelet.RunRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime
+	46, // 21: atelet.WorkloadSpec.containers:type_name -> atelet.Container
+	44, // 22: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume
+	66, // 23: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry
+	67, // 24: atelet.ExternalVolumeSource.publish_context:type_name -> atelet.ExternalVolumeSource.PublishContextEntry
+	0,  // 25: atelet.ActorMetadataItem.field:type_name -> atelet.ActorMetadataField
+	39, // 26: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem
+	40, // 27: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource
+	41, // 28: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource
+	42, // 29: atelet.SystemInfoVolumeSource.data_sources:type_name -> atelet.SystemInfoDataSource
+	36, // 30: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolumeSource
+	37, // 31: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource
+	43, // 32: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolumeSource
+	38, // 33: atelet.Volume.image:type_name -> atelet.ImageVolumeSource
+	49, // 34: atelet.Container.env:type_name -> atelet.EnvVar
+	50, // 35: atelet.Container.wakeup_probe:type_name -> atelet.WakeupProbe
+	45, // 36: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount
+	47, // 37: atelet.Container.security_context:type_name -> atelet.SecurityContext
+	15, // 38: atelet.Container.resources:type_name -> atelet.ResourceRequirements
+	48, // 39: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities
+	51, // 40: atelet.WakeupProbe.http_get:type_name -> atelet.HttpGetAction
+	2,  // 41: atelet.ObjectStorage.sharing:type_name -> atelet.SnapshotSharing
+	1,  // 42: atelet.Snapshot.fidelity:type_name -> atelet.SnapshotFidelity
+	53, // 43: atelet.Snapshot.object:type_name -> atelet.ObjectStorage
+	54, // 44: atelet.SnapshotReport.snapshot:type_name -> atelet.Snapshot
+	68, // 45: atelet.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp
+	68, // 46: atelet.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp
+	55, // 47: atelet.ActorSnapshotState.latest_local:type_name -> atelet.SnapshotReport
+	55, // 48: atelet.ActorSnapshotState.uploading:type_name -> atelet.SnapshotReport
+	55, // 49: atelet.ActorSnapshotState.latest_durable:type_name -> atelet.SnapshotReport
+	68, // 50: atelet.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp
+	3,  // 51: atelet.UploadConfig.mode:type_name -> atelet.UploadMode
+	35, // 52: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec
+	54, // 53: atelet.CheckpointRequest.snapshot:type_name -> atelet.Snapshot
+	57, // 54: atelet.CheckpointRequest.upload_config:type_name -> atelet.UploadConfig
+	56, // 55: atelet.CheckpointResponse.state:type_name -> atelet.ActorSnapshotState
+	35, // 56: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec
+	54, // 57: atelet.RestoreRequest.snapshot:type_name -> atelet.Snapshot
+	34, // 58: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway
+	15, // 59: atelet.RestoreRequest.resources:type_name -> atelet.ResourceRequirements
+	11, // 60: atelet.RestoreRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime
+	54, // 61: atelet.PromoteSnapshotRequest.snapshot:type_name -> atelet.Snapshot
+	56, // 62: atelet.PromoteSnapshotResponse.state:type_name -> atelet.ActorSnapshotState
+	56, // 63: atelet.DetachActorResponse.state:type_name -> atelet.ActorSnapshotState
+	29, // 64: atelet.WorkerSupport.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest
+	4,  // 65: atelet.WorkerSupport.AcquireWorkerEpoch:input_type -> atelet.AcquireWorkerEpochRequest
+	6,  // 66: atelet.WorkerSupport.RegisterWorker:input_type -> atelet.RegisterWorkerRequest
+	7,  // 67: atelet.WorkerSupport.SetWorkerSandboxRuntimes:input_type -> atelet.SetWorkerSandboxRuntimesRequest
+	18, // 68: atelet.WorkerSupport.RequestActorSuspend:input_type -> atelet.RequestActorSuspendRequest
+	22, // 69: atelet.WorkerSupport.GetActorEgressPolicy:input_type -> atelet.GetActorEgressPolicyRequest
+	20, // 70: atelet.WorkerSupport.ReportActorCrashed:input_type -> atelet.ReportActorCrashedRequest
+	33, // 71: atelet.Atelet.Run:input_type -> atelet.RunRequest
+	58, // 72: atelet.Atelet.Checkpoint:input_type -> atelet.CheckpointRequest
+	60, // 73: atelet.Atelet.Restore:input_type -> atelet.RestoreRequest
+	62, // 74: atelet.Atelet.PromoteSnapshot:input_type -> atelet.PromoteSnapshotRequest
+	64, // 75: atelet.Atelet.DetachActor:input_type -> atelet.DetachActorRequest
+	31, // 76: atelet.Atelet.Terminate:input_type -> atelet.TerminateRequest
+	30, // 77: atelet.WorkerSupport.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse
+	5,  // 78: atelet.WorkerSupport.AcquireWorkerEpoch:output_type -> atelet.AcquireWorkerEpochResponse
+	17, // 79: atelet.WorkerSupport.RegisterWorker:output_type -> atelet.RegisterWorkerResponse
+	8,  // 80: atelet.WorkerSupport.SetWorkerSandboxRuntimes:output_type -> atelet.SetWorkerSandboxRuntimesResponse
+	19, // 81: atelet.WorkerSupport.RequestActorSuspend:output_type -> atelet.RequestActorSuspendResponse
+	23, // 82: atelet.WorkerSupport.GetActorEgressPolicy:output_type -> atelet.GetActorEgressPolicyResponse
+	21, // 83: atelet.WorkerSupport.ReportActorCrashed:output_type -> atelet.ReportActorCrashedResponse
+	52, // 84: atelet.Atelet.Run:output_type -> atelet.RunResponse
+	59, // 85: atelet.Atelet.Checkpoint:output_type -> atelet.CheckpointResponse
+	61, // 86: atelet.Atelet.Restore:output_type -> atelet.RestoreResponse
+	63, // 87: atelet.Atelet.PromoteSnapshot:output_type -> atelet.PromoteSnapshotResponse
+	65, // 88: atelet.Atelet.DetachActor:output_type -> atelet.DetachActorResponse
+	32, // 89: atelet.Atelet.Terminate:output_type -> atelet.TerminateResponse
+	77, // [77:90] is the sub-list for method output_type
+	64, // [64:77] is the sub-list for method input_type
+	64, // [64:64] is the sub-list for extension type_name
+	64, // [64:64] is the sub-list for extension extendee
+	0,  // [0:64] is the sub-list for field type_name
 }
 
 func init() { file_atelet_proto_init() }
diff --git a/internal/proto/ateletpb/atelet.proto b/internal/proto/ateletpb/atelet.proto
index f68551c7b6..5ed0cee19d 100644
--- a/internal/proto/ateletpb/atelet.proto
+++ b/internal/proto/ateletpb/atelet.proto
@@ -480,7 +480,9 @@ service Atelet {
   rpc DetachActor(DetachActorRequest) returns (DetachActorResponse) {}
 
   // Terminate tells atelet to terminate/kill any running workload for an actor,
-  // unmount its volumes, and clean up actor state on the node.
+  // unmount its volumes, and clean up actor state on the node. It returns the
+  // actor's state as atelet forgets it, so the caller can record an upload
+  // atelet has not reported yet.
   rpc Terminate(TerminateRequest) returns (TerminateResponse) {}
 }
 
@@ -522,6 +524,11 @@ message TerminateRequest {
 }
 
 message TerminateResponse {
+  // The actor's state as atelet forgets it: latest_local and uploading are
+  // unset, and latest_durable is its latest snapshot atelet knew was durable,
+  // if any. Terminate cancels an upload in progress, so that upload's snapshot
+  // is latest_durable only if the upload finished first.
+  ActorSnapshotState state = 1;
 }
 
 message RunRequest {
diff --git a/internal/proto/ateletpb/atelet_grpc.pb.go b/internal/proto/ateletpb/atelet_grpc.pb.go
index a3c2a54c82..328f9a8373 100644
--- a/internal/proto/ateletpb/atelet_grpc.pb.go
+++ b/internal/proto/ateletpb/atelet_grpc.pb.go
@@ -500,7 +500,9 @@ type AteletClient interface {
 	// snapshot durable, such as when it has no `snapshot.object`.
 	DetachActor(ctx context.Context, in *DetachActorRequest, opts ...grpc.CallOption) (*DetachActorResponse, error)
 	// Terminate tells atelet to terminate/kill any running workload for an actor,
-	// unmount its volumes, and clean up actor state on the node.
+	// unmount its volumes, and clean up actor state on the node. It returns the
+	// actor's state as atelet forgets it, so the caller can record an upload
+	// atelet has not reported yet.
 	Terminate(ctx context.Context, in *TerminateRequest, opts ...grpc.CallOption) (*TerminateResponse, error)
 }
 
@@ -631,7 +633,9 @@ type AteletServer interface {
 	// snapshot durable, such as when it has no `snapshot.object`.
 	DetachActor(context.Context, *DetachActorRequest) (*DetachActorResponse, error)
 	// Terminate tells atelet to terminate/kill any running workload for an actor,
-	// unmount its volumes, and clean up actor state on the node.
+	// unmount its volumes, and clean up actor state on the node. It returns the
+	// actor's state as atelet forgets it, so the caller can record an upload
+	// atelet has not reported yet.
 	Terminate(context.Context, *TerminateRequest) (*TerminateResponse, error)
 	mustEmbedUnimplementedAteletServer()
 }

From 3883bb8d5738ef3c76e19fb67dc5a9a875b1137b Mon Sep 17 00:00:00 2001
From: zoezhao <zoezhao@google.com>
Date: Wed, 21 Oct 2026 11:58:28 -0700
Subject: [PATCH 2/7] ateapi: drop the node binding of an unreachable atelet at
 revert finalize

When atelet is unreachable but the latest local snapshot is already
uploaded, revert recorded a detach before finalizing. FinalizeReverted
already clears the node binding and every snapshot's locality, so the
extra write only added a version conflict. Revert now goes straight on,
and recordDetach loses its revert-only branch.
---
 .../controlapi/workflow_resume_detach.go      |  3 --
 .../controlapi/workflow_resume_test.go        |  7 -----
 .../internal/controlapi/workflow_revert.go    | 17 ++++-------
 .../controlapi/workflow_revert_test.go        | 29 +------------------
 4 files changed, 7 insertions(+), 49 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_detach.go b/cmd/ateapi/internal/controlapi/workflow_resume_detach.go
index 665ed311d5..4800b5fc55 100644
--- a/cmd/ateapi/internal/controlapi/workflow_resume_detach.go
+++ b/cmd/ateapi/internal/controlapi/workflow_resume_detach.go
@@ -139,9 +139,6 @@ func (w *ActorWorkflow) recordDetach(ctx context.Context, actorRef resources.Act
 	if err := applyDetach(applied, node, applied.GetLastAssignedGeneration()); err != nil {
 		return nil, err
 	}
-	if applied.GetState() == ateapipb.ActorState_ACTOR_STATE_REVERTING && applied.GetWorkerAssignment() == nil {
-		applied.AssignedNode = ""
-	}
 	stored, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error {
 		// The precondition makes toUpdate the version applied was built from.
 		toUpdate.Status.Snapshots = applied.GetSnapshots()
diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go
index 531345864c..0ed1d1957c 100644
--- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go
@@ -1150,7 +1150,6 @@ type capturingAtelet struct {
 	// onDetach, when set, runs before DetachActor returns.
 	onDetach     func()
 	terminateErr error
-	onTerminate  func()
 	// terminateState, when set, is the snapshot state Terminate returns.
 	// Terminate is unimplemented otherwise.
 	terminateState *ateletpb.ActorSnapshotState
@@ -1214,12 +1213,6 @@ func (f *capturingAtelet) DetachActor(ctx context.Context, req *ateletpb.DetachA
 }
 
 func (f *capturingAtelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) {
-	f.mu.Lock()
-	hook := f.onTerminate
-	f.mu.Unlock()
-	if hook != nil {
-		hook()
-	}
 	f.mu.Lock()
 	err, state := f.terminateErr, proto.CloneOf(f.terminateState)
 	f.mu.Unlock()
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go
index 8c55251226..8b411bab45 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert.go
@@ -193,13 +193,13 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 	defer func() { err = done(err) }()
 
 	if node := actor.GetStatus().GetAssignedNode(); node != "" {
-		forgetNode := false
 		state, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert)
-		if err != nil {
-			if !canDetachWithoutAtelet(ctx, actor, err) {
-				return nil, err
-			}
-			forgetNode = true
+		// The revert can go on without an unreachable atelet when nothing runs
+		// on the node and its latest local snapshot is already uploaded.
+		// FinalizeReverted drops the node binding, and the actor resumes from
+		// the uploaded copy.
+		if err != nil && !canDetachWithoutAtelet(ctx, actor, err) {
+			return nil, err
 		}
 		if actor, err = w.recordTerminatedUpload(ctx, actorRef, actor, node, state); err != nil {
 			return nil, err
@@ -207,11 +207,6 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 		if err := w.ensureVolumesDetached(ctx, actor, actorTemplate, "DetachVolumesForRevert", ateattr.OperationRevert); err != nil {
 			return nil, err
 		}
-		if forgetNode {
-			if actor, err = w.recordDetach(ctx, actorRef, actor, node, nil); err != nil {
-				return nil, err
-			}
-		}
 		if actor.GetStatus().GetWorkerAssignment() != nil {
 			if _, err := releaseWorker(ctx, w.store, actor); err != nil {
 				return nil, fmt.Errorf("while releasing worker: %w", err)
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
index de2644ffc9..81d1c74a41 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
@@ -19,7 +19,6 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
-	"go.ate.dev/substrate/cmd/ateapi/internal/store"
 	"go.ate.dev/substrate/cmd/ateapi/internal/store/storetest"
 	"go.ate.dev/substrate/internal/apierror"
 	"go.ate.dev/substrate/internal/objectstore/objectstoretest"
@@ -232,7 +231,6 @@ func TestRevertActor_UnreachableNode(t *testing.T) {
 		pending  bool
 		assigned bool
 		crashed  bool
-		conflict bool
 		wantCode codes.Code
 	}{
 		{name: "missing atelet", node: "node-3"},
@@ -242,7 +240,6 @@ func TestRevertActor_UnreachableNode(t *testing.T) {
 		{name: "worker still assigned", node: "node-1", err: apierror.Unavailable("offline"), assigned: true, wantCode: codes.Unavailable},
 		{name: "crashed execution still needs termination", node: "node-1", err: apierror.Unavailable("offline"), crashed: true, wantCode: codes.Unavailable},
 		{name: "other RPC failure", node: "node-1", err: apierror.Internal("injected failure"), wantCode: codes.Internal},
-		{name: "concurrent update", node: "node-1", err: apierror.Unavailable("offline"), conflict: true, wantCode: codes.Aborted},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			ctx := context.Background()
@@ -280,21 +277,6 @@ func TestRevertActor_UnreachableNode(t *testing.T) {
 			})
 			want := proto.CloneOf(actor.GetStatus().GetSnapshots()[1])
 			atelet.terminateErr = tt.err
-			if tt.conflict {
-				atelet.onTerminate = func() {
-					cur, err := st.GetActor(ctx, ref)
-					if err != nil {
-						t.Errorf("GetActor: %v", err)
-						return
-					}
-					if _, err := st.UpdateActor(ctx, ref, store.PreconditionFrom(cur), func(a *ateapipb.Actor) error {
-						a.WorkerSelector = &ateapipb.Selector{MatchLabels: map[string]string{"changed": "true"}}
-						return nil
-					}); err != nil {
-						t.Errorf("UpdateActor: %v", err)
-					}
-				}
-			}
 			reverted, err := w.RevertActor(ctx, ref)
 			if got := apierror.Code(err); got != tt.wantCode {
 				t.Fatalf("RevertActor = %v, want %v", err, tt.wantCode)
@@ -311,16 +293,7 @@ func TestRevertActor_UnreachableNode(t *testing.T) {
 				if !proto.Equal(got.GetSnapshots()[1], want) {
 					t.Fatalf("snapshot changed after failed revert: %v", got.GetSnapshots())
 				}
-				if !tt.conflict {
-					return
-				}
-				atelet.mu.Lock()
-				atelet.onTerminate = nil
-				atelet.mu.Unlock()
-				reverted, err = w.RevertActor(ctx, ref)
-				if err != nil {
-					t.Fatalf("retry after conflict: %v", err)
-				}
+				return
 			}
 			got := reverted.GetStatus()
 			if got.GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDED || got.GetAssignedNode() != "" || got.GetWorkerAssignment() != nil {

From b1d8b4635004bf778226a8ab63b7598448ce45bb Mon Sep 17 00:00:00 2001
From: zoezhao <zoezhao@google.com>
Date: Wed, 21 Oct 2026 15:32:47 -0700
Subject: [PATCH 3/7] ateapi: discard a crashed suspend's snapshot on revert

A suspend that crashes after atelet commits its checkpoint leaves a
PENDING snapshot on no node. Terminate can report it durable, and revert
then returned the actor to it and released the previous snapshot.

recordTerminatedUpload now skips a PENDING snapshot on no node, so
revert deletes it as in progress.
---
 cmd/ateapi/internal/controlapi/workflow_revert.go   |  8 ++++++++
 .../internal/controlapi/workflow_revert_test.go     | 13 +++++++++++--
 2 files changed, 19 insertions(+), 2 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go
index 8b411bab45..dc14d8dfce 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert.go
@@ -224,6 +224,14 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 // upload as in progress. It writes at once because atelet forgot the actor, so
 // a retried Terminate returns nothing.
 func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, node string, state *ateletpb.ActorSnapshotState) (*ateapipb.Actor, error) {
+	reportedGen := state.GetLatestDurable().GetSnapshot().GetGeneration()
+	snap := findSnapshotByGeneration(actor.GetStatus().GetSnapshots(), reportedGen)
+	if snap != nil && snap.GetLocality() == "" && snap.GetDurableSnapshot().GetStatus() == ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING {
+		// A snapshot on no node is the one a crashed suspend minted and never
+		// finalized. It stays PENDING, so the revert deletes it with the rest of the
+		// discarded execution.
+		return actor, nil
+	}
 	applied := proto.CloneOf(actor.GetStatus())
 	if err := applyAteletResponseState(ctx, applied, state); err != nil {
 		return nil, apierror.Internal("atelet on node %q reported a durable snapshot for Actor %s that conflicts with its record: %w", node, actorRef, err)
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
index 81d1c74a41..d69873b7a5 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
@@ -314,7 +314,8 @@ func TestRevertActor_UnreachableNode(t *testing.T) {
 // actor to a local snapshot atelet uploaded, its latest durable snapshot, and
 // collects the durable snapshot the upload replaced. The revert learns of the
 // upload from the record, or from Terminate when atelet's report has not
-// landed. An upload atelet does not report is deleted as in progress.
+// landed. An upload atelet does not report, or that a crashed suspend never
+// finalized, is deleted as in progress.
 func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 	memory := ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY
 	completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
@@ -328,12 +329,16 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 		node string
 		// terminateReports makes Terminate return the upload as durable.
 		terminateReports bool
-		wantUploadKept   bool
+		// unfinalized seeds the upload as the snapshot of a suspend that
+		// crashed before it committed, which is on no node.
+		unfinalized    bool
+		wantUploadKept bool
 	}{
 		{name: "upload recorded", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: completed, wantUploadKept: true},
 		{name: "paused, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1", terminateReports: true, wantUploadKept: true},
 		{name: "crashed, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, wantUploadKept: true},
 		{name: "upload not reported", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1"},
+		{name: "crashed suspend, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, unfinalized: true},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx := context.Background()
@@ -369,6 +374,10 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 				if tc.state == ateapipb.ActorState_ACTOR_STATE_CRASHED {
 					s.Crash = newActorCrash("ResumeActor", "interrupted execution")
 				}
+				if tc.unfinalized {
+					paused.Locality = ""
+					s.Crash = newActorCrash("SuspendActor", "interrupted execution")
+				}
 			})
 			atelet.terminateErr = apierror.NotFound("atelet holds nothing for the actor")
 			if tc.terminateReports {

From 456f1cb7335937a7c28d71ec629dab52f7c8c128 Mon Sep 17 00:00:00 2001
From: zoezhao <zoezhao@google.com>
Date: Thu, 22 Oct 2026 14:06:54 -0700
Subject: [PATCH 4/7] atelet: keep the latest durable snapshot until Terminate
 finishes

Forget deleted the snapshot state file before Terminate pruned the
actor's checkpoints and directories. If either failed, a retried
Terminate returned no latest durable snapshot, and revert deleted the
upload as in progress.

Forget now leaves only the latest durable snapshot in the state file,
and RemoveActorDirs deletes it after the actor's other files. With
nothing local and no report due, an atelet restarted mid-Terminate
starts no background work on the actor.
---
 cmd/atelet/internal/actordirs/actordirs.go    |  7 ++---
 .../internal/snapshottracker/lifecycle.go     | 27 ++++++++++++++-----
 .../internal/snapshottracker/report_test.go   |  4 +--
 .../internal/snapshottracker/tracker_test.go  | 16 +++++++----
 cmd/atelet/lifecycle_test.go                  | 18 ++++++++++---
 cmd/atelet/main.go                            |  2 +-
 6 files changed, 52 insertions(+), 22 deletions(-)

diff --git a/cmd/atelet/internal/actordirs/actordirs.go b/cmd/atelet/internal/actordirs/actordirs.go
index d005817f1a..84190d2222 100644
--- a/cmd/atelet/internal/actordirs/actordirs.go
+++ b/cmd/atelet/internal/actordirs/actordirs.go
@@ -146,9 +146,10 @@ func RemoveActorFilesExceptState(actorUID string) error {
 // It refuses to proceed while a volume directory is still populated, so a
 // failed unmount cannot become a deletion of the mount's contents, and removes
 // a bundle upper dir carrying an image's read-only modes without recreating
-// directories. Terminate calls Forget first, which has already deleted
-// snapshot-state.json; after a crash here, Start finds no state file, so only a
-// retried Terminate finishes the cleanup.
+// directories. Terminate calls Forget first, which leaves only the latest
+// durable snapshot in snapshot-state.json. This deletes that file after the
+// actor's other files, so a Terminate retried after a failure here returns the
+// snapshot again.
 func RemoveActorDirs(actorUID string) error {
 	if err := RemoveActorFilesExceptState(actorUID); err != nil {
 		return err
diff --git a/cmd/atelet/internal/snapshottracker/lifecycle.go b/cmd/atelet/internal/snapshottracker/lifecycle.go
index 2412130628..95edd6ceab 100644
--- a/cmd/atelet/internal/snapshottracker/lifecycle.go
+++ b/cmd/atelet/internal/snapshottracker/lifecycle.go
@@ -16,6 +16,7 @@ package snapshottracker
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"go.ate.dev/substrate/internal/apierror"
@@ -346,16 +347,18 @@ func (t *Tracker) holdLocked(actorUID string, a *actorSnapshots, kind holdKind)
 	}
 }
 
-// Forget drops the actor's state, in memory and on disk, and returns the state
-// it dropped with only its latest durable snapshot, since the caller deletes
-// the actor's local snapshots. It first cancels any background upload or
-// detach of the actor in progress and waits for it to end, releasing the
-// actor's lock meanwhile.
+// Forget drops the actor's state and returns it with only its latest durable
+// snapshot, since the caller deletes the actor's local snapshots. It leaves
+// only that snapshot in the state file, for the caller to delete with the
+// actor's directory, so a retry after a failed cleanup returns it again. It
+// first cancels any background upload or detach of the actor in progress and
+// waits for it to end, releasing the actor's lock meanwhile.
 func (t *Tracker) Forget(actorUID string) (*ateletpb.ActorSnapshotState, error) {
 	t.lock(actorUID)
 	defer t.unlock(actorUID)
 	state := &ateletpb.ActorSnapshotState{}
-	if a := t.findLocked(actorUID); a != nil {
+	a := t.findLocked(actorUID)
+	if a != nil {
 		t.addLocked(a)
 		a.hold = holdTerminate
 		for a.op != nil {
@@ -370,8 +373,18 @@ func (t *Tracker) Forget(actorUID string) (*ateletpb.ActorSnapshotState, error)
 	t.retries.Forget(actorUID)
 	t.reporting.retries.Forget(actorUID)
 	t.dropLocked(actorUID)
+	if a == nil || a.LatestDurable == nil {
+		t.undurableActorsTally.forget(actorUID)
+		return state, removeSnapshotState(actorUID)
+	}
+	// With nothing local and no report due, an atelet restarted before the
+	// caller's cleanup finishes starts no background work on the actor.
+	err := t.writeStateLocked(&actorSnapshots{actorUID: actorUID, actorSnapshotsRecord: actorSnapshotsRecord{LatestDurable: a.LatestDurable}})
 	t.undurableActorsTally.forget(actorUID)
-	return state, removeSnapshotState(actorUID)
+	if err != nil {
+		return state, fmt.Errorf("while saving the actor's latest durable snapshot: %w", err)
+	}
+	return state, nil
 }
 
 // setLatestLocalLocked records snap as the actor's latest local snapshot and
diff --git a/cmd/atelet/internal/snapshottracker/report_test.go b/cmd/atelet/internal/snapshottracker/report_test.go
index 6d0117357e..2956fb9912 100644
--- a/cmd/atelet/internal/snapshottracker/report_test.go
+++ b/cmd/atelet/internal/snapshottracker/report_test.go
@@ -1223,8 +1223,8 @@ func TestTrackerForgetDuringReport(t *testing.T) {
 		t.Errorf("reported %v after Forget, want nothing", r.state)
 	case <-time.After(50 * time.Millisecond):
 	}
-	if _, err := os.Stat(ateletpath.ActorSnapshotStateFile(trackedActorUID)); !os.IsNotExist(err) {
-		t.Errorf("state file once the report ended: stat = %v, want none", err)
+	if got := newTracker(t).State(trackedActorUID); uuidOf(got.GetLatestDurable()) != "s1" || got.GetLatestLocal() != nil {
+		t.Errorf("state file once the report ended = %v, want only s1 durable", got)
 	}
 }
 
diff --git a/cmd/atelet/internal/snapshottracker/tracker_test.go b/cmd/atelet/internal/snapshottracker/tracker_test.go
index 88ebe31e39..ee8f58b2f1 100644
--- a/cmd/atelet/internal/snapshottracker/tracker_test.go
+++ b/cmd/atelet/internal/snapshottracker/tracker_test.go
@@ -478,14 +478,20 @@ func TestTrackerState(t *testing.T) {
 	if err != nil {
 		t.Fatalf("forget: %v", err)
 	}
-	if want := (&ateletpb.ActorSnapshotState{LatestDurable: st.GetLatestDurable()}); !proto.Equal(forgotten, want) {
+	want := &ateletpb.ActorSnapshotState{LatestDurable: st.GetLatestDurable()}
+	if !proto.Equal(forgotten, want) {
 		t.Errorf("Forget returned %v, want %v: s2 durable and nothing local", forgotten, want)
 	}
-	if _, err := os.Stat(ateletpath.ActorSnapshotStateFile(trackedActorUID)); !os.IsNotExist(err) {
-		t.Errorf("state file after forget: stat = %v, want it deleted", err)
+	// Terminate deletes the state file after the actor's other files, so a
+	// Terminate retried after a failed cleanup forgets s2 again.
+	if got := newTracker(t).State(trackedActorUID); !proto.Equal(got, want) {
+		t.Errorf("State() on a fresh tracker after forget = %v, want %v from the state file", got, want)
 	}
-	if got := tr.State(trackedActorUID); !proto.Equal(got, &ateletpb.ActorSnapshotState{}) {
-		t.Errorf("state after forget = %v, want empty", got)
+	if forgotten, err = tr.Forget(trackedActorUID); err != nil {
+		t.Fatalf("second forget: %v", err)
+	}
+	if !proto.Equal(forgotten, want) {
+		t.Errorf("second Forget returned %v, want %v", forgotten, want)
 	}
 }
 
diff --git a/cmd/atelet/lifecycle_test.go b/cmd/atelet/lifecycle_test.go
index fc0af091fd..3a06de9cdd 100644
--- a/cmd/atelet/lifecycle_test.go
+++ b/cmd/atelet/lifecycle_test.go
@@ -1298,7 +1298,8 @@ func TestTerminateReclaimsStateWithoutALiveSandbox(t *testing.T) {
 
 // TestTerminateCanBeRetried pins that a repeated Terminate succeeds, both when
 // a single caller retries and when DeleteActor or RevertActor sends a second
-// Terminate after DeleteWorker's best-effort Terminate.
+// Terminate after DeleteWorker's best-effort Terminate. A retry after a failed
+// Terminate still returns the actor's latest durable snapshot.
 func TestTerminateCanBeRetried(t *testing.T) {
 	for _, tt := range []struct {
 		name string
@@ -1324,7 +1325,8 @@ func TestTerminateCanBeRetried(t *testing.T) {
 			ctx := t.Context()
 
 			snap := a.snapshot("pause-snap-1", 1)
-			if _, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil {
+			checkpointResp, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE)
+			if err != nil {
 				t.Fatalf("Checkpoint: %v", err)
 			}
 
@@ -1349,7 +1351,7 @@ func TestTerminateCanBeRetried(t *testing.T) {
 				WorkerPodUid:          pluginWorkerPodUID,
 				Spec:                  a.spec,
 			}
-			_, err := a.s.Terminate(ctx, firstReq)
+			_, err = a.s.Terminate(ctx, firstReq)
 			if tt.failFirst {
 				if err == nil {
 					t.Fatal("first Terminate succeeded through a populated volume directory")
@@ -1366,9 +1368,17 @@ func TestTerminateCanBeRetried(t *testing.T) {
 			if tt.secondWithoutWorker {
 				secondReq.WorkerPodUid = ""
 			}
-			if _, err := a.s.Terminate(ctx, secondReq); err != nil {
+			resp, err := a.s.Terminate(ctx, secondReq)
+			if err != nil {
 				t.Fatalf("second Terminate: %v", err)
 			}
+			var wantDurable *ateletpb.SnapshotReport
+			if tt.failFirst {
+				wantDurable = checkpointResp.GetState().GetLatestDurable()
+			}
+			if got := resp.GetState().GetLatestDurable(); !proto.Equal(got, wantDurable) {
+				t.Errorf("second Terminate returned latest_durable %v, want %v", got, wantDurable)
+			}
 			if got := a.worker.terminations.Load(); got != tt.wantTeardowns {
 				t.Errorf("worker saw %d TerminateWorkload calls, want %d", got, tt.wantTeardowns)
 			}
diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go
index d6464032fa..0f0cb2f97d 100644
--- a/cmd/atelet/main.go
+++ b/cmd/atelet/main.go
@@ -2007,7 +2007,7 @@ func (s *Atelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest)
 	// most one item.
 	state, err := s.snapshots.Forget(actorUID)
 	if err != nil {
-		return nil, fmt.Errorf("failed to delete snapshot state during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err)
+		return nil, fmt.Errorf("failed to forget snapshot state during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err)
 	}
 	if err := snapshottracker.PruneLocalCheckpoints(ctx, actorUID); err != nil {
 		return nil, fmt.Errorf("failed to prune local checkpoints during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err)

From 1b9e00009b6f604c00915d36ef973132380a0c7b Mon Sep 17 00:00:00 2001
From: zoezhao <zoezhao@google.com>
Date: Thu, 22 Oct 2026 14:47:59 -0700
Subject: [PATCH 5/7] ateapi: record only a terminated upload on the actor's
 node

recordTerminatedUpload skipped only a PENDING snapshot on no node. It
now records a snapshot only when the record places it on the actor's
assigned node, which only a finalized suspend does. Any other reported
snapshot stays PENDING, and the revert deletes it.

It also returns early when Terminate reports no durable snapshot, and
reads the node from the actor instead of taking it as an argument.
---
 .../internal/controlapi/workflow_revert.go    | 40 ++++++++++++-------
 .../controlapi/workflow_revert_test.go        | 33 +++++++--------
 2 files changed, 42 insertions(+), 31 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go
index dc14d8dfce..e0603ff2e3 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert.go
@@ -192,7 +192,7 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 	ctx, done := stepSpan(ctx, "DiscardWorker")
 	defer func() { err = done(err) }()
 
-	if node := actor.GetStatus().GetAssignedNode(); node != "" {
+	if actor.GetStatus().GetAssignedNode() != "" {
 		state, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert)
 		// The revert can go on without an unreachable atelet when nothing runs
 		// on the node and its latest local snapshot is already uploaded.
@@ -201,7 +201,7 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 		if err != nil && !canDetachWithoutAtelet(ctx, actor, err) {
 			return nil, err
 		}
-		if actor, err = w.recordTerminatedUpload(ctx, actorRef, actor, node, state); err != nil {
+		if actor, err = w.recordTerminatedUpload(ctx, actorRef, actor, state); err != nil {
 			return nil, err
 		}
 		if err := w.ensureVolumesDetached(ctx, actor, actorTemplate, "DetachVolumesForRevert", ateattr.OperationRevert); err != nil {
@@ -217,19 +217,29 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 	return actor, w.releaseAssignmentWithoutBacklink(ctx, actor)
 }
 
-// recordTerminatedUpload marks COMPLETED the snapshot that the atelet on node
-// reported durable in state as it terminated the actor, with the same rules
-// as a report. atelet may upload a snapshot without its report landing, and a
-// CRASHED actor refuses reports, so without this the revert would delete the
-// upload as in progress. It writes at once because atelet forgot the actor, so
-// a retried Terminate returns nothing.
-func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, node string, state *ateletpb.ActorSnapshotState) (*ateapipb.Actor, error) {
-	reportedGen := state.GetLatestDurable().GetSnapshot().GetGeneration()
-	snap := findSnapshotByGeneration(actor.GetStatus().GetSnapshots(), reportedGen)
-	if snap != nil && snap.GetLocality() == "" && snap.GetDurableSnapshot().GetStatus() == ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING {
-		// A snapshot on no node is the one a crashed suspend minted and never
-		// finalized. It stays PENDING, so the revert deletes it with the rest of the
-		// discarded execution.
+// recordTerminatedUpload marks COMPLETED the snapshot that the atelet on the
+// actor's assigned node reported durable in state as it terminated the actor,
+// with the same rules as a report. atelet may upload a snapshot without its
+// report landing, and a CRASHED actor refuses reports, so without this the
+// revert would delete the upload as in progress. It writes at once because
+// atelet forgot the actor, so a retried Terminate returns nothing.
+func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, state *ateletpb.ActorSnapshotState) (*ateapipb.Actor, error) {
+	reported := state.GetLatestDurable().GetSnapshot()
+	if reported == nil {
+		return actor, nil
+	}
+	node := actor.GetStatus().GetAssignedNode()
+	// Only a finalized suspend places a snapshot on a node, so any other
+	// snapshot, such as one a crashed suspend created, stays PENDING and the
+	// revert deletes it with the rest of the discarded execution.
+	snap := findSnapshotByGeneration(actor.GetStatus().GetSnapshots(), reported.GetGeneration())
+	if got := snap.GetLocality(); got != node {
+		slog.InfoContext(ctx, "Discarding an upload Terminate reported for a snapshot not on its node",
+			slog.Any("actor", actorRef),
+			slog.String("node", node),
+			slog.String("snapshot_locality", got),
+			slog.Int("generation", int(reported.GetGeneration())),
+			slog.String("uuid", reported.GetSnapshotUuid()))
 		return actor, nil
 	}
 	applied := proto.CloneOf(actor.GetStatus())
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
index d69873b7a5..85a207242b 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
@@ -314,8 +314,8 @@ func TestRevertActor_UnreachableNode(t *testing.T) {
 // actor to a local snapshot atelet uploaded, its latest durable snapshot, and
 // collects the durable snapshot the upload replaced. The revert learns of the
 // upload from the record, or from Terminate when atelet's report has not
-// landed. An upload atelet does not report, or that a crashed suspend never
-// finalized, is deleted as in progress.
+// landed. An upload atelet does not report, or of a snapshot not on the node
+// Terminate ran on, is deleted as in progress.
 func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 	memory := ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY
 	completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
@@ -329,16 +329,17 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 		node string
 		// terminateReports makes Terminate return the upload as durable.
 		terminateReports bool
-		// unfinalized seeds the upload as the snapshot of a suspend that
-		// crashed before it committed, which is on no node.
-		unfinalized    bool
+		// locality is the upload's node. A suspend that crashed before it
+		// committed leaves its snapshot on no node.
+		locality       string
 		wantUploadKept bool
 	}{
-		{name: "upload recorded", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: completed, wantUploadKept: true},
-		{name: "paused, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1", terminateReports: true, wantUploadKept: true},
-		{name: "crashed, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, wantUploadKept: true},
-		{name: "upload not reported", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1"},
-		{name: "crashed suspend, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, unfinalized: true},
+		{name: "upload recorded", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: completed, locality: "node-1", wantUploadKept: true},
+		{name: "paused, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1", terminateReports: true, locality: "node-1", wantUploadKept: true},
+		{name: "crashed, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, locality: "node-1", wantUploadKept: true},
+		{name: "upload not reported", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1", locality: "node-1"},
+		{name: "crashed suspend, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true},
+		{name: "upload on another node reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, locality: "node-2"},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx := context.Background()
@@ -363,7 +364,7 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 			objects.PutSnapshot(t, old, "manifest.json")
 			objects.PutSnapshot(t, uploaded, "manifest.json")
 			mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapipb.ActorStatus) {
-				paused := newPausedSnapshot(t, 2, memory, "", "pause-snap", "node-1")
+				paused := newPausedSnapshot(t, 2, memory, "", "pause-snap", tc.locality)
 				paused.DurableSnapshot = &ateapipb.SnapshotStorage{Status: tc.recorded, Object: &ateapipb.ObjectSnapshot{SnapshotUri: uploaded.String()}}
 				s.AssignedNode = tc.node
 				s.LastAssignedGeneration = 2
@@ -372,11 +373,11 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 					paused,
 				}
 				if tc.state == ateapipb.ActorState_ACTOR_STATE_CRASHED {
-					s.Crash = newActorCrash("ResumeActor", "interrupted execution")
-				}
-				if tc.unfinalized {
-					paused.Locality = ""
-					s.Crash = newActorCrash("SuspendActor", "interrupted execution")
+					op := "ResumeActor"
+					if tc.locality == "" {
+						op = "SuspendActor"
+					}
+					s.Crash = newActorCrash(op, "interrupted execution")
 				}
 			})
 			atelet.terminateErr = apierror.NotFound("atelet holds nothing for the actor")

From 47221c5f45e049d9a22837af791df53d223ca51d Mon Sep 17 00:00:00 2001
From: zoezhao <zoezhao@google.com>
Date: Thu, 22 Oct 2026 15:10:46 -0700
Subject: [PATCH 6/7] ateapi: record the upload Terminate reports when a worker
 pod is gone

DeleteWorker terminated a released actor on its node but dropped the
state Terminate returned. atelet forgets the actor as it terminates it,
so a later revert got no upload back and deleted it as in progress.

The crash now records that upload in the same write, with the same
rules as a revert.
---
 .../workflow_reconcile_assignments.go         |  2 +-
 .../internal/controlapi/workflow_revert.go    | 58 ++++++++-------
 .../controlapi/workflow_revert_test.go        | 70 +++++++++++++++++++
 .../controlapi/workflow_worker_delete.go      | 35 +++++++---
 .../controlapi/workflow_worker_delete_test.go | 57 ++++++++++++---
 5 files changed, 175 insertions(+), 47 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go b/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go
index 28eb5c2ec8..17db669978 100644
--- a/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go
+++ b/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go
@@ -194,7 +194,7 @@ func (w *WorkerWorkflow) releaseEarlierAssignment(ctx context.Context, worker *a
 		return nil
 	}
 	if actor.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDED {
-		if err := w.crashBoundActor(ctx, worker, actorRef, actor, "Releasing actor from a restarted worker", crashMessageWorkerRestarted); err != nil {
+		if err := w.crashBoundActor(ctx, worker, actorRef, actor, nil, "Releasing actor from a restarted worker", crashMessageWorkerRestarted); err != nil {
 			return err
 		}
 	}
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go
index e0603ff2e3..11d1d78915 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert.go
@@ -217,34 +217,16 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 	return actor, w.releaseAssignmentWithoutBacklink(ctx, actor)
 }
 
-// recordTerminatedUpload marks COMPLETED the snapshot that the atelet on the
-// actor's assigned node reported durable in state as it terminated the actor,
-// with the same rules as a report. atelet may upload a snapshot without its
+// recordTerminatedUpload records the upload that Terminate reported in state,
+// as applyTerminatedUpload does. atelet may upload a snapshot without its
 // report landing, and a CRASHED actor refuses reports, so without this the
-// revert would delete the upload as in progress. It writes at once because
-// atelet forgot the actor, so a retried Terminate returns nothing.
+// revert would delete the upload as in progress. It writes at once because atelet forgot the actor, so a retried
+// Terminate returns nothing.
 func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, state *ateletpb.ActorSnapshotState) (*ateapipb.Actor, error) {
-	reported := state.GetLatestDurable().GetSnapshot()
-	if reported == nil {
-		return actor, nil
-	}
 	node := actor.GetStatus().GetAssignedNode()
-	// Only a finalized suspend places a snapshot on a node, so any other
-	// snapshot, such as one a crashed suspend created, stays PENDING and the
-	// revert deletes it with the rest of the discarded execution.
-	snap := findSnapshotByGeneration(actor.GetStatus().GetSnapshots(), reported.GetGeneration())
-	if got := snap.GetLocality(); got != node {
-		slog.InfoContext(ctx, "Discarding an upload Terminate reported for a snapshot not on its node",
-			slog.Any("actor", actorRef),
-			slog.String("node", node),
-			slog.String("snapshot_locality", got),
-			slog.Int("generation", int(reported.GetGeneration())),
-			slog.String("uuid", reported.GetSnapshotUuid()))
-		return actor, nil
-	}
 	applied := proto.CloneOf(actor.GetStatus())
-	if err := applyAteletResponseState(ctx, applied, state); err != nil {
-		return nil, apierror.Internal("atelet on node %q reported a durable snapshot for Actor %s that conflicts with its record: %w", node, actorRef, err)
+	if err := applyTerminatedUpload(ctx, actorRef, applied, state); err != nil {
+		return nil, err
 	}
 	if proto.Equal(applied, actor.GetStatus()) {
 		return actor, nil
@@ -263,6 +245,34 @@ func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef res
 	return stored, nil
 }
 
+// applyTerminatedUpload marks COMPLETED in status the snapshot that the atelet
+// on the actor's assigned node reported durable in state as it terminated the
+// actor, with the same rules as a report.
+func applyTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, status *ateapipb.ActorStatus, state *ateletpb.ActorSnapshotState) error {
+	reported := state.GetLatestDurable().GetSnapshot()
+	if reported == nil {
+		return nil
+	}
+	node := status.GetAssignedNode()
+	// Only a finalized suspend places a snapshot on a node, so any other
+	// snapshot, such as one a crashed suspend created, stays PENDING and a
+	// revert deletes it with the rest of the discarded execution.
+	snap := findSnapshotByGeneration(status.GetSnapshots(), reported.GetGeneration())
+	if got := snap.GetLocality(); got != node {
+		slog.InfoContext(ctx, "Discarding an upload Terminate reported for a snapshot not on its node",
+			slog.Any("actor", actorRef),
+			slog.String("node", node),
+			slog.String("snapshot_locality", got),
+			slog.Int("generation", int(reported.GetGeneration())),
+			slog.String("uuid", reported.GetSnapshotUuid()))
+		return nil
+	}
+	if err := applyAteletResponseState(ctx, status, state); err != nil {
+		return apierror.Internal("atelet on node %q reported a durable snapshot for Actor %s that conflicts with its record: %w", node, actorRef, err)
+	}
+	return nil
+}
+
 // ensureInProgressSnapshotDiscarded removes pending uploads of the discarded execution.
 // The actor keeps its latest completed durable snapshot.
 func (w *ActorWorkflow) ensureInProgressSnapshotDiscarded(ctx context.Context, actor *ateapipb.Actor) (err error) {
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
index 85a207242b..7c932856b5 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go
@@ -413,6 +413,76 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) {
 	}
 }
 
+// TestRecordTerminatedUpload verifies a revert records the upload Terminate
+// reports. A report the record already holds writes nothing. A report that
+// conflicts with the record, or a write that races another, fails the revert.
+func TestRecordTerminatedUpload(t *testing.T) {
+	completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
+	pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING
+	for _, tc := range []struct {
+		name string
+		// uploaded seeds the snapshot's upload as already recorded.
+		uploaded bool
+		// uuid is the snapshot uuid Terminate reports.
+		uuid string
+		// stale passes the actor as read before another write landed.
+		stale       bool
+		wantCode    codes.Code
+		wantDurable ateapipb.SnapshotStorageStatus
+		wantWrite   bool
+	}{
+		{name: "records the upload", uuid: "pause-snap", wantDurable: completed, wantWrite: true},
+		{name: "upload already recorded", uploaded: true, uuid: "pause-snap", wantDurable: completed},
+		{name: "report conflicts with the record", uuid: "other-snap", wantCode: codes.Internal, wantDurable: pending},
+		{name: "concurrent update", uuid: "pause-snap", stale: true, wantCode: codes.Aborted, wantDurable: pending},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			st, cleanup := storetest.SetupTestStore(t)
+			defer cleanup()
+			w := newTestActorWorkflow(t, st, "ns", "tmpl1")
+
+			ref := resources.ActorRef{Atespace: "team-a", Name: "id1"}
+			shape := withPausedShape(t, "node-1")
+			if tc.uploaded {
+				shape = withUploadedPausedShape(t, "node-1")
+			}
+			seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_REVERTING, shape)
+			actor, err := st.GetActor(ctx, ref)
+			if err != nil {
+				t.Fatalf("get seeded actor: %v", err)
+			}
+			before := actor
+			if tc.stale {
+				before = mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapipb.ActorStatus) {
+					s.Crash = newActorCrash("ResumeActor", "interrupted execution")
+				})
+			}
+			state := &ateletpb.ActorSnapshotState{LatestDurable: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{
+				SnapshotUuid: tc.uuid,
+				Generation:   1,
+				Fidelity:     ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY,
+				Object:       &ateletpb.ObjectStorage{SnapshotUri: someActorSnapshotURI(t, testStorageLocation, "team-a", "pause-snap")},
+			}}}
+
+			_, err = w.recordTerminatedUpload(ctx, ref, actor, state)
+			if got := apierror.Code(err); got != tc.wantCode {
+				t.Fatalf("recordTerminatedUpload = %v, want %v", err, tc.wantCode)
+			}
+			stored, err := st.GetActor(ctx, ref)
+			if err != nil {
+				t.Fatalf("GetActor: %v", err)
+			}
+			if got := stored.GetStatus().GetSnapshots()[0].GetDurableSnapshot().GetStatus(); got != tc.wantDurable {
+				t.Errorf("snapshot durable status = %v, want %v", got, tc.wantDurable)
+			}
+			if got := stored.GetMetadata().GetVersion() != before.GetMetadata().GetVersion(); got != tc.wantWrite {
+				t.Errorf("wrote the actor = %v, want %v", got, tc.wantWrite)
+			}
+		})
+	}
+}
+
 // TestRevertActor_RejectsSuspended covers both ways an actor can already be
 // SUSPENDED when a revert arrives: it never ran, or a previous revert already
 // finished. Both are FailedPrecondition — revert is not a no-op success on its
diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go
index 9513e68bee..07f849b39c 100644
--- a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go
+++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go
@@ -30,6 +30,7 @@ import (
 	"go.ate.dev/substrate/pkg/proto/ateapipb"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 // defaultWorkerReclaimTimeout bounds each Terminate a worker delete sends to
@@ -233,18 +234,18 @@ func (w *WorkerWorkflow) releaseBoundActor(ctx context.Context, worker *ateapipb
 	// crash keeps the actor's assigned node, so if this best-effort reclaim is
 	// skipped or fails, deleting or reverting the actor reclaims the state
 	// there later.
-	w.reclaimActorStateOnNode(ctx, worker, actor)
+	state := w.reclaimActorStateOnNode(ctx, worker, actor)
 
-	return w.crashBoundActor(ctx, worker, actorRef, actor, "Releasing actor from a worker whose pod is gone", crashMessageWorkerPodGone)
+	return w.crashBoundActor(ctx, worker, actorRef, actor, state, "Releasing actor from a worker whose pod is gone", crashMessageWorkerPodGone)
 }
 
 // crashBoundActor moves an Actor that lost its sandbox on the Worker to
 // ACTOR_STATE_CRASHED and clears its pointers at the Worker. crashMsg is
-// recorded in the Actor's status. An Actor in ACTOR_STATE_DELETING is left
-// untouched. The Actor's assignment row in the worker is left for the caller to
-// release.
+// recorded in the Actor's status, along with the upload Terminate reported in
+// state, if any. An Actor in ACTOR_STATE_DELETING is left untouched. The
+// Actor's assignment row in the worker is left for the caller to release.
 // A concurrent write to the Actor fails this as ABORTED.
-func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.Worker, actorRef resources.ActorRef, actor *ateapipb.Actor, logMsg, crashMsg string) error {
+func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.Worker, actorRef resources.ActorRef, actor *ateapipb.Actor, state *ateletpb.ActorSnapshotState, logMsg, crashMsg string) error {
 	// DELETING is final. A CRASHED actor can be reverted and resumed, and the
 	// delete may already have collected its volumes and snapshots. The delete's
 	// retry clears the actor's pointers.
@@ -266,6 +267,13 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.W
 
 	wasAlreadyCrashed := actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_CRASHED
 
+	// atelet forgot the actor as it terminated it, and a CRASHED actor refuses
+	// reports, so the crash records the upload Terminate reported.
+	applied := proto.CloneOf(actor.GetStatus())
+	if err := applyTerminatedUpload(ctx, actorRef, applied, state); err != nil {
+		return err
+	}
+
 	// Snapshot crash attributes before pod and pool pointers are cleared on actor.
 	sandboxClass := actor.GetStatus().GetWorkerAssignment().GetSandboxRuntime().GetSandboxClass()
 	crashAttrs := ateattr.ActorMetricAttributes(actor, sandboxClass, opName)
@@ -279,6 +287,8 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.W
 			toUpdate.Status.Crash = newActorCrash(opName, crashMsg)
 		}
 		toUpdate.Status.WorkerAssignment = nil
+		// The precondition makes toUpdate the version applied was built from.
+		toUpdate.Status.Snapshots = applied.GetSnapshots()
 		return nil
 	})
 	switch {
@@ -309,17 +319,18 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.W
 //
 // This call is best-effort so an unreachable node cannot wedge worker
 // deregistration. If it fails, the actor's revert or delete reclaims the state
-// on its assigned node. Each Terminate gets w.reclaimTimeout so a hung unmount,
+// on its assigned node. It returns the actor's state as Terminate returned it,
+// or nil if Terminate was skipped or failed. Each Terminate gets w.reclaimTimeout so a hung unmount,
 // such as against a dead NFS server, cannot run out the delete's own deadline
 // and fail the release after it.
-func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *ateapipb.Worker, actor *ateapipb.Actor) {
+func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *ateapipb.Worker, actor *ateapipb.Actor) *ateletpb.ActorSnapshotState {
 	ctx, done := stepSpan(ctx, "ReclaimActorStateOnNode")
 	defer func() { _ = done(nil) }()
 
 	nodeName := worker.GetNodeName()
 	if w.nodeGaveUp(nodeName) {
 		markSkipped(ctx, "an earlier reclaim on the node timed out or could not reach atelet")
-		return
+		return nil
 	}
 
 	actorRef := resources.ActorRefFromActor(actor)
@@ -337,7 +348,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at
 		w.giveUpOnNode(nodeName)
 		slog.WarnContext(ctx, "Could not reach the atelet holding a released actor's state; leaving it, and other actors' state on the node, for their revert or delete",
 			append(logAttrs, slog.Any("err", err))...)
-		return
+		return nil
 	}
 
 	slog.InfoContext(ctx, "Reclaiming the node state of an actor released from a worker whose pod is gone", logAttrs...)
@@ -345,7 +356,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at
 	// outlive it. The fallback spec carries the external volumes to unmount.
 	callCtx, cancel := context.WithTimeout(ctx, w.reclaimTimeout)
 	defer cancel()
-	_, err = ateletpb.NewAteletClient(conn).Terminate(callCtx, &ateletpb.TerminateRequest{
+	resp, err := ateletpb.NewAteletClient(conn).Terminate(callCtx, &ateletpb.TerminateRequest{
 		WorkerPodUid:          worker.GetWorkerPodUid(),
 		Atespace:              actor.GetMetadata().GetAtespace(),
 		ActorName:             actor.GetMetadata().GetName(),
@@ -356,6 +367,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at
 	})
 	switch status.Code(err) {
 	case codes.OK:
+		return resp.GetState()
 	case codes.NotFound:
 		slog.InfoContext(ctx, "Actor already terminated on its node", logAttrs...)
 	case codes.DeadlineExceeded, codes.Unavailable:
@@ -366,6 +378,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at
 		slog.WarnContext(ctx, "Failed to reclaim a released actor's node state; leaving it for the actor's revert or delete",
 			append(logAttrs, slog.Any("err", err))...)
 	}
+	return nil
 }
 
 // finalizeDeleted removes the worker from the store and returns the deleted
diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go
index 0f29a746af..b09ad5c76e 100644
--- a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go
+++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go
@@ -437,8 +437,13 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) {
 		// seed further shapes the stored actor.
 		seed func(*ateapipb.Actor)
 		// terminateErr is what the atelet answers, if anything.
-		terminateErr  error
-		wantTerminate bool
+		terminateErr error
+		// terminateState is the snapshot state a successful Terminate returns.
+		terminateState *ateletpb.ActorSnapshotState
+		wantTerminate  bool
+		// wantDurable is the durable status of the actor's snapshot after the
+		// crash, if it has one.
+		wantDurable ateapipb.SnapshotStorageStatus
 	}{
 		{
 			name:          "running actor is reclaimed",
@@ -453,6 +458,23 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) {
 			state:         ateapipb.ActorState_ACTOR_STATE_RUNNING,
 			seed:          withPausedShape(t, "node-1"),
 			wantTerminate: true,
+			wantDurable:   ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING,
+		},
+		{
+			// atelet forgets the actor as it terminates it, so the crash
+			// records the upload Terminate reports, or a revert would delete
+			// it as in progress.
+			name:  "upload terminate reports is recorded",
+			state: ateapipb.ActorState_ACTOR_STATE_RUNNING,
+			seed:  withPausedShape(t, "node-1"),
+			terminateState: &ateletpb.ActorSnapshotState{LatestDurable: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{
+				SnapshotUuid: "pause-snap",
+				Generation:   1,
+				Fidelity:     ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY,
+				Object:       &ateletpb.ObjectStorage{SnapshotUri: someActorSnapshotURI(t, testStorageLocation, "team-a", "pause-snap")},
+			}}},
+			wantTerminate: true,
+			wantDurable:   ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED,
 		},
 		{
 			// A cleanly suspended actor already ran Checkpoint's teardown,
@@ -478,7 +500,7 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) {
 			persistence, cleanup := storetest.SetupTestStore(t)
 			t.Cleanup(cleanup)
 
-			atelet := &capturingTerminator{err: tt.terminateErr}
+			atelet := &capturingTerminator{err: tt.terminateErr, state: tt.terminateState}
 			wf := NewWorkerWorkflow(persistence, newNodeAteletDialer(t, atelet))
 
 			seedAPIWorker(t, ctx, persistence, validWorker(apiWorkerName))
@@ -515,6 +537,18 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) {
 			if want := validWorker(apiWorkerName).GetWorkerPodUid(); req.GetWorkerPodUid() != want {
 				t.Errorf("Terminate worker_pod_uid = %q, want the worker's pod uid %q", req.GetWorkerPodUid(), want)
 			}
+
+			crashed, err := persistence.GetActor(ctx, apiActorRef)
+			if err != nil {
+				t.Fatalf("GetActor: %v", err)
+			}
+			var gotDurable ateapipb.SnapshotStorageStatus
+			if snaps := crashed.GetStatus().GetSnapshots(); len(snaps) > 0 {
+				gotDurable = snaps[0].GetDurableSnapshot().GetStatus()
+			}
+			if gotDurable != tt.wantDurable {
+				t.Errorf("snapshot durable status = %v, want %v", gotDurable, tt.wantDurable)
+			}
 		})
 	}
 }
@@ -725,21 +759,22 @@ func TestDeleteWorkerWorkflow_HungReclaimDoesNotWedgeDelete(t *testing.T) {
 }
 
 // capturingTerminator is an atelet that records the Terminate calls it is sent
-// and answers them with err, or, with hang set, blocks until the caller gives
-// up, like an atelet stuck in an unmount.
+// and answers them with err, or with state if err is nil. With hang set, it
+// blocks until the caller gives up, like an atelet stuck in an unmount.
 type capturingTerminator struct {
 	ateletpb.UnimplementedAteletServer
 
-	mu   sync.Mutex
-	err  error
-	hang bool
-	reqs []*ateletpb.TerminateRequest
+	mu    sync.Mutex
+	err   error
+	state *ateletpb.ActorSnapshotState
+	hang  bool
+	reqs  []*ateletpb.TerminateRequest
 }
 
 func (f *capturingTerminator) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) {
 	f.mu.Lock()
 	f.reqs = append(f.reqs, proto.Clone(req).(*ateletpb.TerminateRequest))
-	hang, err := f.hang, f.err
+	hang, err, state := f.hang, f.err, proto.CloneOf(f.state)
 	f.mu.Unlock()
 	if hang {
 		<-ctx.Done()
@@ -748,7 +783,7 @@ func (f *capturingTerminator) Terminate(ctx context.Context, req *ateletpb.Termi
 	if err != nil {
 		return nil, err
 	}
-	return &ateletpb.TerminateResponse{}, nil
+	return &ateletpb.TerminateResponse{State: state}, nil
 }
 
 func (f *capturingTerminator) requests() []*ateletpb.TerminateRequest {

From 1e906dfa2194e5756c21f532de2a5b120791ecf4 Mon Sep 17 00:00:00 2001
From: zoezhao <zoezhao@google.com>
Date: Thu, 22 Oct 2026 15:13:34 -0700
Subject: [PATCH 7/7] ateapi: record a CRASHED actor's reported upload

A report for a CRASHED actor failed with FailedPrecondition, which
atelet does not retry. The upload then reached the record only through
Terminate's response, and was lost if that response or the write after
it failed.

A report for a CRASHED actor now records its finished upload when the
record places the snapshot on the reporting node. It does not apply a
detach. Terminate's upload goes through the same check.
---
 .../internal/controlapi/snapshot_report.go    | 51 ++++++++++----
 .../controlapi/snapshot_report_test.go        | 70 ++++++++++++++++++-
 .../internal/controlapi/workflow_revert.go    | 29 ++------
 .../controlapi/workflow_worker_delete.go      |  4 +-
 pkg/proto/ateapipb/ateapi.proto               |  3 +-
 pkg/proto/ateapipb/ateapi_grpc.pb.go          |  6 +-
 6 files changed, 120 insertions(+), 43 deletions(-)

diff --git a/cmd/ateapi/internal/controlapi/snapshot_report.go b/cmd/ateapi/internal/controlapi/snapshot_report.go
index 549dc63aa3..fa0db2e062 100644
--- a/cmd/ateapi/internal/controlapi/snapshot_report.go
+++ b/cmd/ateapi/internal/controlapi/snapshot_report.go
@@ -37,9 +37,9 @@ func (s *RPCService) ApplyActorSnapshotState(ctx context.Context, actorRef resou
 }
 
 // ApplyReportedSnapshotState applies a report from the atelet on node to the
-// Actor. It holds the Actor's lease and applies reports only to a SUSPENDED or
-// RUNNING Actor, so a report never interleaves with a workflow. A report that
-// changes nothing writes nothing.
+// Actor. It holds the Actor's lease and applies reports only to a SUSPENDED,
+// RUNNING, or CRASHED Actor, so a report never interleaves with a workflow. A
+// report that changes nothing writes nothing.
 func (w *ActorWorkflow) ApplyReportedSnapshotState(ctx context.Context, actorRef resources.ActorRef, actorUID, node string, state *ateapipb.ActorSnapshotState) error {
 	ctx = preview.ContextWithOptIn(ctx)
 	// Check before taking the lease, so an atelet can't hold the lease of an
@@ -57,23 +57,26 @@ func (w *ActorWorkflow) ApplyReportedSnapshotState(ctx context.Context, actorRef
 		return err
 	}
 
+	applied := proto.Clone(before).(*ateapipb.Actor)
 	// A workflow owns a transitioning Actor's snapshots, even after its lease
-	// lapses. A crashed Actor's snapshots wait for a revert.
+	// lapses.
 	switch st := before.GetStatus().GetState(); st {
 	case ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ateapipb.ActorState_ACTOR_STATE_RUNNING:
+		if err := applyActorSnapshotState(leaseCtx, applied.GetStatus(), node, state); err != nil {
+			return err
+		}
 	case ateapipb.ActorState_ACTOR_STATE_CRASHED:
-		return apierror.FailedPrecondition("actor %s is %s; revert it first", actorRef, st)
+		// An Actor whose later generation crashed may have another async
+		// upload that uploaded its snapshot.
+		if err := applyUploadOnNode(leaseCtx, actorRef, applied.GetStatus(), node, state.GetLatestDurable()); err != nil {
+			return err
+		}
 	case ateapipb.ActorState_ACTOR_STATE_SUSPENDING, ateapipb.ActorState_ACTOR_STATE_RESUMING,
 		ateapipb.ActorState_ACTOR_STATE_DELETING, ateapipb.ActorState_ACTOR_STATE_REVERTING:
 		return apierror.Aborted("actor %s is %s; retry once that finishes", actorRef, st)
 	default:
 		return apierror.Internal("actor %s has unexpected state %s", actorRef, st)
 	}
-
-	applied := proto.Clone(before).(*ateapipb.Actor)
-	if err := applyActorSnapshotState(leaseCtx, applied.GetStatus(), node, state); err != nil {
-		return err
-	}
 	if proto.Equal(applied.GetStatus(), before.GetStatus()) {
 		return nil
 	}
@@ -154,14 +157,36 @@ func applyActorSnapshotState(ctx context.Context, status *ateapipb.ActorStatus,
 }
 
 // applyAteletResponseState records an upload that atelet's state, as returned
-// by Checkpoint, PromoteSnapshot, or Terminate, says finished, with the same
-// rules as a report. It records the fidelity atelet reports, which is the one
-// the upload was requested at. Unlike a report, a state with no local snapshot is not a
+// by Checkpoint or PromoteSnapshot, says finished, with the same rules as a
+// report. It records the fidelity atelet reports, which for both RPCs is the
+// one requested. Unlike a report, a state with no local snapshot is not a
 // detach. The workflow that made the call records where its snapshot lives.
 func applyAteletResponseState(ctx context.Context, status *ateapipb.ActorStatus, state *ateletpb.ActorSnapshotState) error {
 	return applyLatestDurable(ctx, status, snapshotReportFromAtelet(state.GetLatestDurable()))
 }
 
+// applyUploadOnNode marks the reported durable snapshot COMPLETED, as
+// applyLatestDurable does, if the record places that snapshot on node. Only a
+// finalized suspend places a snapshot on a node, so any other snapshot, such
+// as one a crashed suspend created, stays PENDING and a revert deletes it with
+// the rest of the discarded execution.
+func applyUploadOnNode(ctx context.Context, actorRef resources.ActorRef, status *ateapipb.ActorStatus, node string, reported *ateapipb.SnapshotReport) error {
+	if reported == nil {
+		return nil
+	}
+	snap := findSnapshotByGeneration(status.GetSnapshots(), reported.GetGeneration())
+	if got := snap.GetLocality(); got != node {
+		slog.InfoContext(ctx, "Discarding an upload reported for a snapshot not on its node",
+			slog.Any("actor", actorRef),
+			slog.String("node", node),
+			slog.String("snapshot_locality", got),
+			slog.Int("generation", int(reported.GetGeneration())),
+			slog.String("uuid", reported.GetUuid()))
+		return nil
+	}
+	return applyLatestDurable(ctx, status, reported)
+}
+
 // applyLatestDurable marks the reported durable snapshot COMPLETED, and records
 // its fidelity. Its uuid and upload URI must match the recorded ones, and its
 // fidelity must be set. The snapshot keeps its locality, so an Actor with a
diff --git a/cmd/ateapi/internal/controlapi/snapshot_report_test.go b/cmd/ateapi/internal/controlapi/snapshot_report_test.go
index 435244ff7c..054408c0e1 100644
--- a/cmd/ateapi/internal/controlapi/snapshot_report_test.go
+++ b/cmd/ateapi/internal/controlapi/snapshot_report_test.go
@@ -468,7 +468,7 @@ func TestApplyReportedSnapshotState(t *testing.T) {
 			assertUnwritten(t)
 		})
 	}
-	// A transitioning actor gets Aborted, and a crashed one FailedPrecondition.
+	// A transitioning actor gets Aborted.
 	for _, tc := range []struct {
 		state ateapipb.ActorState
 		want  codes.Code
@@ -477,7 +477,6 @@ func TestApplyReportedSnapshotState(t *testing.T) {
 		{ateapipb.ActorState_ACTOR_STATE_RESUMING, codes.Aborted},
 		{ateapipb.ActorState_ACTOR_STATE_DELETING, codes.Aborted},
 		{ateapipb.ActorState_ACTOR_STATE_REVERTING, codes.Aborted},
-		{ateapipb.ActorState_ACTOR_STATE_CRASHED, codes.FailedPrecondition},
 	} {
 		t.Run("refused for a "+tc.state.String()+" actor", func(t *testing.T) {
 			actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { s.State = tc.state })
@@ -519,6 +518,73 @@ func TestApplyReportedSnapshotState(t *testing.T) {
 	assertUnwritten(t)
 }
 
+// A report for a CRASHED actor records only the upload, and only of a snapshot
+// the record places on the reporting node. atelet forgets the actor when a
+// revert or delete terminates it, so the report is the upload's last chance
+// to be recorded before the revert deletes it as in progress.
+func TestApplyReportedSnapshotState_Crashed(t *testing.T) {
+	completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
+	pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING
+	for _, tc := range []struct {
+		name string
+		// locality is the snapshot's node. A suspend that crashed before it
+		// committed leaves its snapshot on no node.
+		locality string
+		// detach sends the report atelet sends once it detached the actor:
+		// no local snapshot and no upload in progress.
+		detach      bool
+		wantDurable ateapipb.SnapshotStorageStatus
+	}{
+		{name: "records the upload", locality: reportNode, wantDurable: completed},
+		{name: "does not apply a detach", locality: reportNode, detach: true, wantDurable: completed},
+		{name: "leaves a crashed suspend's upload pending", wantDurable: pending},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			persistence := newTestPersistence(t)
+			w := NewActorWorkflow(persistence, nil, nil, nil, nil, "", nil, nil)
+			actorRef := resources.ActorRef{Atespace: reportAtespace, Name: "actor-1"}
+			actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{
+				Metadata:      &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name},
+				ActorTemplate: &ateapipb.ObjectRef{Atespace: reportAtespace, Name: "tmpl"},
+				Status:        &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_CRASHED},
+			})
+			uid := actor.GetMetadata().GetUid()
+			uri := ownedURI(t, uid, 2)
+			actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) {
+				s.AssignedNode = reportNode
+				s.LastAssignedGeneration = 2
+				snap := newDurableSnapshot(2, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", snapName(2), uri, pending)
+				snap.Locality = tc.locality
+				s.Snapshots = []*ateapipb.Snapshot{snap}
+			})
+			uploaded := reportedAt(t, 2, uri)
+			state := &ateapipb.ActorSnapshotState{LatestLocal: uploaded, LatestDurable: uploaded}
+			if tc.detach {
+				state.LatestLocal = nil
+			}
+
+			if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, state); err != nil {
+				t.Fatalf("ApplyReportedSnapshotState() failed: %v", err)
+			}
+			got, err := persistence.GetActor(ctx, actorRef)
+			if err != nil {
+				t.Fatal(err)
+			}
+			snap := got.GetStatus().GetSnapshots()[0]
+			if st := snap.GetDurableSnapshot().GetStatus(); st != tc.wantDurable {
+				t.Errorf("durable snapshot status = %v, want %v", st, tc.wantDurable)
+			}
+			if snap.GetLocality() != tc.locality || got.GetStatus().GetAssignedNode() != reportNode {
+				t.Errorf("snapshot locality %q, assigned node %q, want %q and %q", snap.GetLocality(), got.GetStatus().GetAssignedNode(), tc.locality, reportNode)
+			}
+			if got.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_CRASHED {
+				t.Errorf("state = %v, want CRASHED", got.GetStatus().GetState())
+			}
+		})
+	}
+}
+
 func TestActorOnNode(t *testing.T) {
 	actor := func(assigned string, localities ...string) *ateapipb.Actor {
 		status := &ateapipb.ActorStatus{AssignedNode: assigned}
diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go
index 11d1d78915..cccb6d9809 100644
--- a/cmd/ateapi/internal/controlapi/workflow_revert.go
+++ b/cmd/ateapi/internal/controlapi/workflow_revert.go
@@ -219,8 +219,8 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso
 
 // recordTerminatedUpload records the upload that Terminate reported in state,
 // as applyTerminatedUpload does. atelet may upload a snapshot without its
-// report landing, and a CRASHED actor refuses reports, so without this the
-// revert would delete the upload as in progress. It writes at once because atelet forgot the actor, so a retried
+// report landing, so without this the revert would delete the upload as in
+// progress. It writes at once because atelet forgot the actor, so a retried
 // Terminate returns nothing.
 func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, state *ateletpb.ActorSnapshotState) (*ateapipb.Actor, error) {
 	node := actor.GetStatus().GetAssignedNode()
@@ -245,29 +245,12 @@ func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef res
 	return stored, nil
 }
 
-// applyTerminatedUpload marks COMPLETED in status the snapshot that the atelet
-// on the actor's assigned node reported durable in state as it terminated the
-// actor, with the same rules as a report.
+// applyTerminatedUpload records in status the upload that the atelet on the
+// actor's assigned node reported durable in state as it terminated the actor,
+// as applyUploadOnNode does.
 func applyTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, status *ateapipb.ActorStatus, state *ateletpb.ActorSnapshotState) error {
-	reported := state.GetLatestDurable().GetSnapshot()
-	if reported == nil {
-		return nil
-	}
 	node := status.GetAssignedNode()
-	// Only a finalized suspend places a snapshot on a node, so any other
-	// snapshot, such as one a crashed suspend created, stays PENDING and a
-	// revert deletes it with the rest of the discarded execution.
-	snap := findSnapshotByGeneration(status.GetSnapshots(), reported.GetGeneration())
-	if got := snap.GetLocality(); got != node {
-		slog.InfoContext(ctx, "Discarding an upload Terminate reported for a snapshot not on its node",
-			slog.Any("actor", actorRef),
-			slog.String("node", node),
-			slog.String("snapshot_locality", got),
-			slog.Int("generation", int(reported.GetGeneration())),
-			slog.String("uuid", reported.GetSnapshotUuid()))
-		return nil
-	}
-	if err := applyAteletResponseState(ctx, status, state); err != nil {
+	if err := applyUploadOnNode(ctx, actorRef, status, node, snapshotReportFromAtelet(state.GetLatestDurable())); err != nil {
 		return apierror.Internal("atelet on node %q reported a durable snapshot for Actor %s that conflicts with its record: %w", node, actorRef, err)
 	}
 	return nil
diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go
index 07f849b39c..ca58f2763d 100644
--- a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go
+++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go
@@ -267,8 +267,8 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.W
 
 	wasAlreadyCrashed := actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_CRASHED
 
-	// atelet forgot the actor as it terminated it, and a CRASHED actor refuses
-	// reports, so the crash records the upload Terminate reported.
+	// atelet forgot the actor as it terminated it, so the crash records the
+	// upload Terminate reported.
 	applied := proto.CloneOf(actor.GetStatus())
 	if err := applyTerminatedUpload(ctx, actorRef, applied, state); err != nil {
 		return err
diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto
index e453a318c9..c7f643be7d 100644
--- a/pkg/proto/ateapipb/ateapi.proto
+++ b/pkg/proto/ateapipb/ateapi.proto
@@ -2729,7 +2729,8 @@ service WorkerService {
   // atelet detached the Actor from the node. Only atelet observes either.
   //
   // atelet calls this with its own client certificate, for an Actor on its
-  // node. Each report carries the whole state, so repeating one is safe.
+  // node. Each report carries the whole state, so repeating one is safe. For a
+  // CRASHED Actor, the control plane records only the finished upload.
   // Returns ABORTED while another operation is changing the Actor; retry.
   // Returns NOT_FOUND once the Actor is gone or no longer on the caller's
   // node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report
diff --git a/pkg/proto/ateapipb/ateapi_grpc.pb.go b/pkg/proto/ateapipb/ateapi_grpc.pb.go
index f8a5e5be72..429f0610e5 100644
--- a/pkg/proto/ateapipb/ateapi_grpc.pb.go
+++ b/pkg/proto/ateapipb/ateapi_grpc.pb.go
@@ -1988,7 +1988,8 @@ type WorkerServiceClient interface {
 	// atelet detached the Actor from the node. Only atelet observes either.
 	//
 	// atelet calls this with its own client certificate, for an Actor on its
-	// node. Each report carries the whole state, so repeating one is safe.
+	// node. Each report carries the whole state, so repeating one is safe. For a
+	// CRASHED Actor, the control plane records only the finished upload.
 	// Returns ABORTED while another operation is changing the Actor; retry.
 	// Returns NOT_FOUND once the Actor is gone or no longer on the caller's
 	// node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report
@@ -2186,7 +2187,8 @@ type WorkerServiceServer interface {
 	// atelet detached the Actor from the node. Only atelet observes either.
 	//
 	// atelet calls this with its own client certificate, for an Actor on its
-	// node. Each report carries the whole state, so repeating one is safe.
+	// node. Each report carries the whole state, so repeating one is safe. For a
+	// CRASHED Actor, the control plane records only the finished upload.
 	// Returns ABORTED while another operation is changing the Actor; retry.
 	// Returns NOT_FOUND once the Actor is gone or no longer on the caller's
 	// node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report