fix(orchestrator): prevent stale pending move replay (#2735)

Co-authored-by: ayattara <alassane.yattara@savoirfairelinux.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/backendapp/services.go

@@ -395,11 +395,11 @@ func isCanonicalKandevRepositoryInput(
 	repository *taskmodels.Repository,
 ) bool {
 	if repository != nil {
-		return repository.Provider == "github" &&
+		return repository.Provider == gitCredentialGitHubProviderID &&
 			isPublicGitHubProviderHost(repository.ProviderHost) &&
 			repository.ProviderOwner == canonicalKandevOwner && repository.ProviderName == canonicalKandevName
 	}
-	return input != nil && input.Provider == "github" &&
+	return input != nil && input.Provider == gitCredentialGitHubProviderID &&
 		isPublicGitHubProviderHost(input.ProviderHost) &&
 		input.ProviderOwner == canonicalKandevOwner && input.ProviderName == canonicalKandevName
 }

apps/backend/internal/backendapp/services_github_broker_test.go

@@ -348,18 +348,26 @@ func TestGitHubBrokerScopeAuthorizerAllowsOnlyTheBoundContributionDestination(t
 }
 
 func TestIsCanonicalKandevRepositoryInputRequiresPublicGitHubHost(t *testing.T) {
-	for _, host := range []string{"", "https://github.enterprise.example"} {
-		input := &taskservice.TaskRepositoryInput{
-			Provider: "github", ProviderHost: host, ProviderOwner: "kdlbs", ProviderName: "kandev",
-		}
-		if isCanonicalKandevRepositoryInput(input, nil) {
-			t.Fatalf("host %q was accepted as the public canonical GitHub repository", host)
-		}
+	tests := []struct {
+		name  string
+		input *taskservice.TaskRepositoryInput
+		repo  *taskmodels.Repository
+		valid bool
+	}{
+		{name: "input canonical", input: &taskservice.TaskRepositoryInput{Provider: "github", ProviderHost: "https://github.com", ProviderOwner: "kdlbs", ProviderName: "kandev"}, valid: true},
+		{name: "repository canonical", repo: &taskmodels.Repository{Provider: "github", ProviderHost: "github.com", ProviderOwner: "kdlbs", ProviderName: "kandev"}, valid: true},
+		{name: "input empty host", input: &taskservice.TaskRepositoryInput{Provider: "github", ProviderOwner: "kdlbs", ProviderName: "kandev"}},
+		{name: "input enterprise host", input: &taskservice.TaskRepositoryInput{Provider: "github", ProviderHost: "https://github.enterprise.example", ProviderOwner: "kdlbs", ProviderName: "kandev"}},
+		{name: "input provider mismatch", input: &taskservice.TaskRepositoryInput{Provider: "gitlab", ProviderHost: "https://github.com", ProviderOwner: "kdlbs", ProviderName: "kandev"}},
+		{name: "input owner mismatch", input: &taskservice.TaskRepositoryInput{Provider: "github", ProviderHost: "https://github.com", ProviderOwner: "other", ProviderName: "kandev"}},
+		{name: "repository name mismatch", repo: &taskmodels.Repository{Provider: "github", ProviderHost: "github.com", ProviderOwner: "kdlbs", ProviderName: "other"}},
 	}
-	if !isCanonicalKandevRepositoryInput(&taskservice.TaskRepositoryInput{
-		Provider: "github", ProviderHost: "https://github.com", ProviderOwner: "kdlbs", ProviderName: "kandev",
-	}, nil) {
-		t.Fatal("public GitHub canonical repository was rejected")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isCanonicalKandevRepositoryInput(tt.input, tt.repo); got != tt.valid {
+				t.Fatalf("isCanonicalKandevRepositoryInput() = %v, want %v", got, tt.valid)
+			}
+		})
 	}
 }
 

apps/backend/internal/mcp/handlers/config_handlers_test.go

@@ -384,7 +384,11 @@ type recordingMessageQueuer struct {
 	calls []messagequeue.QueuedMessage
 }
 
-func (r *recordingMessageQueuer) QueueMessage(_ context.Context, sessionID, taskID, content, model, userID string, planMode bool, _ []messagequeue.MessageAttachment) (*messagequeue.QueuedMessage, error) {
+func (r *recordingMessageQueuer) QueueMessage(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []messagequeue.MessageAttachment) (*messagequeue.QueuedMessage, error) {
+	return r.QueueMessageWithMetadata(ctx, sessionID, taskID, content, model, userID, planMode, attachments, nil)
+}
+
+func (r *recordingMessageQueuer) QueueMessageWithMetadata(_ context.Context, sessionID, taskID, content, model, userID string, planMode bool, _ []messagequeue.MessageAttachment, metadata map[string]interface{}) (*messagequeue.QueuedMessage, error) {
 	msg := messagequeue.QueuedMessage{
 		SessionID: sessionID,
 		TaskID:    taskID,
@@ -392,6 +396,7 @@ func (r *recordingMessageQueuer) QueueMessage(_ context.Context, sessionID, task
 		Model:     model,
 		PlanMode:  planMode,
 		QueuedBy:  userID,
+		Metadata:  metadata,
 	}
 	r.calls = append(r.calls, msg)
 	return &msg, nil
@@ -465,7 +470,7 @@ func TestQueueMoveTaskPrompt_QueuesWithExpectedFields(t *testing.T) {
 	assert.Equal(t, "session-99", got.SessionID)
 	assert.Equal(t, "task-1", got.TaskID)
 	assert.Equal(t, "Please fix the failing test in foo_test.go", got.Content)
-	assert.Equal(t, "mcp-move-task", got.QueuedBy)
+	assert.Equal(t, messagequeue.QueuedByMoveTask, got.QueuedBy)
 	assert.False(t, got.PlanMode)
 	assert.Equal(t, "", got.Model)
 }
@@ -998,6 +1003,7 @@ func TestDeferMoveTask_AcceptsValidStep(t *testing.T) {
 		"workflow_id":       "wf-defer3",
 		"workflow_step_id":  "dst-step3",
 		"position":          0,
+		"prompt":            "continue the work",
 		"sender_session_id": "sess-caller3",
 	})
 
@@ -1007,6 +1013,9 @@ func TestDeferMoveTask_AcceptsValidStep(t *testing.T) {
 	require.Len(t, queue.pendingMoves, 1)
 	assert.Equal(t, "dst-step3", queue.pendingMoves[0].WorkflowStepID)
 	assert.Equal(t, "sess-caller3", queue.pendingMoves[0].SenderSessionID)
+	assert.NotEmpty(t, queue.pendingMoves[0].MoveID)
+	require.Len(t, queue.calls, 1)
+	assert.Equal(t, queue.pendingMoves[0].MoveID, queue.calls[0].Metadata[messagequeue.MetadataDeferredMoveID])
 }
 
 func TestMoveTaskErrorMessage_SanitizesClassifiedErrors(t *testing.T) {

apps/backend/internal/mcp/handlers/config_task_handlers.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/google/uuid"
+
 	"github.com/kandev/kandev/internal/automation"
 	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
 	"github.com/kandev/kandev/internal/steptelemetry"
@@ -154,9 +156,10 @@ func (h *Handlers) deferMoveTask(
 		}
 	}
 
+	moveID := uuid.NewString()
 	if req.Prompt != "" {
 		wrapped := "You were moved to this step with the following message: " + req.Prompt
-		if err := h.queueMoveTaskPrompt(ctx, req.TaskID, session.ID, wrapped); err != nil {
+		if err := h.queueMoveTaskPromptWithMoveID(ctx, req.TaskID, session.ID, wrapped, moveID); err != nil {
 			h.logger.Error("move_task: failed to queue hand-off prompt",
 				zap.String("task_id", req.TaskID), zap.String("session_id", session.ID), zap.Error(err))
 			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError,
@@ -164,6 +167,7 @@ func (h *Handlers) deferMoveTask(
 		}
 	}
 	h.messageQueue.SetPendingMove(ctx, session.ID, &messagequeue.PendingMove{
+		MoveID:          moveID,
 		TaskID:          req.TaskID,
 		WorkflowID:      req.WorkflowID,
 		WorkflowStepID:  req.WorkflowStepID,
@@ -321,13 +325,27 @@ func (h *Handlers) lookupSession(ctx context.Context, taskID string) (*models.Ta
 // or proceed (idle path), since a queue failure makes the deferred contract
 // impossible to honor.
 func (h *Handlers) queueMoveTaskPrompt(ctx context.Context, taskID, sessionID, prompt string) error {
+	return h.queueMoveTaskPromptWithMoveID(ctx, taskID, sessionID, prompt, "")
+}
+
+func (h *Handlers) queueMoveTaskPromptWithMoveID(ctx context.Context, taskID, sessionID, prompt, moveID string) error {
 	if h.messageQueue == nil {
 		return fmt.Errorf("message queue is unavailable")
 	}
 	if sessionID == "" {
 		return fmt.Errorf("task has no primary session")
 	}
-	if _, err := h.messageQueue.QueueMessage(ctx, sessionID, taskID, prompt, "", "mcp-move-task", false, nil); err != nil {
+	metadata := map[string]interface{}(nil)
+	if moveID != "" {
+		metadata = map[string]interface{}{messagequeue.MetadataDeferredMoveID: moveID}
+	}
+	if queueWithMetadata, ok := h.messageQueue.(messageMetadataQueuer); ok {
+		if _, err := queueWithMetadata.QueueMessageWithMetadata(ctx, sessionID, taskID, prompt, "", messagequeue.QueuedByMoveTask, false, nil, metadata); err != nil {
+			return fmt.Errorf("queue message: %w", err)
+		}
+		return nil
+	}
+	if _, err := h.messageQueue.QueueMessage(ctx, sessionID, taskID, prompt, "", messagequeue.QueuedByMoveTask, false, nil); err != nil {
 		return fmt.Errorf("queue message: %w", err)
 	}
 	return nil

apps/backend/internal/mcp/handlers/handlers.go

@@ -185,6 +185,13 @@ type MessageQueuer interface {
 	TakeQueued(ctx context.Context, sessionID string) (*messagequeue.QueuedMessage, bool)
 }
 
+// messageMetadataQueuer is an optional extension implemented by the
+// production queue service. Keeping metadata out of MessageQueuer preserves
+// compatibility with lightweight test and alternate queue implementations.
+type messageMetadataQueuer interface {
+	QueueMessageWithMetadata(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []messagequeue.MessageAttachment, metadata map[string]interface{}) (*messagequeue.QueuedMessage, error)
+}
+
 // PromptReferenceResolver expands saved prompt references that appear inside
 // agent-sent prompts while preserving the original @mentions in the visible
 // prompt body.

apps/backend/internal/orchestrator/event_handlers_pending_move_test.go

@@ -138,6 +138,138 @@ func TestPendingMove_OutOfTerminalStepReopensCompletedTask(t *testing.T) {
 	}
 }
 
+func TestPendingMove_DoesNotReplayAfterStaleSnapshotRestored(t *testing.T) {
+	sc := buildPendingMoveScenario(t)
+	session, err := sc.repo.GetTaskSession(sc.ctx, sc.reviewSessionID)
+	if err != nil {
+		t.Fatalf("load review session: %v", err)
+	}
+	queuedAt := time.Date(2026, 8, 17, 2, 23, 48, 0, time.UTC)
+	firstMove := &messagequeue.PendingMove{
+		TaskID:         "task-1",
+		WorkflowID:     "wf1",
+		WorkflowStepID: stepInProgressID,
+		QueuedAt:       queuedAt,
+	}
+
+	// Consume the deferred move once, as handleAgentReady does at turn end.
+	sc.svc.applyPendingMove(sc.ctx, "task-1", sc.reviewSessionID, session, firstMove)
+	task, err := sc.repo.GetTask(sc.ctx, "task-1")
+	if err != nil {
+		t.Fatalf("load task after first move: %v", err)
+	}
+	if task.WorkflowStepID != stepInProgressID {
+		t.Fatalf("workflow_step_id after first move = %q, want %q", task.WorkflowStepID, stepInProgressID)
+	}
+
+	// Simulate a legitimate later return to Review, followed by a stale queue
+	// rollback restoring the original already-consumed legacy pending move. The
+	// restored row predates move_id, so applyPendingMove must derive the same
+	// stable identity from the persisted legacy fields instead of relying on the
+	// mutated firstMove pointer above.
+	task.WorkflowStepID = stepInReviewID
+	if err := sc.repo.UpdateTask(sc.ctx, task); err != nil {
+		t.Fatalf("return task to review: %v", err)
+	}
+	staleRestoredMove := &messagequeue.PendingMove{
+		TaskID:         "task-1",
+		WorkflowID:     "wf1",
+		WorkflowStepID: stepInProgressID,
+		QueuedAt:       queuedAt,
+	}
+	sc.svc.applyPendingMove(sc.ctx, "task-1", sc.reviewSessionID, session, staleRestoredMove)
+
+	task, err = sc.repo.GetTask(sc.ctx, "task-1")
+	if err != nil {
+		t.Fatalf("load task after stale replay: %v", err)
+	}
+	if task.WorkflowStepID != stepInReviewID {
+		t.Fatalf("workflow_step_id after stale replay = %q, want %q", task.WorkflowStepID, stepInReviewID)
+	}
+}
+
+func TestPendingMove_EqualTargetRecordsAppliedMoveID(t *testing.T) {
+	sc := buildPendingMoveScenario(t)
+	session, err := sc.repo.GetTaskSession(sc.ctx, sc.reviewSessionID)
+	if err != nil {
+		t.Fatalf("load review session: %v", err)
+	}
+	task, err := sc.repo.GetTask(sc.ctx, "task-1")
+	if err != nil {
+		t.Fatalf("load task: %v", err)
+	}
+	task.WorkflowStepID = stepInProgressID
+	if err := sc.repo.UpdateTask(sc.ctx, task); err != nil {
+		t.Fatalf("move task to target step: %v", err)
+	}
+
+	const moveID = "move-equal-target"
+	sc.svc.applyPendingMove(sc.ctx, "task-1", sc.reviewSessionID, session, &messagequeue.PendingMove{
+		MoveID:         moveID,
+		TaskID:         "task-1",
+		WorkflowID:     "wf1",
+		WorkflowStepID: stepInProgressID,
+	})
+
+	updated, err := sc.repo.GetTask(sc.ctx, "task-1")
+	if err != nil {
+		t.Fatalf("reload task: %v", err)
+	}
+	applied, ok := updated.Metadata[models.MetaKeyAppliedDeferredMoves].(map[string]interface{})
+	if !ok || applied[moveID] != true {
+		t.Fatalf("applied deferred moves = %#v, want %q", updated.Metadata[models.MetaKeyAppliedDeferredMoves], moveID)
+	}
+}
+
+func TestPendingMove_DuplicateRemovesOnlyMatchingHandoffPrompt(t *testing.T) {
+	sc := buildPendingMoveScenario(t)
+	session, err := sc.repo.GetTaskSession(sc.ctx, sc.reviewSessionID)
+	if err != nil {
+		t.Fatalf("load review session: %v", err)
+	}
+
+	const moveID = "move-duplicate"
+	task, err := sc.repo.GetTask(sc.ctx, "task-1")
+	if err != nil {
+		t.Fatalf("load task: %v", err)
+	}
+	task.Metadata = map[string]interface{}{
+		models.MetaKeyAppliedDeferredMoves: map[string]interface{}{moveID: true},
+	}
+	if err := sc.repo.UpdateTask(sc.ctx, task); err != nil {
+		t.Fatalf("mark deferred move as applied: %v", err)
+	}
+	if _, ok := sc.svc.messageQueue.TakeQueued(sc.ctx, sc.reviewSessionID); !ok {
+		t.Fatalf("remove scenario handoff prompt")
+	}
+
+	if _, err := sc.svc.messageQueue.QueueMessageWithMetadata(
+		sc.ctx, sc.reviewSessionID, "task-1", "stale handoff", "",
+		messagequeue.QueuedByMoveTask, false, nil,
+		map[string]interface{}{messagequeue.MetadataDeferredMoveID: moveID},
+	); err != nil {
+		t.Fatalf("queue stale handoff: %v", err)
+	}
+	if _, err := sc.svc.messageQueue.QueueMessage(
+		sc.ctx, sc.reviewSessionID, "task-1", "unrelated prompt", "",
+		messagequeue.QueuedByUser, false, nil,
+	); err != nil {
+		t.Fatalf("queue unrelated prompt: %v", err)
+	}
+
+	sc.svc.applyPendingMove(sc.ctx, "task-1", sc.reviewSessionID, session, &messagequeue.PendingMove{
+		MoveID:         moveID,
+		TaskID:         "task-1",
+		WorkflowID:     "wf1",
+		WorkflowStepID: stepInProgressID,
+	})
+
+	status := sc.svc.messageQueue.GetStatus(sc.ctx, sc.reviewSessionID)
+	if len(status.Entries) != 1 || status.Entries[0].Content != "unrelated prompt" {
+		t.Fatalf("remaining queue entries = %#v, want only unrelated prompt", status.Entries)
+	}
+}
+
 func TestPendingMove_DropsForeignWorkflowStepWithoutMovingTask(t *testing.T) {
 	sc := buildPendingMoveScenario(t)
 	sc.stepGetter.steps["foreign-step"] = &wfmodels.WorkflowStep{
@@ -273,7 +405,7 @@ func buildPendingMoveScenario(t *testing.T) *pendingMoveScenario {
 	const handoffPrompt = "You were moved to this step with the following message: " +
 		"The file fibonacci.py has two bugs — fix them."
 	if _, err := svc.messageQueue.QueueMessage(
-		ctx, reviewSessionID, "task-1", handoffPrompt, "", "mcp-move-task", false, nil,
+		ctx, reviewSessionID, "task-1", handoffPrompt, "", messagequeue.QueuedByMoveTask, false, nil,
 	); err != nil {
 		t.Fatalf("queue hand-off prompt: %v", err)
 	}

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -2,9 +2,11 @@ package orchestrator
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -27,6 +29,8 @@ import (
 
 type turnCompletionCause string
 
+var errDeferredMoveAlreadyApplied = errors.New("deferred move already applied")
+
 type taskMetadataKeyRemover interface {
 	RemoveTaskMetadataKey(context.Context, string, string) (bool, error)
 }
@@ -1944,9 +1948,9 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
 // for the source step and on_enter for the target step. Bypasses
 // task.Service.MoveTask (and the task.moved event) so the orchestrator's async
 // task.moved handler doesn't run a second processStepExitAndEnter for the same
-// transition. The message queue is left intact — any user-supplied prompt
-// already queued by handleMoveTask is delivered by the on_enter path or by
-// drainQueuedMessageForPromptableSession.
+// transition. The move hand-off prompt is correlated by MoveID and removed
+// only when the move is already complete or rejected; unrelated queued
+// messages remain available for the normal drain path.
 func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string, session *models.TaskSession, move *messagequeue.PendingMove) {
 	// reinsertPendingMove restores the move so a future agent.ready can retry.
 	// Used on early failure paths (load errors, config issues) where the state
@@ -1968,6 +1972,9 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string
 		reinsertPendingMove()
 		return
 	}
+	if move.MoveID == "" {
+		move.MoveID = legacyPendingMoveID(sessionID, move)
+	}
 
 	task, err := s.repo.GetTask(ctx, taskID)
 	if err != nil {
@@ -1979,12 +1986,21 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string
 	}
 	fromStepID := task.WorkflowStepID
 	if fromStepID == move.WorkflowStepID {
-		// Step already matches — nothing to transition. Just leave the queued
-		// prompt for the natural drain path. Don't reinsert: the move is
-		// effectively complete since the task is already at the target step.
+		if err := s.workflowStore.MarkDeferredMoveApplied(ctx, taskID, move.MoveID); errors.Is(err, errDeferredMoveAlreadyApplied) {
+			s.logger.Info("dropping already-applied pending move at current target",
+				zap.String("task_id", taskID), zap.String("move_id", move.MoveID))
+		} else if err != nil {
+			s.logger.Error("failed to record pending move at current target",
+				zap.String("task_id", taskID), zap.String("move_id", move.MoveID), zap.Error(err))
+			reinsertPendingMove()
+			return
+		}
+		// Step already matches. The move is complete, so consume its hand-off
+		// prompt before the natural drain path handles unrelated messages.
 		s.logger.Info("pending move target equals current step; skipping transition",
 			zap.String("task_id", taskID),
 			zap.String("step_id", fromStepID))
+		s.removePendingMoveHandoffPrompt(ctx, sessionID, taskID, move.MoveID)
 		s.drainQueuedMessageForPromptableSessionLocked(ctx, sessionID)
 		return
 	}
@@ -2011,14 +2027,7 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string
 		// misdelivered to the source step's agent on a future turn (it was
 		// authored for the move's *target* step) — mirrors the cleanup done on
 		// the ApplyTransition failure path below.
-		if s.messageQueue != nil {
-			if _, ok := s.messageQueue.TakeQueued(ctx, sessionID); ok {
-				s.publishQueueStatusEvent(ctx, sessionID)
-				s.logger.Warn("dropped hand-off prompt after pending-move workflow mismatch",
-					zap.String("task_id", taskID),
-					zap.String("session_id", sessionID))
-			}
-		}
+		s.removePendingMoveHandoffPrompt(ctx, sessionID, taskID, move.MoveID)
 		return
 	}
 
@@ -2040,7 +2049,11 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string
 		deferredMoveAttribution.SessionID = move.SenderSessionID
 	}
 	deferredMoveCtx := steptelemetry.WithAttribution(ctx, deferredMoveAttribution)
-	if err := s.workflowStore.ApplyTransition(deferredMoveCtx, taskID, sessionID, fromStepID, move.WorkflowStepID, engine.TriggerOnEnter); err != nil {
+	if err := s.workflowStore.ApplyDeferredMoveTransition(deferredMoveCtx, taskID, sessionID, fromStepID, move.WorkflowStepID, move.MoveID); errors.Is(err, errDeferredMoveAlreadyApplied) {
+		s.logger.Info("dropping already-applied pending move", zap.String("task_id", taskID), zap.String("move_id", move.MoveID))
+		s.removePendingMoveHandoffPrompt(ctx, sessionID, taskID, move.MoveID)
+		return
+	} else if err != nil {
 		s.logger.Error("failed to apply pending move transition",
 			zap.String("task_id", taskID),
 			zap.Error(err))
@@ -2049,14 +2062,7 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string
 		// drained the queue won't run, and handleAgentReady has already returned.
 		// Drop the orphan so it can't be misdelivered to the source step's agent
 		// on a future turn (it was authored for the move's *target* step).
-		if s.messageQueue != nil {
-			if _, ok := s.messageQueue.TakeQueued(ctx, sessionID); ok {
-				s.publishQueueStatusEvent(ctx, sessionID)
-				s.logger.Warn("dropped hand-off prompt after pending-move transition failure",
-					zap.String("task_id", taskID),
-					zap.String("session_id", sessionID))
-			}
-		}
+		s.removePendingMoveHandoffPrompt(ctx, sessionID, taskID, move.MoveID)
 		return
 	}
 
@@ -2096,6 +2102,45 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string
 	go s.processStepExitAndEnter(context.WithoutCancel(ctx), taskID, session, fromStepID, move.WorkflowStepID, taskDescription)
 }
 
+func legacyPendingMoveID(sessionID string, move *messagequeue.PendingMove) string {
+	identity := fmt.Sprintf("%s\x00%s\x00%s\x00%s\x00%d\x00%s\x00%s\x00%s",
+		sessionID, move.TaskID, move.WorkflowID, move.WorkflowStepID, move.Position,
+		move.QueuedAt.UTC().Format(time.RFC3339Nano), move.Actor, move.SenderSessionID)
+	sum := sha256.Sum256([]byte(identity))
+	return fmt.Sprintf("legacy-%x", sum[:])
+}
+
+func (s *Service) removePendingMoveHandoffPrompt(ctx context.Context, sessionID, taskID, moveID string) {
+	if s.messageQueue == nil {
+		return
+	}
+	for _, entry := range s.messageQueue.GetStatus(ctx, sessionID).Entries {
+		if entry.TaskID != taskID || entry.QueuedBy != messagequeue.QueuedByMoveTask {
+			continue
+		}
+		entryMoveID, _ := entry.Metadata[messagequeue.MetadataDeferredMoveID].(string)
+		matches := entryMoveID == moveID
+		if entryMoveID == "" && strings.HasPrefix(moveID, "legacy-") {
+			matches = true
+		}
+		if !matches {
+			continue
+		}
+		_, removed, err := s.messageQueue.TakeQueuedEntry(ctx, sessionID, entry.ID)
+		if err != nil {
+			s.logger.Warn("failed to remove stale pending-move hand-off prompt",
+				zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.Error(err))
+			return
+		}
+		if removed {
+			s.publishQueueStatusEvent(ctx, sessionID)
+			s.logger.Warn("dropped stale pending-move hand-off prompt",
+				zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.String("move_id", moveID))
+		}
+		return
+	}
+}
+
 func (s *Service) syncTaskStateForPendingMove(ctx context.Context, taskID, fromStepID, toStepID string) {
 	if s.workflowStepIsTerminal(ctx, toStepID) {
 		s.markTaskCompletedForTerminalStep(ctx, taskID, toStepID)
@@ -2324,6 +2369,38 @@ func workflowMessageMetadata(planMode bool, origin workflowMessageOrigin, refere
 	return meta
 }
 
+func (s *Service) resolveAutoStartPromptContext(
+	ctx context.Context,
+	taskID string,
+	session *models.TaskSession,
+) (bool, *models.Task, bool, error) {
+	isOfficeTask, err := s.lookupOfficeTask(ctx, taskID)
+	if err != nil {
+		return false, nil, false, fmt.Errorf("resolve MCP mode for workflow auto-start: %w", err)
+	}
+	if isOfficeTask {
+		return true, nil, false, nil
+	}
+
+	taskForPrompt, err := s.repo.GetTask(ctx, taskID)
+	if err != nil {
+		return false, nil, false, fmt.Errorf("load task for autopilot prompt: %w", err)
+	}
+	if session.State != models.TaskSessionStateCreated {
+		return false, taskForPrompt, false, nil
+	}
+
+	configMode, _ := session.Metadata["config_mode"].(bool)
+	if configMode {
+		return false, taskForPrompt, false, nil
+	}
+	titleOwner, err := s.ClaimTaskTitleSession(ctx, taskID, session.ID)
+	if err != nil {
+		return false, nil, false, fmt.Errorf("claim task title for workflow auto-start: %w", err)
+	}
+	return false, taskForPrompt, titleOwner, nil
+}
+
 // handleCreatedAutoStartLaunchFailure preserves a workflow prompt when the
 // created-session launch loses a concurrent-start race. The prompt was already
 // recorded in chat history, so queueing it is the only way to deliver it after
@@ -2415,31 +2492,17 @@ func (s *Service) autoStartStepPrompt(
 	titleOwner := false
 	isOfficeTask := false
 	var taskForPrompt *models.Task
-	if session.State == models.TaskSessionStateCreated {
-		var officeErr error
-		isOfficeTask, officeErr = s.lookupOfficeTask(ctx, taskID)
-		if officeErr != nil {
+	needsRuntimeContext := session.State == models.TaskSessionStateCreated ||
+		(step != nil && step.HasOnEnterAction(wfmodels.OnEnterResetAgentContext))
+	if needsRuntimeContext {
+		var contextErr error
+		isOfficeTask, taskForPrompt, titleOwner, contextErr = s.resolveAutoStartPromptContext(ctx, taskID, session)
+		if contextErr != nil {
 			requeueTaken()
-			return fmt.Errorf("resolve MCP mode for workflow auto-start: %w", officeErr)
-		}
-		if !isOfficeTask {
-			taskForPrompt, officeErr = s.repo.GetTask(ctx, taskID)
-			if officeErr != nil {
-				requeueTaken()
-				return fmt.Errorf("load task for autopilot prompt: %w", officeErr)
-			}
-		}
-		configMode, _ := session.Metadata["config_mode"].(bool)
-		if !configMode && !isOfficeTask {
-			var claimErr error
-			titleOwner, claimErr = s.ClaimTaskTitleSession(ctx, taskID, sessionID)
-			if claimErr != nil {
-				requeueTaken()
-				return fmt.Errorf("claim task title for workflow auto-start: %w", claimErr)
-			}
+			return contextErr
 		}
 	}
-	if session.State == models.TaskSessionStateCreated && !session.IsPassthrough && (agentPrompt != "" || len(attachments) > 0) {
+	if (session.State == models.TaskSessionStateCreated || step.HasOnEnterAction(wfmodels.OnEnterResetAgentContext)) && !session.IsPassthrough && (agentPrompt != "" || len(attachments) > 0) {
 		configMode, _ := session.Metadata["config_mode"].(bool)
 		requiresSignal := step != nil && step.AutoAdvanceRequiresSignal
 		referenceContext := EntityReferenceContext(references)
@@ -2483,7 +2546,7 @@ func (s *Service) autoStartStepPrompt(
 
 	const maxRetryAttempts = 5
 	for attempt := 1; attempt <= maxRetryAttempts; attempt++ {
-		_, err := s.PromptTask(ctx, taskID, sessionID, agentPrompt, "", planMode, attachments, false)
+		_, err := s.PromptTask(ctx, taskID, sessionID, recordedPrompt, "", planMode, attachments, false)
 		if err == nil {
 			return nil
 		}
@@ -2500,7 +2563,7 @@ func (s *Service) autoStartStepPrompt(
 				zap.String("session_id", sessionID),
 				zap.String("step_name", stepName))
 			return s.fallbackFreshLaunchOnMissingExecution(
-				ctx, taskID, sessionID, agentPrompt, planMode, takenMsg, attachments, references,
+				ctx, taskID, sessionID, recordedPrompt, planMode, takenMsg, attachments, references,
 			)
 		}
 

apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go

@@ -15,10 +15,35 @@ import (
 	"github.com/kandev/kandev/internal/orchestrator/scheduler"
 	"github.com/kandev/kandev/internal/sysprompt"
 	"github.com/kandev/kandev/internal/task/models"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
 	"github.com/kandev/kandev/internal/workflow/engine"
 	wfmodels "github.com/kandev/kandev/internal/workflow/models"
 )
 
+func seedAutopilotTaskAndSession(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID string, sessionState models.TaskSessionState) {
+	t.Helper()
+	ctx := context.Background()
+	now := time.Now().UTC()
+	if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil {
+		t.Fatalf("create workspace: %v", err)
+	}
+	if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "Test Workflow", CreatedAt: now, UpdatedAt: now}); err != nil {
+		t.Fatalf("create workflow: %v", err)
+	}
+	if err := repo.CreateTask(ctx, &models.Task{
+		ID: taskID, WorkspaceID: "ws1", WorkflowID: "wf1", Title: "Test Task",
+		State: v1.TaskStateInProgress, ParentID: "parent-task", Autopilot: true,
+		CreatedAt: now, UpdatedAt: now,
+	}); err != nil {
+		t.Fatalf("create autopilot task: %v", err)
+	}
+	if err := repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID: sessionID, TaskID: taskID, State: sessionState, StartedAt: now, UpdatedAt: now,
+	}); err != nil {
+		t.Fatalf("create task session: %v", err)
+	}
+}
+
 func TestAutoStartStepPrompt_OfficeWithoutRuntimeEnvFailsClosed(t *testing.T) {
 	ctx := context.Background()
 	repo := setupTestRepo(t)
@@ -106,6 +131,79 @@ func TestAutoStartStepPrompt_OfficeWithoutRuntimeEnvFailsClosed(t *testing.T) {
 	}
 }
 
+func TestAutoStartStepPrompt_ResetContextInjectsCompletionContractForReusedSession(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedAutopilotTaskAndSession(t, repo, "task-reused", "session-reused", models.TaskSessionStateWaitingForInput)
+	task, err := repo.GetTask(ctx, "task-reused")
+	if err != nil {
+		t.Fatalf("load task: %v", err)
+	}
+	session, _ := repo.GetTaskSession(ctx, "session-reused")
+	session.AgentExecutionID = "execution-reused"
+	session.AgentProfileID = "profile-review"
+	_ = repo.UpdateTaskSession(ctx, session)
+	seedExecutorRunning(t, repo, session.ID, task.ID, session.AgentExecutionID)
+	step := &wfmodels.WorkflowStep{ID: "step-review", WorkflowID: "wf1", Name: "Review", AutoAdvanceRequiresSignal: true, Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterResetAgentContext}, {Type: wfmodels.OnEnterAutoStartAgent}}}}
+	stepGetter := newMockStepGetter()
+	stepGetter.steps[step.ID] = step
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo, isAgentRunning: true}
+	messages := &mockMessageCreator{}
+	svc := createTestServiceWithScheduler(repo, stepGetter, newMockTaskRepo(), agentMgr)
+	svc.messageCreator = messages
+	err = svc.autoStartStepPrompt(ctx, "task-reused", session, step, "Review the change", false, false)
+	if err != nil {
+		t.Fatalf("autoStartStepPrompt returned error: %v", err)
+	}
+	if len(messages.userMessages) != 1 || !strings.Contains(messages.userMessages[0].content, "step_complete_kandev") {
+		t.Fatalf("reused reset-context prompt lacks completion contract: %#v", messages.userMessages)
+	}
+	if len(agentMgr.capturedPromptCalls) != 1 || !strings.Contains(agentMgr.capturedPromptCalls[0].Prompt, "step_complete_kandev") {
+		t.Fatalf("executor prompt lacks completion contract: %#v", agentMgr.capturedPromptCalls)
+	}
+	if !strings.Contains(messages.userMessages[0].content, "ask_parent_question_kandev") || strings.Contains(messages.userMessages[0].content, "ask_user_question_kandev") {
+		t.Fatalf("reused autopilot prompt has the wrong question contract: %s", messages.userMessages[0].content)
+	}
+}
+
+func TestAutoStartStepPrompt_ResetContextPreservesOfficeModeForReusedSession(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "task-office-reused", "session-office-reused", models.TaskSessionStateWaitingForInput)
+	task, err := repo.GetTask(ctx, "task-office-reused")
+	if err != nil {
+		t.Fatalf("load task: %v", err)
+	}
+	task.ProjectID = "project-office"
+	if err := repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("set Office ownership: %v", err)
+	}
+	session, err := repo.GetTaskSession(ctx, "session-office-reused")
+	if err != nil {
+		t.Fatalf("load session: %v", err)
+	}
+	session.AgentExecutionID = "execution-office-reused"
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("update session: %v", err)
+	}
+	seedExecutorRunning(t, repo, session.ID, task.ID, session.AgentExecutionID)
+	step := &wfmodels.WorkflowStep{ID: "step-office-reused", WorkflowID: "wf1", Name: "Office", Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterResetAgentContext}, {Type: wfmodels.OnEnterAutoStartAgent}}}}
+	stepGetter := newMockStepGetter()
+	stepGetter.steps[step.ID] = step
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo, isAgentRunning: true}
+	messages := &mockMessageCreator{}
+	svc := createTestServiceWithScheduler(repo, stepGetter, newMockTaskRepo(), agentMgr)
+	svc.messageCreator = messages
+	if err := svc.autoStartStepPrompt(ctx, task.ID, session, step, "Run the Office task", false, false); err != nil {
+		t.Fatalf("autoStartStepPrompt returned error: %v", err)
+	}
+	if len(messages.userMessages) != 1 {
+		t.Fatalf("recorded messages = %d, want 1", len(messages.userMessages))
+	}
+	if !strings.Contains(messages.userMessages[0].content, "KANDEV OFFICE MCP TOOLS") || strings.Contains(messages.userMessages[0].content, "list_workspaces_kandev") {
+		t.Fatalf("reused Office prompt has the wrong tool contract: %s", messages.userMessages[0].content)
+	}
+}
 func TestResolveStepAgentProfile(t *testing.T) {
 	t.Run("returns step profile when set", func(t *testing.T) {
 		svc := createTestService(setupTestRepo(t), newMockStepGetter(), newMockTaskRepo())

apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go

@@ -160,6 +160,7 @@ func (r *sqliteRepository) initSchema() error {
 
 	CREATE TABLE IF NOT EXISTS pending_moves (
 		id               TEXT PRIMARY KEY,
+		move_id          TEXT NOT NULL DEFAULT '',
 		session_id       TEXT NOT NULL UNIQUE,
 		task_id          TEXT NOT NULL DEFAULT '',
 		workflow_id      TEXT NOT NULL DEFAULT '',
@@ -191,6 +192,9 @@ func (r *sqliteRepository) initSchema() error {
 	if _, alterErr := r.db.Exec(`ALTER TABLE pending_moves ADD COLUMN sender_session_id TEXT NOT NULL DEFAULT ''`); alterErr != nil && !internaldb.IsDuplicateColumnError(alterErr) {
 		return alterErr
 	}
+	if _, alterErr := r.db.Exec(`ALTER TABLE pending_moves ADD COLUMN move_id TEXT NOT NULL DEFAULT ''`); alterErr != nil && !internaldb.IsDuplicateColumnError(alterErr) {
+		return alterErr
+	}
 	return nil
 }
 
@@ -2147,10 +2151,10 @@ func (r *sqliteRepository) ReplaceSession(ctx context.Context, sessionID string,
 			queuedAt = time.Now().UTC()
 		}
 		if _, err := tx.ExecContext(ctx, r.db.Rebind(`
-			INSERT INTO pending_moves (id, session_id, task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+			INSERT INTO pending_moves (id, move_id, session_id, task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id)
+			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 		`),
-			uuid.New().String(), sessionID, pendingMove.TaskID, pendingMove.WorkflowID,
+			uuid.New().String(), pendingMove.MoveID, sessionID, pendingMove.TaskID, pendingMove.WorkflowID,
 			pendingMove.WorkflowStepID, pendingMove.Position, queuedAt, pendingMove.Actor,
 			pendingMove.SenderSessionID,
 		); err != nil {
@@ -2178,8 +2182,8 @@ func (r *sqliteRepository) SetPendingMove(ctx context.Context, sessionID string,
 		return err
 	}
 	if _, err := tx.ExecContext(ctx, r.db.Rebind(`
-		INSERT INTO pending_moves (id, session_id, task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+		INSERT INTO pending_moves (id, move_id, session_id, task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 		ON CONFLICT(session_id) DO UPDATE SET
 			task_id = excluded.task_id,
 			workflow_id = excluded.workflow_id,
@@ -2187,9 +2191,10 @@ func (r *sqliteRepository) SetPendingMove(ctx context.Context, sessionID string,
 			step_position = excluded.step_position,
 			queued_at = excluded.queued_at,
 			actor = excluded.actor,
-			sender_session_id = excluded.sender_session_id
+			sender_session_id = excluded.sender_session_id,
+			move_id = excluded.move_id
 	`),
-		uuid.New().String(), sessionID, move.TaskID, move.WorkflowID, move.WorkflowStepID, move.Position, move.QueuedAt, move.Actor, move.SenderSessionID,
+		uuid.New().String(), move.MoveID, sessionID, move.TaskID, move.WorkflowID, move.WorkflowStepID, move.Position, move.QueuedAt, move.Actor, move.SenderSessionID,
 	); err != nil {
 		return fmt.Errorf("upsert pending move: %w", err)
 	}
@@ -2199,21 +2204,22 @@ func (r *sqliteRepository) SetPendingMove(ctx context.Context, sessionID string,
 // GetPendingMove returns the deferred workflow move for a session, or nil when absent.
 func (r *sqliteRepository) GetPendingMove(ctx context.Context, sessionID string) (*PendingMove, error) {
 	var (
-		taskID, workflowID, workflowStepID string
-		position                           int
-		queuedAt                           time.Time
-		actor, senderSessionID             string
+		moveID, taskID, workflowID, workflowStepID string
+		position                                   int
+		queuedAt                                   time.Time
+		actor, senderSessionID                     string
 	)
 	if err := r.ro.QueryRowxContext(ctx, r.db.Rebind(`
-		SELECT task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id
+		SELECT move_id, task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id
 		FROM pending_moves WHERE session_id = ?
-	`), sessionID).Scan(&taskID, &workflowID, &workflowStepID, &position, &queuedAt, &actor, &senderSessionID); err != nil {
+	`), sessionID).Scan(&moveID, &taskID, &workflowID, &workflowStepID, &position, &queuedAt, &actor, &senderSessionID); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, nil
 		}
 		return nil, fmt.Errorf("read pending move: %w", err)
 	}
 	return &PendingMove{
+		MoveID:          moveID,
 		TaskID:          taskID,
 		WorkflowID:      workflowID,
 		WorkflowStepID:  workflowStepID,
@@ -2238,15 +2244,15 @@ func (r *sqliteRepository) TakePendingMove(ctx context.Context, sessionID string
 	}
 
 	var (
-		taskID, workflowID, workflowStepID string
-		position                           int
-		queuedAt                           time.Time
-		actor, senderSessionID             string
+		moveID, taskID, workflowID, workflowStepID string
+		position                                   int
+		queuedAt                                   time.Time
+		actor, senderSessionID                     string
 	)
 	if err := tx.QueryRowxContext(ctx, r.db.Rebind(`
-		SELECT task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id
+		SELECT move_id, task_id, workflow_id, workflow_step_id, step_position, queued_at, actor, sender_session_id
 		FROM pending_moves WHERE session_id = ?
-	`), sessionID).Scan(&taskID, &workflowID, &workflowStepID, &position, &queuedAt, &actor, &senderSessionID); err != nil {
+	`), sessionID).Scan(&moveID, &taskID, &workflowID, &workflowStepID, &position, &queuedAt, &actor, &senderSessionID); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, nil
 		}
@@ -2259,6 +2265,7 @@ func (r *sqliteRepository) TakePendingMove(ctx context.Context, sessionID string
 		return nil, err
 	}
 	return &PendingMove{
+		MoveID:          moveID,
 		TaskID:          taskID,
 		WorkflowID:      workflowID,
 		WorkflowStepID:  workflowStepID,

apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go

@@ -774,6 +774,7 @@ func TestSQLiteRepository_ReplaceSessionPreservesQueuedIdentity(t *testing.T) {
 	}
 
 	if err := repo.ReplaceSession(ctx, "s1", []QueuedMessage{*original}, &PendingMove{
+		MoveID:          "move-restore",
 		TaskID:          "t1",
 		WorkflowStepID:  "step-a",
 		QueuedAt:        original.QueuedAt,
@@ -801,6 +802,9 @@ func TestSQLiteRepository_ReplaceSessionPreservesQueuedIdentity(t *testing.T) {
 	if move == nil || move.WorkflowStepID != "step-a" {
 		t.Fatalf("pending move = %#v, want step-a", move)
 	}
+	if move.MoveID != "move-restore" {
+		t.Fatalf("pending move move_id = %q, want move-restore", move.MoveID)
+	}
 	if move.SenderSessionID != "sender-s1" {
 		t.Fatalf("pending move sender_session_id = %q, want sender-s1", move.SenderSessionID)
 	}
@@ -814,12 +818,13 @@ func TestSQLiteRepository_PendingMove(t *testing.T) {
 		t.Fatalf("expected nil move on empty, got %v err=%v", move, err)
 	}
 
-	move := &PendingMove{TaskID: "t1", WorkflowID: "w1", WorkflowStepID: "step-A", Position: 0, Actor: "agent", SenderSessionID: "sender-s1"}
+	move := &PendingMove{MoveID: "move-a", TaskID: "t1", WorkflowID: "w1", WorkflowStepID: "step-A", Position: 0, Actor: "agent", SenderSessionID: "sender-s1"}
 	if err := repo.SetPendingMove(ctx, "s1", move); err != nil {
 		t.Fatalf("set pending: %v", err)
 	}
 
 	// Upsert: replace with new target.
+	move.MoveID = "move-b"
 	move.WorkflowStepID = "step-B"
 	if err := repo.SetPendingMove(ctx, "s1", move); err != nil {
 		t.Fatalf("upsert pending: %v", err)
@@ -832,6 +837,9 @@ func TestSQLiteRepository_PendingMove(t *testing.T) {
 	if got == nil || got.WorkflowStepID != "step-B" {
 		t.Errorf("expected step-B after upsert, got %+v", got)
 	}
+	if got == nil || got.MoveID != "move-b" {
+		t.Errorf("expected move-b move ID after upsert, got %+v", got)
+	}
 	if got == nil || got.Actor != "agent" {
 		t.Errorf("expected agent actor after upsert, got %+v", got)
 	}
@@ -875,7 +883,7 @@ func TestSQLiteRepository_PendingMoveSenderSessionMigration(t *testing.T) {
 	}
 	ctx := context.Background()
 	if err := repo.SetPendingMove(ctx, "s1", &PendingMove{
-		TaskID: "t1", WorkflowStepID: "step-a", SenderSessionID: "sender-s1",
+		MoveID: "move-migrated", TaskID: "t1", WorkflowStepID: "step-a", SenderSessionID: "sender-s1",
 	}); err != nil {
 		t.Fatalf("set migrated pending move: %v", err)
 	}
@@ -886,6 +894,9 @@ func TestSQLiteRepository_PendingMoveSenderSessionMigration(t *testing.T) {
 	if move == nil || move.SenderSessionID != "sender-s1" {
 		t.Fatalf("migrated pending move = %#v, want sender-s1", move)
 	}
+	if move.MoveID != "move-migrated" {
+		t.Fatalf("migrated pending move move_id = %q, want move-migrated", move.MoveID)
+	}
 }
 
 // TestSQLiteRepository_ConcurrentInsertCap exercises the cap under contention:

apps/backend/internal/orchestrator/messagequeue/types.go

@@ -18,6 +18,7 @@ const (
 	QueuedByAgent    = "agent"
 	QueuedByWorkflow = "workflow"
 	QueuedByServer   = "server"
+	QueuedByMoveTask = "mcp-move-task"
 )
 
 // IsReservedQueuedBy reports identities owned by backend dispatch paths.
@@ -72,6 +73,11 @@ const MetadataLifecycleReserved = "lifecycle_reserved_in_flight"
 // never mixes prompts issued by different agents.
 const MetadataSenderTaskID = "sender_task_id"
 
+// MetadataDeferredMoveID identifies the hand-off prompt created for one
+// deferred workflow move. The orchestrator uses it to remove only stale move
+// prompts after a replay.
+const MetadataDeferredMoveID = "deferred_move_id"
+
 // QueueFullErrorCode is the well-known WS / MCP error code surfaced when an
 // insert would exceed the per-session cap. Shared between the user-side WS
 // handlers and the inter-task MCP handler so the wire contract stays in sync.
@@ -205,6 +211,10 @@ type QueueStatus struct {
 // move_task_kandev) while its turn is still active. Applied by handleAgentReady
 // once the turn ends.
 type PendingMove struct {
+	// MoveID identifies one deferred move request across queue snapshots. A
+	// rollback can restore a previously consumed snapshot, so the orchestrator
+	// needs a durable identity to reject that stale replay.
+	MoveID         string    `json:"move_id"`
 	TaskID         string    `json:"task_id"`
 	WorkflowID     string    `json:"workflow_id"`
 	WorkflowStepID string    `json:"workflow_step_id"`

apps/backend/internal/orchestrator/workflow_store.go

@@ -184,6 +184,31 @@ func (s *workflowStore) LoadPreviousStep(ctx context.Context, workflowID string,
 }
 
 func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID, fromStepID, toStepID string, trigger engine.Trigger) error {
+	return s.applyTransition(ctx, taskID, sessionID, fromStepID, toStepID, trigger, "")
+}
+
+func (s *workflowStore) ApplyDeferredMoveTransition(ctx context.Context, taskID, sessionID, fromStepID, toStepID, moveID string) error {
+	return s.applyTransition(ctx, taskID, sessionID, fromStepID, toStepID, engine.TriggerOnEnter, moveID)
+}
+
+func (s *workflowStore) MarkDeferredMoveApplied(ctx context.Context, taskID, moveID string) error {
+	if moveID == "" {
+		return nil
+	}
+	task, err := s.repo.GetTask(ctx, taskID)
+	if err != nil {
+		return fmt.Errorf("load task for deferred move identity: %w", err)
+	}
+	if err := markDeferredMoveApplied(task, moveID); err != nil {
+		return err
+	}
+	if err := s.repo.UpdateTask(ctx, task); err != nil {
+		return fmt.Errorf("persist deferred move identity: %w", err)
+	}
+	return nil
+}
+
+func (s *workflowStore) applyTransition(ctx context.Context, taskID, sessionID, fromStepID, toStepID string, trigger engine.Trigger, moveID string) error {
 	task, err := s.repo.GetTask(ctx, taskID)
 	if err != nil {
 		return fmt.Errorf("load task for transition: %w", err)
@@ -197,6 +222,10 @@ func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID,
 	// task.WorkflowID already), but applyPendingMove uses this path for
 	// cross-workflow move_task_kandev hand-offs too — without this, the task
 	// would end up with a step ID from a workflow its WorkflowID doesn't match.
+	if err := markDeferredMoveApplied(task, moveID); err != nil {
+		return err
+	}
+
 	oldWorkflowID := task.WorkflowID
 	if targetStep != nil {
 		task.WorkflowID = targetStep.WorkflowID
@@ -247,6 +276,25 @@ func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID,
 	return nil
 }
 
+func markDeferredMoveApplied(task *models.Task, moveID string) error {
+	if moveID == "" {
+		return nil
+	}
+	applied, _ := task.Metadata[models.MetaKeyAppliedDeferredMoves].(map[string]interface{})
+	if _, exists := applied[moveID]; exists {
+		return errDeferredMoveAlreadyApplied
+	}
+	if applied == nil {
+		applied = make(map[string]interface{})
+	}
+	applied[moveID] = true
+	if task.Metadata == nil {
+		task.Metadata = make(map[string]interface{})
+	}
+	task.Metadata[models.MetaKeyAppliedDeferredMoves] = applied
+	return nil
+}
+
 func (s *workflowStore) updateTransitionTask(ctx context.Context, task *models.Task, targetStep *wfmodels.WorkflowStep) error {
 	if targetStep == nil {
 		return s.repo.UpdateTask(ctx, task)

apps/backend/internal/task/models/models.go

@@ -107,6 +107,9 @@ const (
 	// after queue promotion. It prevents duplicate task.queue_promoted events
 	// from repeating on_enter or auto-start behavior.
 	MetaKeyQueuePromotionPending = "queue_promotion_pending"
+	// MetaKeyAppliedDeferredMoves stores deferred move IDs that have already
+	// been applied, preventing a stale queue rollback from replaying one.
+	MetaKeyAppliedDeferredMoves = "applied_deferred_moves"
 	// DeferredLaunchStartWhenUnblockedKey marks a deferred launch intent as
 	// belonging to a task dependency chain rather than to WIP overflow. The
 	// record itself is identical; the flag lets dependency resolution recognise