fix: complete same-step MCP task moves (#3872) (#3876)

This commit is contained in:
Carlos Florêncio
2026-09-23 10:57:22 +01:00
committed by GitHub
parent 41eeaf0c70
commit 0386e3e9b7
20 changed files with 1001 additions and 16 deletions
@@ -9,15 +9,18 @@ import (
"github.com/google/uuid"
"github.com/kandev/kandev/internal/authz"
"github.com/kandev/kandev/internal/automation"
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
"github.com/kandev/kandev/internal/steptelemetry"
"github.com/kandev/kandev/internal/task/dto"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/repository/repoerrors"
taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite"
"github.com/kandev/kandev/internal/task/service"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
workflowmove "github.com/kandev/kandev/internal/workflow/move"
workflowservice "github.com/kandev/kandev/internal/workflow/service"
v1 "github.com/kandev/kandev/pkg/api/v1"
ws "github.com/kandev/kandev/pkg/websocket"
"go.uber.org/zap"
@@ -69,6 +72,9 @@ func (h *Handlers) handleMoveTask(ctx context.Context, msg *ws.Message) (*ws.Mes
}
req.EntryOptions = entryOptions
req.Prompt = ""
if response, handled, err := h.completeSameStepMove(ctx, msg, req); handled || err != nil {
return response, err
}
// entry_options are OPTIONAL — config-mode/admin moves don't always have an
// agent to hand off to. When supplied, one-shot instructions/reset/profile
@@ -100,6 +106,111 @@ func (h *Handlers) handleMoveTask(ctx context.Context, msg *ws.Message) (*ws.Mes
return h.applyMoveTaskImmediate(ctx, msg, req, session)
}
func (h *Handlers) completeSameStepMove(ctx context.Context, msg *ws.Message, req moveTaskRequest) (*ws.Message, bool, error) {
if h.taskSvc == nil {
return moveTaskErrorResponse(msg, ws.ErrorCodeInternalError, "task service is not configured")
}
task, err := h.taskSvc.GetTask(ctx, req.TaskID)
if err != nil || task == nil {
h.logger.Error("move_task: failed to load task for destination classification",
zap.String("task_id", req.TaskID), zap.Error(err))
return moveTaskErrorResponse(msg, ws.ErrorCodeInternalError, "failed to load task for move validation")
}
if task.WorkflowID != req.WorkflowID || task.WorkflowStepID != req.WorkflowStepID {
return nil, false, nil
}
if code, message := h.validateSameStepMove(ctx, req, task); code != "" {
return moveTaskErrorResponse(msg, code, message)
}
response, err := ws.NewResponse(msg.ID, msg.Action, dto.MoveTaskResponse{
Task: dto.FromTask(task),
Disposition: moveDispositionApplied,
})
return response, true, err
}
func (h *Handlers) validateSameStepMove(ctx context.Context, req moveTaskRequest, task *models.Task) (string, string) {
if err := h.taskSvc.AuthorizeWorkspaceScope(ctx, task.WorkspaceID, authz.ScopeTaskWrite); err != nil {
if service.IsForbidden(err) {
return ws.ErrorCodeForbidden, "task is not writable"
}
h.logger.Error("move_task: failed to authorize task for same-step completion",
zap.String("task_id", req.TaskID), zap.Error(err))
return ws.ErrorCodeInternalError, "failed to authorize task for move"
}
if task.ArchivedAt != nil {
return ws.ErrorCodeConflict, "archived tasks cannot be moved"
}
if code, message := h.validateSameStepMoveTargets(ctx, req, task); code != "" {
return code, message
}
if err := workflowmove.ValidateEntryOptions(req.EntryOptions, workflowmove.MoveChangePositionOnly); err != nil {
return ws.ErrorCodeValidation, err.Error()
}
return "", ""
}
func (h *Handlers) validateSameStepMoveTargets(
ctx context.Context,
req moveTaskRequest,
task *models.Task,
) (string, string) {
if code, message := h.validateSameStepTargetWorkflow(ctx, req, task); code != "" {
return code, message
}
return h.validateSameStepTargetStep(ctx, req)
}
func (h *Handlers) validateSameStepTargetWorkflow(
ctx context.Context,
req moveTaskRequest,
task *models.Task,
) (string, string) {
targetWorkflow, err := h.taskSvc.GetWorkflow(ctx, req.WorkflowID)
if err != nil {
h.logger.Error("move_task: failed to look up target workflow",
zap.String("task_id", req.TaskID), zap.String("workflow_id", req.WorkflowID), zap.Error(err))
if errors.Is(err, repoerrors.ErrWorkflowNotFound) || errors.Is(err, workflowservice.ErrNotVisible) {
return ws.ErrorCodeValidation, "target workflow_id does not exist"
}
return ws.ErrorCodeInternalError, "failed to validate target workflow"
}
if targetWorkflow == nil {
return ws.ErrorCodeValidation, "target workflow_id does not exist"
}
if targetWorkflow.WorkspaceID != task.WorkspaceID {
return ws.ErrorCodeValidation, "target workflow is in a different workspace"
}
return "", ""
}
func (h *Handlers) validateSameStepTargetStep(ctx context.Context, req moveTaskRequest) (string, string) {
if h.workflowCtrl == nil {
return ws.ErrorCodeInternalError, "workflow validation is not configured"
}
stepResponse, err := h.workflowCtrl.GetStep(ctx, req.WorkflowStepID)
if err != nil {
h.logger.Error("move_task: failed to look up target workflow step",
zap.String("task_id", req.TaskID), zap.String("workflow_step_id", req.WorkflowStepID), zap.Error(err))
if errors.Is(err, workflowservice.ErrNotVisible) || errors.Is(err, wfmodels.ErrWorkflowStepNotFound) {
return ws.ErrorCodeValidation, "target workflow_step_id does not exist"
}
return ws.ErrorCodeInternalError, "failed to validate target workflow step"
}
if stepResponse == nil || stepResponse.Step == nil {
return ws.ErrorCodeValidation, "target workflow_step_id does not exist"
}
if stepResponse.Step.WorkflowID != req.WorkflowID {
return ws.ErrorCodeValidation, "target workflow_step_id does not belong to the requested workflow_id"
}
return "", ""
}
func moveTaskErrorResponse(msg *ws.Message, code, message string) (*ws.Message, bool, error) {
response, err := ws.NewError(msg.ID, msg.Action, code, message, nil)
return response, true, err
}
// deferMoveTask records a PendingMove for the agent's turn-end handler to
// apply. Optionally queues a hand-off prompt when the caller supplied one.
// Returns a synthetic moved-task DTO so the agent's tool call resolves
@@ -0,0 +1,390 @@
package handlers
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/jmoiron/sqlx"
"github.com/kandev/kandev/internal/auth/authn"
"github.com/kandev/kandev/internal/authz"
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
"github.com/kandev/kandev/internal/task/dto"
"github.com/kandev/kandev/internal/task/models"
sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
"github.com/kandev/kandev/internal/task/service"
workflowcontroller "github.com/kandev/kandev/internal/workflow/controller"
workflowmodels "github.com/kandev/kandev/internal/workflow/models"
workflowmove "github.com/kandev/kandev/internal/workflow/move"
workflowrepo "github.com/kandev/kandev/internal/workflow/repository"
v1 "github.com/kandev/kandev/pkg/api/v1"
ws "github.com/kandev/kandev/pkg/websocket"
)
type sameStepMoveSession struct {
id string
state models.TaskSessionState
primary bool
}
type sameStepMoveFixture struct {
svc *service.Service
repo *sqliterepo.Repository
workflowCtrl *workflowcontroller.Controller
workflowRepo *workflowrepo.Repository
db *sqlx.DB
eventBus *bus.MemoryEventBus
taskID string
workspaceID string
workflowID string
stepID string
}
func newSameStepMoveFixture(t *testing.T, sessions ...sameStepMoveSession) *sameStepMoveFixture {
t.Helper()
svc, repo, workflowCtrl, workflowRepo, db, eventBus := newTestTaskServiceWithWorkflowDBAndEventBus(t)
fixture := &sameStepMoveFixture{
svc: svc,
repo: repo,
workflowCtrl: workflowCtrl,
workflowRepo: workflowRepo,
db: db,
eventBus: eventBus,
taskID: "task-same-step",
workspaceID: "ws-same-step",
workflowID: "wf-same-step",
stepID: "step-work",
}
ctx := context.Background()
now := time.Now().UTC()
require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{
ID: fixture.workspaceID, Name: "Same step", OwnerID: "task-owner", CreatedAt: now, UpdatedAt: now,
}))
require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{
ID: fixture.workflowID, WorkspaceID: fixture.workspaceID, Name: "Board", CreatedAt: now, UpdatedAt: now,
}))
require.NoError(t, workflowRepo.CreateStep(ctx, &workflowmodels.WorkflowStep{
ID: fixture.stepID, WorkflowID: fixture.workflowID, Name: "Work", Position: 0,
CreatedAt: now, UpdatedAt: now,
}))
require.NoError(t, repo.CreateTask(ctx, &models.Task{
ID: fixture.taskID, WorkspaceID: fixture.workspaceID, WorkflowID: fixture.workflowID,
WorkflowStepID: fixture.stepID, Position: 7, Title: "Task", State: v1.TaskStateInProgress,
Metadata: map[string]interface{}{"same_step_test": "preserve"}, CreatedAt: now, UpdatedAt: now,
}))
_, err := db.Exec(db.Rebind(`UPDATE tasks SET position = ? WHERE id = ?`), 7, fixture.taskID)
require.NoError(t, err)
for _, session := range sessions {
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{
ID: session.id, TaskID: fixture.taskID, State: session.state, IsPrimary: session.primary,
Metadata: map[string]interface{}{"session_test": session.id}, StartedAt: now, UpdatedAt: now,
}))
}
return fixture
}
func (f *sameStepMoveFixture) handler(t *testing.T, queue MessageQueuer) *Handlers {
return &Handlers{
taskSvc: f.svc,
workflowCtrl: f.workflowCtrl,
messageQueue: queue,
logger: testLogger(t),
}
}
func (f *sameStepMoveFixture) message(t *testing.T, fields map[string]interface{}) *ws.Message {
t.Helper()
payload := map[string]interface{}{
"task_id": f.taskID, "workflow_id": f.workflowID, "workflow_step_id": f.stepID,
}
for key, value := range fields {
payload[key] = value
}
return makeWSMessage(t, ws.ActionMCPMoveTask, payload)
}
func decodeMoveTaskResponse(t *testing.T, response *ws.Message) dto.MoveTaskResponse {
t.Helper()
require.Equal(t, ws.MessageTypeResponse, response.Type, "payload: %s", string(response.Payload))
var result dto.MoveTaskResponse
require.NoError(t, json.Unmarshal(response.Payload, &result))
return result
}
func transitionCount(t *testing.T, db *sqlx.DB, taskID string) int {
t.Helper()
var count int
require.NoError(t, db.Get(&count, db.Rebind(`SELECT COUNT(*) FROM task_step_transitions WHERE task_id = ?`), taskID))
return count
}
// @covers AC-TASKS-MCP-MOVE-RESULTS-001.1
func TestHandleMoveTask_SameStepApplied(t *testing.T) {
cases := []struct {
name string
sessions []sameStepMoveSession
}{
{name: "running primary", sessions: []sameStepMoveSession{{id: "session-running", state: models.TaskSessionStateRunning, primary: true}}},
{name: "starting primary", sessions: []sameStepMoveSession{{id: "session-starting", state: models.TaskSessionStateStarting, primary: true}}},
{name: "idle primary", sessions: []sameStepMoveSession{{id: "session-idle", state: models.TaskSessionStateIdle, primary: true}}},
{name: "no session"},
{name: "mixed sibling sessions", sessions: []sameStepMoveSession{
{id: "session-primary-idle", state: models.TaskSessionStateIdle, primary: true},
{id: "session-sibling-running", state: models.TaskSessionStateRunning},
}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
fixture := newSameStepMoveFixture(t, tc.sessions...)
response, err := fixture.handler(t, nil).handleMoveTask(context.Background(), fixture.message(t, nil))
require.NoError(t, err)
result := decodeMoveTaskResponse(t, response)
assert.Equal(t, moveDispositionApplied, result.Disposition)
assert.Equal(t, fixture.taskID, result.Task.ID)
assert.Equal(t, fixture.stepID, result.Task.WorkflowStepID)
assert.Equal(t, 7, result.Task.Position)
assert.Empty(t, result.MoveID)
assert.Nil(t, result.EntryOptions)
assert.Empty(t, result.WorkflowEntryIdentity)
})
}
}
// @covers AC-TASKS-MCP-MOVE-RESULTS-001.2
func TestHandleMoveTask_SameStepReturnsStoredTask(t *testing.T) {
fixture := newSameStepMoveFixture(t)
for _, tc := range []struct {
name string
fields map[string]interface{}
}{
{name: "position omitted"},
{name: "requested position ignored", fields: map[string]interface{}{"position": 99}},
{name: "empty entry options normalized", fields: map[string]interface{}{"entry_options": map[string]interface{}{}}},
} {
t.Run(tc.name, func(t *testing.T) {
response, err := fixture.handler(t, nil).handleMoveTask(context.Background(), fixture.message(t, tc.fields))
require.NoError(t, err)
result := decodeMoveTaskResponse(t, response)
assert.Equal(t, 7, result.Task.Position)
assert.Equal(t, map[string]interface{}{"same_step_test": "preserve"}, result.Task.Metadata)
assert.Empty(t, result.MoveID)
assert.Nil(t, result.EntryOptions)
})
}
}
// @covers AC-TASKS-MCP-MOVE-RESULTS-001.3
func TestHandleMoveTask_SameStepPreservesEffects(t *testing.T) {
fixture := newSameStepMoveFixture(t, sameStepMoveSession{
id: "session-running", state: models.TaskSessionStateRunning, primary: true,
})
ctx := context.Background()
queue := &pendingMoveRecordingQueuer{
recordingMessageQueuer: recordingMessageQueuer{calls: []messagequeue.QueuedMessage{{
ID: "existing-prompt", SessionID: "other-session", TaskID: "other-task", Content: "keep this prompt",
}}},
pendingSessionID: "other-session",
pendingMoves: []messagequeue.PendingMove{{
MoveID: "existing-move", TaskID: "other-task", WorkflowID: "other-workflow", WorkflowStepID: "other-step", Position: 4,
}},
}
beforeMessages := append([]messagequeue.QueuedMessage(nil), queue.calls...)
beforeMoves := append([]messagequeue.PendingMove(nil), queue.pendingMoves...)
beforeTask, err := fixture.svc.GetTask(ctx, fixture.taskID)
require.NoError(t, err)
beforeSession, err := fixture.repo.GetTaskSession(ctx, "session-running")
require.NoError(t, err)
beforeTransitions := transitionCount(t, fixture.db, fixture.taskID)
eventCount := 0
subscription, err := fixture.eventBus.Subscribe(">", func(context.Context, *bus.Event) error {
eventCount++
return nil
})
require.NoError(t, err)
t.Cleanup(func() { _ = subscription.Unsubscribe() })
handler := fixture.handler(t, queue)
for range 47 {
response, err := handler.handleMoveTask(ctx, fixture.message(t, map[string]interface{}{
"entry_options": map[string]interface{}{},
}))
require.NoError(t, err)
result := decodeMoveTaskResponse(t, response)
assert.Equal(t, moveDispositionApplied, result.Disposition)
}
afterTask, err := fixture.svc.GetTask(ctx, fixture.taskID)
require.NoError(t, err)
afterSession, err := fixture.repo.GetTaskSession(ctx, "session-running")
require.NoError(t, err)
assert.Equal(t, beforeTask, afterTask)
assert.Equal(t, beforeSession, afterSession)
assert.Equal(t, beforeMessages, queue.calls)
assert.Equal(t, beforeMoves, queue.pendingMoves)
assert.Equal(t, beforeTransitions, transitionCount(t, fixture.db, fixture.taskID))
assert.Zero(t, eventCount)
}
// @covers AC-TASKS-MCP-MOVE-RESULTS-001.3
func TestHandleMoveTask_SameStepPersistentQueue(t *testing.T) {
fixture := newSameStepMoveFixture(t, sameStepMoveSession{
id: "session-running", state: models.TaskSessionStateRunning, primary: true,
})
ctx := context.Background()
queueRepo, err := messagequeue.NewSQLiteRepository(fixture.db, fixture.db)
require.NoError(t, err)
queue := messagequeue.NewService(queueRepo, 0, testLogger(t))
now := time.Now().UTC()
require.NoError(t, fixture.repo.CreateTask(ctx, &models.Task{
ID: "task-unrelated", WorkspaceID: fixture.workspaceID, WorkflowID: fixture.workflowID,
WorkflowStepID: fixture.stepID, Position: 2, Title: "Unrelated task", State: v1.TaskStateInProgress,
CreatedAt: now, UpdatedAt: now,
}))
require.NoError(t, fixture.repo.CreateTaskSession(ctx, &models.TaskSession{
ID: "session-unrelated", TaskID: "task-unrelated", IsPrimary: true,
State: models.TaskSessionStateRunning, StartedAt: now, UpdatedAt: now,
}))
require.NoError(t, queueRepo.SetPendingMove(ctx, "session-unrelated", &messagequeue.PendingMove{
MoveID: "unrelated-move", TaskID: "task-unrelated", WorkflowID: fixture.workflowID,
WorkflowStepID: "step-next", Position: 3, Actor: "agent", SenderSessionID: "sender-session",
EntryOptions: &workflowmove.EntryOptions{Instructions: "retain this override"},
}))
beforeMoves, err := queueRepo.ListPendingMoves(ctx)
require.NoError(t, err)
require.Len(t, beforeMoves, 1)
beforeTransitions := transitionCount(t, fixture.db, fixture.taskID)
dispatcher := ws.NewDispatcher()
fixture.handler(t, queue).RegisterHandlers(dispatcher)
for range 47 {
response, err := dispatcher.Dispatch(ctx, fixture.message(t, nil))
require.NoError(t, err)
result := decodeMoveTaskResponse(t, response)
assert.Equal(t, moveDispositionApplied, result.Disposition)
}
afterMoves, err := queueRepo.ListPendingMoves(ctx)
require.NoError(t, err)
assert.Equal(t, beforeMoves, afterMoves)
currentMove, err := queueRepo.GetPendingMove(ctx, "session-running")
require.NoError(t, err)
assert.Nil(t, currentMove)
task, err := fixture.svc.GetTask(ctx, fixture.taskID)
require.NoError(t, err)
assert.Equal(t, fixture.stepID, task.WorkflowStepID)
assert.Equal(t, 7, task.Position)
assert.Equal(t, beforeTransitions, transitionCount(t, fixture.db, fixture.taskID))
}
// @covers AC-TASKS-MCP-MOVE-RESULTS-001.4
func TestHandleMoveTask_SameStepValidation(t *testing.T) {
type testCase struct {
name string
fields map[string]interface{}
prepare func(*testing.T, *sameStepMoveFixture) context.Context
withoutTask bool
withoutTarget bool
wantCode string
}
cases := []testCase{
{name: "task read failure", fields: map[string]interface{}{"task_id": "missing-task"}, wantCode: ws.ErrorCodeInternalError},
{name: "task write denied", prepare: func(t *testing.T, f *sameStepMoveFixture) context.Context {
require.NoError(t, f.repo.UpsertWorkspaceMember(context.Background(), &models.WorkspaceMember{
WorkspaceID: f.workspaceID, UserID: "viewer", Role: string(authz.WorkspaceRoleViewer),
}))
return authn.WithIdentity(context.Background(), authn.Identity{UserID: "viewer", Role: authn.RoleMember})
}, wantCode: ws.ErrorCodeForbidden},
{name: "task outside caller scope", prepare: func(*testing.T, *sameStepMoveFixture) context.Context {
return authn.WithIdentity(context.Background(), authn.Identity{UserID: "outsider", Role: authn.RoleMember})
}, wantCode: ws.ErrorCodeInternalError},
{name: "archived task", prepare: func(t *testing.T, f *sameStepMoveFixture) context.Context {
require.NoError(t, f.repo.ArchiveTask(context.Background(), f.taskID))
return context.Background()
}, wantCode: ws.ErrorCodeConflict},
{name: "missing step", prepare: func(t *testing.T, f *sameStepMoveFixture) context.Context {
require.NoError(t, f.workflowRepo.DeleteStep(context.Background(), f.stepID))
return context.Background()
}, wantCode: ws.ErrorCodeValidation},
{name: "step belongs to another workflow", prepare: func(t *testing.T, f *sameStepMoveFixture) context.Context {
ctx := context.Background()
require.NoError(t, f.repo.CreateWorkflow(ctx, &models.Workflow{
ID: "wf-other", WorkspaceID: f.workspaceID, Name: "Other", CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(),
}))
require.NoError(t, f.workflowRepo.DeleteStep(ctx, f.stepID))
require.NoError(t, f.workflowRepo.CreateStep(ctx, &workflowmodels.WorkflowStep{
ID: f.stepID, WorkflowID: "wf-other", Name: "Foreign step", Position: 0,
CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(),
}))
return ctx
}, wantCode: ws.ErrorCodeValidation},
{name: "missing workflow", fields: map[string]interface{}{"workflow_id": "missing-workflow"}, prepare: func(t *testing.T, f *sameStepMoveFixture) context.Context {
_, err := f.db.Exec(f.db.Rebind(`UPDATE tasks SET workflow_id = ? WHERE id = ?`), "missing-workflow", f.taskID)
require.NoError(t, err)
return context.Background()
}, wantCode: ws.ErrorCodeValidation},
{name: "workflow lookup database failure", prepare: func(t *testing.T, f *sameStepMoveFixture) context.Context {
_, err := f.db.Exec(`ALTER TABLE workflows RENAME TO workflows_unavailable`)
require.NoError(t, err)
return context.Background()
}, wantCode: ws.ErrorCodeInternalError},
{name: "workflow belongs to another workspace", prepare: func(t *testing.T, f *sameStepMoveFixture) context.Context {
ctx := context.Background()
require.NoError(t, f.repo.CreateWorkspace(ctx, &models.Workspace{
ID: "ws-other", Name: "Other", CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(),
}))
_, err := f.db.Exec(f.db.Rebind(`UPDATE workflows SET workspace_id = ? WHERE id = ?`), "ws-other", f.workflowID)
require.NoError(t, err)
return ctx
}, wantCode: ws.ErrorCodeValidation},
{name: "task service missing", withoutTask: true, wantCode: ws.ErrorCodeInternalError},
{name: "workflow validation missing", withoutTarget: true, wantCode: ws.ErrorCodeInternalError},
{name: "instructions require a step change", fields: map[string]interface{}{"entry_options": map[string]interface{}{"instructions": "handoff"}}, wantCode: ws.ErrorCodeValidation},
{name: "reset requires a step change", fields: map[string]interface{}{"entry_options": map[string]interface{}{"reset_context": true}}, wantCode: ws.ErrorCodeValidation},
{name: "skip prompt requires a step change", fields: map[string]interface{}{"entry_options": map[string]interface{}{"skip_step_prompt": true}}, wantCode: ws.ErrorCodeValidation},
{name: "legacy prompt requires a step change", fields: map[string]interface{}{"prompt": "handoff"}, wantCode: ws.ErrorCodeValidation},
{name: "conflicting prompt aliases", fields: map[string]interface{}{
"prompt": "legacy", "entry_options": map[string]interface{}{"instructions": "nested"},
}, wantCode: ws.ErrorCodeValidation},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
fixture := newSameStepMoveFixture(t)
ctx := context.Background()
if tc.prepare != nil {
ctx = tc.prepare(t, fixture)
}
handler := fixture.handler(t, nil)
if tc.withoutTask {
handler.taskSvc = nil
}
if tc.withoutTarget {
handler.workflowCtrl = nil
}
response, err := handler.handleMoveTask(ctx, fixture.message(t, tc.fields))
require.NoError(t, err)
assertWSError(t, response, tc.wantCode)
if tc.wantCode == ws.ErrorCodeInternalError {
assert.NotContains(t, string(response.Payload), "no such table")
}
})
}
}
func TestHandleMoveTask_SameStepStepLookupDatabaseFailure(t *testing.T) {
fixture := newSameStepMoveFixture(t)
_, _, failedStepController, _, failedDB := newTestTaskServiceWithWorkflowDB(t)
_, err := failedDB.Exec(`ALTER TABLE workflow_steps RENAME TO workflow_steps_unavailable`)
require.NoError(t, err)
handler := fixture.handler(t, nil)
handler.workflowCtrl = failedStepController
response, err := handler.handleMoveTask(context.Background(), fixture.message(t, nil))
require.NoError(t, err)
assertWSError(t, response, ws.ErrorCodeInternalError)
assert.NotContains(t, string(response.Payload), "no such table")
}
@@ -124,6 +124,18 @@ func newTestTaskServiceWithWorkflow(t *testing.T) (*service.Service, *sqliterepo
// without disturbing the rest of the fixture's tables.
func newTestTaskServiceWithWorkflowDB(t *testing.T) (
*service.Service, *sqliterepo.Repository, *workflowcontroller.Controller, *workflowrepo.Repository, *sqlx.DB,
) {
svc, repo, workflowCtrl, workflowRepo, sqlxDB, _ := newTestTaskServiceWithWorkflowDBAndEventBus(t)
return svc, repo, workflowCtrl, workflowRepo, sqlxDB
}
func newTestTaskServiceWithWorkflowDBAndEventBus(t *testing.T) (
*service.Service,
*sqliterepo.Repository,
*workflowcontroller.Controller,
*workflowrepo.Repository,
*sqlx.DB,
*bus.MemoryEventBus,
) {
t.Helper()
dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "test.db"))
@@ -163,7 +175,7 @@ func newTestTaskServiceWithWorkflowDB(t *testing.T) (
svc.SetWorkspacePolicyAttacher(testWorkspacePolicyAttacher{})
workflowSvc := workflowservice.NewService(workflowRepo, log)
t.Cleanup(func() { _ = workflowSvc.Close() })
return svc, repo, workflowcontroller.NewController(workflowSvc), workflowRepo, sqlxDB
return svc, repo, workflowcontroller.NewController(workflowSvc), workflowRepo, sqlxDB, eventBus
}
func TestHandleListWorkspacesAutomationIsScopedToPrincipalWorkspace(t *testing.T) {
@@ -362,12 +362,12 @@ func (s *Server) registerConfigTaskTools() {
)
s.mcpServer.AddTool(
mcp.NewTool("move_task_kandev",
mcp.WithDescription(`Move a task to a different workflow step. When the source session is mid-turn (RUNNING), the move is deferred to turn-end automatically — prompt is optional (use it for cross-agent hand-offs). Idle-session and admin moves apply immediately. Returns a move-result envelope: "disposition" is "applied" when the move committed immediately or "deferred" when it was recorded for turn-end, "task" is the moved (or target-step) task, and an optioned move also returns "move_id" and the accepted "entry_options" so you can correlate the one-shot override with the eventual entry.`),
mcp.WithDescription(`Move a task to a different workflow step. A request naming the task's current workflow and step, without entry options, returns "applied" with the stored task because the destination already holds; no retry is needed. Only an actual step change is deferred to turn-end while the source session is RUNNING or STARTING. Idle-session and admin step changes apply immediately. Returns a move-result envelope: "disposition" is "applied" for a completed move or current-step no-op, or "deferred" when a step change was recorded for turn-end; "task" is the stored or moved task. An optioned step change also returns "move_id" and the accepted "entry_options" so you can correlate the one-shot override with the eventual entry.`),
mcp.WithString("task_id", mcp.Required(), mcp.Description("The task ID")),
mcp.WithString("workflow_id", mcp.Required(), mcp.Description("Target workflow ID")),
mcp.WithString("workflow_step_id", mcp.Required(), mcp.Description("Target workflow step ID")),
mcp.WithNumber("position", mcp.Description("Position within the step (0-based)")),
mcp.WithString("prompt", mcp.Description("Legacy alias for entry_options.instructions. Optional hand-off message applied once when the task enters the new step. Supplying both this and entry_options.instructions is rejected. Omit for self-moves like Work → Done.")),
mcp.WithNumber("position", mcp.Description("Requested position (0-based); the server determines arrival order, and a current-step request preserves the stored position.")),
mcp.WithString("prompt", mcp.Description("Legacy alias for entry_options.instructions. Optional hand-off message applied once for a new step. Supplying both this and entry_options.instructions is rejected; non-empty prompts are invalid for a current-step request.")),
moveTaskEntryOptionsToolOption(),
),
s.wrapHandler("move_task_kandev", s.moveTaskHandler()),
@@ -97,6 +97,49 @@ func TestMoveTaskToolSchemasExposeEntryOptions(t *testing.T) {
}
}
func TestMoveTaskSameStepDescriptionsAndResponseForwarding(t *testing.T) {
for name, backend := range map[string]*testBackend{
"task": {response: map[string]interface{}{"disposition": "applied", "task": map[string]interface{}{"id": "task-1", "workflow_step_id": "step-work", "position": 7}}},
"config": {response: map[string]interface{}{"disposition": "applied", "task": map[string]interface{}{"id": "task-1", "workflow_step_id": "step-work", "position": 7}}},
} {
var server *Server
if name == "task" {
server = newTaskModeServer(t, backend, "task-current")
} else {
server = newTestServer(t, backend)
}
tool := server.mcpServer.ListTools()["move_task_kandev"]
assert.Contains(t, tool.Tool.Description, "current workflow and step")
assert.Contains(t, tool.Tool.Description, "no retry is needed")
assert.Contains(t, tool.Tool.Description, "Only an actual step change")
position, ok := toolInputProperties(t, server, "move_task_kandev")["position"].(map[string]interface{})
require.True(t, ok)
assert.Contains(t, position["description"], "server determines arrival order")
assert.Contains(t, position["description"], "preserves the stored position")
result := callTool(t, server, "move_task_kandev", map[string]interface{}{
"task_id": "task-1", "workflow_id": "wf-1", "workflow_step_id": "step-work", "position": 99,
})
require.False(t, result.IsError)
require.Len(t, result.Content, 1)
content, ok := result.Content[0].(mcplib.TextContent)
require.True(t, ok)
var response map[string]interface{}
require.NoError(t, json.Unmarshal([]byte(content.Text), &response))
assert.Equal(t, "applied", response["disposition"])
task, ok := response["task"].(map[string]interface{})
require.True(t, ok)
assert.Equal(t, float64(7), task["position"])
assert.Equal(t, ws.ActionMCPMoveTask, backend.lastAction)
payload, ok := backend.lastPayload.(map[string]interface{})
require.True(t, ok)
assert.Equal(t, "task-1", payload["task_id"])
assert.Equal(t, "wf-1", payload["workflow_id"])
assert.Equal(t, "step-work", payload["workflow_step_id"])
}
}
// --- Action constant tests ---
func TestActionConstants_MatchWebSocketActions(t *testing.T) {
+3 -3
View File
@@ -1269,12 +1269,12 @@ func (s *Server) registerKanbanTools() {
)
s.mcpServer.AddTool(
mcp.NewTool("move_task_kandev",
mcp.WithDescription(`Move a task to a different workflow step. When the source session is mid-turn (RUNNING), the move is deferred to turn-end automatically — prompt is optional (use it for cross-agent hand-offs). Idle-session and admin moves apply immediately. Returns a move-result envelope: "disposition" is "applied" when the move committed immediately or "deferred" when it was recorded for turn-end, "task" is the moved (or target-step) task, and an optioned move also returns "move_id" and the accepted "entry_options" so you can correlate the one-shot override with the eventual entry.`),
mcp.WithDescription(`Move a task to a different workflow step. A request naming the task's current workflow and step, without entry options, returns "applied" with the stored task because the destination already holds; no retry is needed. Only an actual step change is deferred to turn-end while the source session is RUNNING or STARTING. Idle-session and admin step changes apply immediately. Returns a move-result envelope: "disposition" is "applied" for a completed move or current-step no-op, or "deferred" when a step change was recorded for turn-end; "task" is the stored or moved task. An optioned step change also returns "move_id" and the accepted "entry_options" so you can correlate the one-shot override with the eventual entry.`),
mcp.WithString("task_id", mcp.Required(), mcp.Description("The task ID")),
mcp.WithString("workflow_id", mcp.Required(), mcp.Description("Target workflow ID")),
mcp.WithString("workflow_step_id", mcp.Required(), mcp.Description("Target workflow step ID")),
mcp.WithNumber("position", mcp.Description("Position within the step (0-based)")),
mcp.WithString("prompt", mcp.Description("Optional hand-off message for the receiving agent at the new step. Mid-turn moves are always deferred; include a prompt when the next agent needs context (e.g. QA → review). Omit for self-moves like Work → Done.")),
mcp.WithNumber("position", mcp.Description("Requested position (0-based); the server determines arrival order, and a current-step request preserves the stored position.")),
mcp.WithString("prompt", mcp.Description("Optional hand-off message for the receiving agent at a new step. Only actual step changes are deferred during an active turn; non-empty prompts are invalid for a current-step request.")),
moveTaskEntryOptionsToolOption(),
),
s.wrapHandler("move_task_kandev", s.moveTaskHandler()),
@@ -9,6 +9,9 @@ var ErrWorkspaceNameMismatch = errors.New("workspace name mismatch")
// ErrWorkspaceNotFound reports that no workspace row matched the supplied id.
var ErrWorkspaceNotFound = errors.New("workspace not found")
// ErrWorkflowNotFound reports that no workflow row matched the supplied id.
var ErrWorkflowNotFound = errors.New("workflow not found")
// ErrTaskNotFound reports that no task row matched the supplied id.
var ErrTaskNotFound = errors.New("task not found")
@@ -12,6 +12,7 @@ import (
"github.com/kandev/kandev/internal/db/dialect"
"github.com/kandev/kandev/internal/steptelemetry"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/repository/repoerrors"
)
// AddTaskToWorkflow adds a task to a workflow with placement. Wrapped in a
@@ -242,7 +243,7 @@ func (r *Repository) GetWorkflow(ctx context.Context, id string) (*models.Workfl
FROM workflows WHERE id = ?
`, workflowSelectColumns)), id))
if err == sql.ErrNoRows {
return nil, fmt.Errorf("workflow not found: %s", id)
return nil, fmt.Errorf("%w: %s", repoerrors.ErrWorkflowNotFound, id)
}
if err != nil {
return nil, err
@@ -2,9 +2,11 @@ package repository
import (
"context"
"errors"
"testing"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/repository/repoerrors"
)
// Workflow CRUD tests
@@ -142,6 +144,8 @@ func TestSQLiteRepository_WorkflowNotFound(t *testing.T) {
_, err := repo.GetWorkflow(ctx, "nonexistent")
if err == nil {
t.Error("expected error for nonexistent workflow")
} else if !errors.Is(err, repoerrors.ErrWorkflowNotFound) {
t.Errorf("GetWorkflow error = %v, want ErrWorkflowNotFound", err)
}
err = repo.UpdateWorkflow(ctx, &models.Workflow{ID: "nonexistent", Name: "Test"})
@@ -8,6 +8,9 @@ import (
// ErrWIPLimitExceeded identifies a workflow-step capacity rejection.
var ErrWIPLimitExceeded = errors.New("workflow step WIP limit exceeded")
// ErrWorkflowStepNotFound reports that no workflow step row matched the supplied id.
var ErrWorkflowStepNotFound = errors.New("workflow step not found")
// WIPLimitError carries workflow-step capacity details while preserving
// errors.Is(err, ErrWIPLimitExceeded) classification at service boundaries.
type WIPLimitError struct {
@@ -770,7 +770,7 @@ func (r *Repository) GetStep(ctx context.Context, id string) (*models.WorkflowSt
step, err := r.scanStep(row)
if err == sql.ErrNoRows {
return nil, fmt.Errorf("workflow step not found: %s", id)
return nil, fmt.Errorf("%w: %s", models.ErrWorkflowStepNotFound, id)
}
if err != nil {
return nil, err
@@ -3,6 +3,7 @@ package repository
import (
"context"
"database/sql"
"errors"
"reflect"
"testing"
@@ -12,6 +13,14 @@ import (
"github.com/kandev/kandev/internal/workflow/models"
)
func TestGetMissingWorkflowStepReturnsTypedNotFound(t *testing.T) {
repo := setupTestRepo(t)
_, err := repo.GetStep(context.Background(), "missing-step")
if !errors.Is(err, models.ErrWorkflowStepNotFound) {
t.Fatalf("GetStep error = %v, want ErrWorkflowStepNotFound", err)
}
}
func setupTestRepo(t *testing.T) *Repository {
repo, _ := setupTestRepoWithDB(t)
return repo
@@ -9,6 +9,7 @@ import (
"github.com/kandev/kandev/internal/auth/authn"
taskmodels "github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/repository/repoerrors"
workflowmodels "github.com/kandev/kandev/internal/workflow/models"
)
// Per-user scoping for the workflow-step surface (opt-in authentication).
@@ -205,9 +206,11 @@ func isNotFoundError(err error) bool {
return false
}
if errors.Is(err, ErrNotVisible) ||
errors.Is(err, repoerrors.ErrWorkflowNotFound) ||
errors.Is(err, repoerrors.ErrWorkspaceNotFound) ||
errors.Is(err, repoerrors.ErrTaskNotFound) ||
errors.Is(err, taskmodels.ErrTaskSessionNotFound) {
errors.Is(err, taskmodels.ErrTaskSessionNotFound) ||
errors.Is(err, workflowmodels.ErrWorkflowStepNotFound) {
return true
}
return strings.Contains(strings.ToLower(err.Error()), "not found")
@@ -262,10 +262,12 @@ func (o opaqueWrap) Unwrap() error { return o.err }
func TestIsNotFoundMatchesSentinelsWithoutRelyingOnWording(t *testing.T) {
for name, sentinel := range map[string]error{
"not visible": ErrNotVisible,
"workspace": repoerrors.ErrWorkspaceNotFound,
"task": repoerrors.ErrTaskNotFound,
"task session": taskmodels.ErrTaskSessionNotFound,
"not visible": ErrNotVisible,
"workflow": repoerrors.ErrWorkflowNotFound,
"workspace": repoerrors.ErrWorkspaceNotFound,
"task": repoerrors.ErrTaskNotFound,
"task session": taskmodels.ErrTaskSessionNotFound,
"workflow step": models.ErrWorkflowStepNotFound,
} {
t.Run(name, func(t *testing.T) {
require.True(t, IsNotFound(sentinel))