Refresh the actor when AssignWorkerStep's write hits a version conflict.

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -244,7 +244,22 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat
 
 	updatedActor, err := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion())
 	if err != nil {
-		return err
+		if !errors.Is(err, store.ErrPersistenceRetry) {
+			return err
+		}
+		// refresh the version of actor to avoid always failure in rest retries.
+		fresh, gerr := s.store.GetActor(ctx, input.Atespace, input.ActorName)
+		if gerr != nil {
+			slog.WarnContext(ctx, "Failed to refresh actor after assignment conflict", slog.Any("err", gerr))
+			return err
+		}
+		switch fresh.GetStatus() {
+		case ateapipb.Actor_STATUS_SUSPENDED, ateapipb.Actor_STATUS_PAUSED:
+			state.Actor = fresh
+			return err
+		default:
+			return status.Errorf(codes.Aborted, "actor %s is %s and can no longer be resumed", input.ActorName, fresh.GetStatus())
+		}
 	}
 	state.Actor = updatedActor
 	state.Worker = assignedWorker

cmd/ateapi/internal/controlapi/workflow_resume_test.go

@@ -16,9 +16,12 @@ package controlapi
 
 import (
 	"context"
+	"errors"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
@@ -404,6 +407,117 @@ func TestAssignWorkerStep_RetryAfterConflictPicksFreshWorker(t *testing.T) {
 	}
 }
 
+// conflictInjectingStore wraps a store and runs inject exactly once,
+// immediately before the first UpdateActor, simulating a concurrent writer
+// racing the step's read-modify-write window.
+type conflictInjectingStore struct {
+	store.Interface
+	once   sync.Once
+	inject func()
+}
+
+func (c *conflictInjectingStore) UpdateActor(ctx context.Context, actor *ateapipb.Actor, expectedVersion int64) (*ateapipb.Actor, error) {
+	c.once.Do(c.inject)
+	return c.Interface.UpdateActor(ctx, actor, expectedVersion)
+}
+
+// seedAssignFixture stores one free gvisor worker and a SUSPENDED actor and
+// returns the actor plus a started worker cache.
+func seedAssignFixture(t *testing.T, ctx context.Context, persistence store.Interface) (*ateapipb.Actor, *workercache.Cache) {
+	t.Helper()
+	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
+		WorkerNamespace: "worker-ns",
+		WorkerPool:      "pool",
+		WorkerPod:       "pod-1",
+		SandboxClass:    "gvisor",
+	}); err != nil {
+		t.Fatalf("CreateWorker: %v", err)
+	}
+	actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{
+		Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "id1"},
+		Status:   ateapipb.Actor_STATUS_SUSPENDED,
+	})
+	if err != nil {
+		t.Fatalf("CreateActor: %v", err)
+	}
+	cacheCtx, cancel := context.WithCancel(ctx)
+	t.Cleanup(cancel)
+	wc := workercache.New(persistence, time.Minute)
+	if err := wc.Start(cacheCtx); err != nil {
+		t.Fatalf("workercache.Start: %v", err)
+	}
+	return actor, wc
+}
+
+// runAssignConflict seeds the fixture, applies mutate as a concurrent write
+// racing the step's first UpdateActor, executes the step once, and returns
+// the stored actor, the state, the store, and the Execute error.
+func runAssignConflict(t *testing.T, mutate func(*ateapipb.Actor)) (*ateapipb.Actor, *ResumeState, store.Interface, error) {
+	t.Helper()
+	ctx := context.Background()
+	persistence := newTestPersistence(t)
+	actor, wc := seedAssignFixture(t, ctx, persistence)
+
+	st := &conflictInjectingStore{Interface: persistence, inject: func() {
+		fresh, err := persistence.GetActor(ctx, "team-a", "id1")
+		if err != nil {
+			t.Errorf("inject GetActor: %v", err)
+			return
+		}
+		mutate(fresh)
+		if _, err := persistence.UpdateActor(ctx, fresh, fresh.GetMetadata().GetVersion()); err != nil {
+			t.Errorf("inject UpdateActor: %v", err)
+		}
+	}}
+
+	step := &AssignWorkerStep{store: st, workerCache: wc}
+	state := &ResumeState{
+		Actor: actor,
+		ActorTemplate: &atev1alpha1.ActorTemplate{
+			Spec: atev1alpha1.ActorTemplateSpec{SandboxClass: atev1alpha1.SandboxClassGvisor},
+		},
+	}
+	execErr := step.Execute(ctx, &ResumeInput{ActorName: "id1", Atespace: "team-a"}, state)
+
+	stored, err := persistence.GetActor(ctx, "team-a", "id1")
+	if err != nil {
+		t.Fatalf("GetActor: %v", err)
+	}
+	return stored, state, persistence, execErr
+}
+
+// TestAssignWorkerStep_ConflictRefreshesActor verifies the actor write's
+// conflict handling within a single Execute: a concurrent spec write leaves
+// ErrPersistenceRetry with state.Actor refreshed, so the engine's backoff
+// retry applies on top of the concurrent write instead of replaying the same
+// stale version until the backoff is exhausted; an actor that left the
+// resumable states aborts the retry instead of being resurrected to RESUMING.
+func TestAssignWorkerStep_ConflictRefreshesActor(t *testing.T) {
+	t.Run("spec write refreshes state.Actor", func(t *testing.T) {
+		_, state, _, err := runAssignConflict(t, func(a *ateapipb.Actor) {
+			a.WorkerSelector = &ateapipb.Selector{MatchLabels: map[string]string{"team": "blue"}}
+		})
+		if !errors.Is(err, store.ErrPersistenceRetry) {
+			t.Fatalf("Execute: %v, want ErrPersistenceRetry", err)
+		}
+		if got := state.Actor.GetWorkerSelector().GetMatchLabels()["team"]; got != "blue" {
+			t.Errorf("state.Actor selector team = %q, want blue (refreshed with the concurrent write)", got)
+		}
+	})
+
+	t.Run("crash aborts instead of resurrecting", func(t *testing.T) {
+		stored, _, _, err := runAssignConflict(t, func(a *ateapipb.Actor) {
+			a.Status = ateapipb.Actor_STATUS_CRASHED
+		})
+		if got := status.Code(err); got != codes.Aborted {
+			t.Fatalf("status.Code(err) = %v, want %v (err: %v)", got, codes.Aborted, err)
+		}
+		if stored.GetStatus() != ateapipb.Actor_STATUS_CRASHED {
+			t.Errorf("status = %v, want CRASHED (retry must not resurrect the actor)", stored.GetStatus())
+		}
+	})
+}
+
 // TestResumeActorWorkflow_RejectedAndIdempotentPaths covers the two
 // short-circuit paths of the resume workflow: rejection by AssignWorkerStep's
 // CheckPrerequisite and the IsComplete idempotent fast-forward.