fix(planning): a stale step copy could reset a claimed step and run it twice

The pending → ready and pending → cancelled transitions, and cancellation
of open steps, saved a step unconditionally. Under load an executor
holding a stale "pending" copy could write "ready" over a step another
executor had already claimed and was running, after which the step could
be claimed again: the full race suite caught two restarts from one plan.
These transitions now use SaveIf, which writes only when the stored status
is still the expected one. A deterministic regression test hands the
executor a stale copy and fails without the fix; the concurrency test now
races eight executors ticking repeatedly.

internal/planning/executor.go

@@ -247,14 +247,14 @@ func (e *Executor) advanceStep(ctx context.Context, p *Plan, it volition.Intenti
 			}
 			if dep.Status == StepFailed || dep.Status == StepCancelled {
 				st.Status, st.StatusReason = StepCancelled, "dependency "+d+" did not complete"
-				return true, e.Plans.Save(ctx, nil, *st)
+				return e.Plans.SaveIf(ctx, *st, StepPending)
 			}
 			if !StepDone(dep.Status) {
 				return false, nil
 			}
 		}
 		st.Status = StepReady
-		return true, e.Plans.Save(ctx, nil, *st)
+		return e.Plans.SaveIf(ctx, *st, StepPending)
 	case StepReady:
 		if !due || it.Status != volition.IntentionActive {
 			return false, nil
@@ -857,8 +857,9 @@ func (e *Executor) cancelOpenSteps(ctx context.Context, p Plan, reason string) {
 			e.Caps.WithdrawCapabilityRequest(ctx, p.MindID, s.RequestID, reason) //nolint:errcheck
 		}
 		e.dropSubordinate(ctx, &p, &s, reason)
+		from := s.Status
 		s.Status, s.StatusReason = StepCancelled, reason
-		e.Plans.Save(ctx, nil, s) //nolint:errcheck
+		e.Plans.SaveIf(ctx, s, from) //nolint:errcheck // a step that moved on meanwhile keeps its newer state
 	}
 }
 

internal/planning/executor_test.go

@@ -415,13 +415,17 @@ func TestConcurrentExecutorsRunAStepOnce(t *testing.T) {
 	f.host.fixOnRestart = true
 	it := f.intention(false)
 	p := f.start(it)
-	other := f.newExecutor(f.exec.Predictions, f.exec.Planner, f.exec.Log)
+	// Many executors race over the same plan from its first transition, each
+	// ticking repeatedly, as duplicate schedulers or several daemons would.
 	var wg sync.WaitGroup
-	for _, e := range []*Executor{f.exec, other, f.newExecutor(f.exec.Predictions, f.exec.Planner, f.exec.Log)} {
+	for i := 0; i < 8; i++ {
+		e := f.newExecutor(f.exec.Predictions, f.exec.Planner, f.exec.Log)
 		wg.Add(1)
 		go func(e *Executor) {
 			defer wg.Done()
+			for j := 0; j < 3; j++ {
 				e.TickMind(f.ctx, f.mind.ID) //nolint:errcheck
+			}
 		}(e)
 	}
 	wg.Wait()
@@ -505,3 +509,32 @@ func TestInvalidModelPlanNeverRuns(t *testing.T) {
 		t.Fatal("invalid plan executed")
 	}
 }
+
+// Regression: an executor holding a stale "pending" copy of a step must not
+// move it back to "ready" after another executor has claimed it, or the
+// step can be claimed and run a second time.
+func TestStaleStepCopyCannotMoveStepBackwards(t *testing.T) {
+	f := newFixture(t)
+	f.allow("service.restart")
+	it := f.intention(false)
+	p := f.start(it)
+	stale := f.step(p, "P1") // pending
+	fresh := stale
+	fresh.Status = StepReady
+	if ok, err := f.plans.SaveIf(f.ctx, fresh, StepPending); err != nil || !ok {
+		t.Fatalf("pending → ready: %v %v", ok, err)
+	}
+	if ok, err := f.plans.Claim(f.ctx, &fresh, StepReady, StepExecuting, time.Minute); err != nil || !ok {
+		t.Fatalf("claim: %v %v", ok, err)
+	}
+	byKey := map[string]*Step{}
+	for i := range p.Steps {
+		byKey[p.Steps[i].Key] = &p.Steps[i]
+	}
+	if _, err := f.exec.advanceStep(f.ctx, &p, it, &stale, byKey); err != nil {
+		t.Fatal(err)
+	}
+	if got := f.step(p, "P1"); got.Status != StepExecuting {
+		t.Fatalf("stale copy overwrote a claimed step: %s", got.Status)
+	}
+}

internal/planning/store.go

@@ -367,6 +367,24 @@ func (s *Store) Save(ctx context.Context, q storage.Querier, st Step) error {
 	return err
 }
 
+// SaveIf writes a step only if its stored status is still expected, so a
+// caller holding a stale copy can never move a step backwards (for example
+// resetting a step another executor is running to "ready"). It reports
+// whether the write happened.
+func (s *Store) SaveIf(ctx context.Context, st Step, expected string) (bool, error) {
+	now := time.Now()
+	var finished *time.Time
+	if StepTerminal(st.Status) {
+		finished = &now
+	}
+	tag, err := s.db.Exec(ctx, `UPDATE plan_steps SET status=$2, status_reason=$3, finished_at=coalesce($4, finished_at), updated_at=$5 WHERE id=$1 AND status=$6`,
+		st.ID, st.Status, trunc(st.StatusReason, 500), finished, now, expected)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
 // SetStatus writes a plan's status and reason.
 func (s *Store) SetStatus(ctx context.Context, q storage.Querier, id ids.ID, status, reason string) error {
 	if q == nil {