diff --git a/apps/backend/internal/backendapp/services.go b/apps/backend/internal/backendapp/services.go index 806d76592a..efd2dbf121 100644 --- a/apps/backend/internal/backendapp/services.go +++ b/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 } diff --git a/apps/backend/internal/backendapp/services_github_broker_test.go b/apps/backend/internal/backendapp/services_github_broker_test.go index de4ffe9993..64abf6adf3 100644 --- a/apps/backend/internal/backendapp/services_github_broker_test.go +++ b/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) + } + }) } } diff --git a/apps/backend/internal/mcp/handlers/config_handlers_test.go b/apps/backend/internal/mcp/handlers/config_handlers_test.go index 569d66071c..0035c32e43 100644 --- a/apps/backend/internal/mcp/handlers/config_handlers_test.go +++ b/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) { diff --git a/apps/backend/internal/mcp/handlers/config_task_handlers.go b/apps/backend/internal/mcp/handlers/config_task_handlers.go index 452c51c9b1..060b961c08 100644 --- a/apps/backend/internal/mcp/handlers/config_task_handlers.go +++ b/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 diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index 5cedfa9208..4b0e01191b 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/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. diff --git a/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go b/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go index 2835d358f1..65501c4925 100644 --- a/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go +++ b/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) } diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index e052296171..ac12a1f1cd 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/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, ) } diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go index 676c5a32f5..4cb75d3415 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go +++ b/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()) diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go index a0ec7f587e..b9af325049 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go +++ b/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, diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go index 22ad6e1fc7..477a235518 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go +++ b/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: diff --git a/apps/backend/internal/orchestrator/messagequeue/types.go b/apps/backend/internal/orchestrator/messagequeue/types.go index d87d90631d..6961df48d8 100644 --- a/apps/backend/internal/orchestrator/messagequeue/types.go +++ b/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"` diff --git a/apps/backend/internal/orchestrator/workflow_store.go b/apps/backend/internal/orchestrator/workflow_store.go index efc619ea95..a9e1779009 100644 --- a/apps/backend/internal/orchestrator/workflow_store.go +++ b/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) diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 6fad99eb36..4c6b2c3195 100644 --- a/apps/backend/internal/task/models/models.go +++ b/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