From 00a6567402a623c0a5d0d8c616cd6ad1459b8a40 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Thu, 30 Apr 2026 22:53:13 +0100 Subject: [PATCH] feat(mcp): add message_task_kandev tool (#745) --- .../backend/internal/mcp/handlers/handlers.go | 144 +++++++- .../internal/mcp/handlers/handlers_test.go | 17 + .../mcp/handlers/message_task_test.go | 330 ++++++++++++++++++ apps/backend/internal/mcp/server/handlers.go | 23 ++ .../internal/mcp/server/handlers_test.go | 46 +++ apps/backend/internal/mcp/server/server.go | 20 +- .../internal/mcp/server/server_test.go | 8 +- .../task/handlers/message_handlers.go | 41 +-- .../internal/task/service/service_sessions.go | 45 +++ apps/backend/pkg/websocket/actions.go | 1 + 10 files changed, 632 insertions(+), 43 deletions(-) create mode 100644 apps/backend/internal/mcp/handlers/message_task_test.go diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index 41b04b8a6f..8601c2333e 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -17,6 +17,7 @@ import ( "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator" + "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/task/dto" "github.com/kandev/kandev/internal/task/models" @@ -56,9 +57,14 @@ type EventBus interface { Publish(ctx context.Context, topic string, event *bus.Event) error } -// SessionLauncher provides session launch capability for auto-starting tasks. +// SessionLauncher provides session launch and prompt-dispatch capabilities. +// Both methods are implemented by *orchestrator.Service. type SessionLauncher interface { LaunchSession(ctx context.Context, req *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) + PromptTask(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool, attachments []v1.MessageAttachment) (*orchestrator.PromptResult, error) + StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode bool, attachments []v1.MessageAttachment) (*executor.TaskExecution, error) + ResumeTaskSession(ctx context.Context, taskID, sessionID string) (*executor.TaskExecution, error) + GetMessageQueue() *messagequeue.Service } // MessageQueuer queues a prompt message for delivery to a session on its next turn. @@ -141,13 +147,14 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionMCPListTasks, h.handleListTasks) d.RegisterFunc(ws.ActionMCPCreateTask, h.handleCreateTask) d.RegisterFunc(ws.ActionMCPUpdateTask, h.handleUpdateTask) + d.RegisterFunc(ws.ActionMCPMessageTask, h.handleMessageTask) d.RegisterFunc(ws.ActionMCPAskUserQuestion, h.handleAskUserQuestion) d.RegisterFunc(ws.ActionMCPCreateTaskPlan, h.handleCreateTaskPlan) d.RegisterFunc(ws.ActionMCPGetTaskPlan, h.handleGetTaskPlan) d.RegisterFunc(ws.ActionMCPUpdateTaskPlan, h.handleUpdateTaskPlan) d.RegisterFunc(ws.ActionMCPDeleteTaskPlan, h.handleDeleteTaskPlan) d.RegisterFunc(ws.ActionMCPClarificationTimeout, h.handleClarificationTimeout) - count := 12 + count := 13 // Config-mode handlers (registered when config deps are set) if h.workflowSvc != nil { @@ -601,6 +608,139 @@ func (h *Handlers) handleUpdateTask(ctx context.Context, msg *ws.Message) (*ws.M return ws.NewResponse(msg.ID, msg.Action, dto.FromTask(task)) } +// handleMessageTask sends a prompt to an existing task. The dispatch path depends +// on the primary session's state: RUNNING/STARTING messages are queued and drained +// at turn end; idle sessions are prompted directly; CREATED sessions are started. +func (h *Handlers) handleMessageTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + TaskID string `json:"task_id"` + Prompt string `json:"prompt"` + } + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + if req.Prompt == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "prompt is required", nil) + } + + session, err := h.taskSvc.GetPrimarySession(ctx, req.TaskID) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "task not found or has no session: "+err.Error(), nil) + } + if session == nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "task has no active session — use create_task_kandev to start one", nil) + } + + status, err := h.dispatchTaskMessage(ctx, req.TaskID, session, req.Prompt) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "task_id": req.TaskID, + "session_id": session.ID, + "status": status, + }) +} + +// dispatchTaskMessage routes a message to the right delivery path based on session state. +// Returns the action taken: "queued", "sent", or "started". +func (h *Handlers) dispatchTaskMessage(ctx context.Context, taskID string, session *models.TaskSession, prompt string) (string, error) { + if h.sessionLauncher == nil { + return "", errors.New("orchestrator not available") + } + + switch session.State { + case models.TaskSessionStateFailed, models.TaskSessionStateCancelled: + return "", fmt.Errorf("session is %s — cannot send message", session.State) + + case models.TaskSessionStateRunning, models.TaskSessionStateStarting: + queue := h.sessionLauncher.GetMessageQueue() + if queue == nil { + return "", errors.New("message queue not available") + } + if _, err := queue.QueueMessage(ctx, session.ID, taskID, prompt, "", "agent", false, nil); err != nil { + return "", fmt.Errorf("failed to queue message: %w", err) + } + h.publishQueueStatusEvent(ctx, session.ID, queue) + return "queued", nil + + case models.TaskSessionStateCreated: + if _, err := h.sessionLauncher.StartCreatedSession(ctx, taskID, session.ID, session.AgentProfileID, prompt, false, false, nil); err != nil { + return "", fmt.Errorf("failed to start session: %w", err) + } + return "started", nil + + default: // WAITING_FOR_INPUT, COMPLETED, or any other promptable state + h.recordUserMessage(ctx, taskID, session.ID, prompt) + return h.promptWithAutoResume(ctx, taskID, session.ID, prompt) + } +} + +// recordUserMessage writes the prompt to the task's chat as a user message so it +// is visible in the conversation. Mirrors the message.add path used by the UI. +func (h *Handlers) recordUserMessage(ctx context.Context, taskID, sessionID, prompt string) { + if h.taskSvc == nil { + return + } + if _, err := h.taskSvc.CreateMessage(ctx, &service.CreateMessageRequest{ + TaskSessionID: sessionID, + TaskID: taskID, + Content: prompt, + AuthorType: "user", + }); err != nil { + h.logger.Warn("failed to record user message for message_task", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + } +} + +// promptWithAutoResume sends a prompt to a session and resumes the agent +// transparently if it has been torn down (mirrors message.add behaviour). +func (h *Handlers) promptWithAutoResume(ctx context.Context, taskID, sessionID, prompt string) (string, error) { + _, err := h.sessionLauncher.PromptTask(ctx, taskID, sessionID, prompt, "", false, nil) + if err == nil { + return "sent", nil + } + if !errors.Is(err, executor.ErrExecutionNotFound) { + return "", fmt.Errorf("failed to send prompt: %w", err) + } + if _, resumeErr := h.sessionLauncher.ResumeTaskSession(ctx, taskID, sessionID); resumeErr != nil { + return "", fmt.Errorf("failed to resume session: %w", resumeErr) + } + // ResumeTaskSession starts the agent asynchronously. Poll until the session + // is ready to accept prompts so the retry doesn't race the agent boot. + if waitErr := h.taskSvc.WaitForSessionReady(ctx, sessionID); waitErr != nil { + return "", fmt.Errorf("session not ready after resume: %w", waitErr) + } + if _, retryErr := h.sessionLauncher.PromptTask(ctx, taskID, sessionID, prompt, "", false, nil); retryErr != nil { + return "", fmt.Errorf("failed to send prompt after resume: %w", retryErr) + } + return "sent", nil +} + +// publishQueueStatusEvent fires a queue.status_changed event so the frontend +// can update the queue indicator. +func (h *Handlers) publishQueueStatusEvent(ctx context.Context, sessionID string, queue *messagequeue.Service) { + if h.eventBus == nil { + return + } + status := queue.GetStatus(ctx, sessionID) + _ = h.eventBus.Publish(ctx, events.MessageQueueStatusChanged, bus.NewEvent( + events.MessageQueueStatusChanged, + "mcp-handlers", + map[string]interface{}{ + "session_id": sessionID, + "is_queued": status.IsQueued, + "message": status.Message, + }, + )) +} + // handleAskUserQuestion creates a clarification request and blocks until the user responds. // The agent's MCP tool call stays open (same turn) while waiting. If the agent times out, // the event-based fallback in the orchestrator handles resuming with a new turn. diff --git a/apps/backend/internal/mcp/handlers/handlers_test.go b/apps/backend/internal/mcp/handlers/handlers_test.go index b81940e69f..0c3e365b11 100644 --- a/apps/backend/internal/mcp/handlers/handlers_test.go +++ b/apps/backend/internal/mcp/handlers/handlers_test.go @@ -13,11 +13,14 @@ import ( "github.com/kandev/kandev/internal/db" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator" + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" "github.com/kandev/kandev/internal/task/service" "github.com/kandev/kandev/internal/worktree" + v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -150,6 +153,20 @@ func (m *mockSessionLauncher) getRequest() *orchestrator.LaunchSessionRequest { return m.req } +// The following methods satisfy the SessionLauncher interface but are not used by +// the autoStartTask tests. handleMessageTask tests use a dedicated fakeOrchestrator +// (see message_task_test.go) that exercises these paths. +func (m *mockSessionLauncher) PromptTask(context.Context, string, string, string, string, bool, []v1.MessageAttachment) (*orchestrator.PromptResult, error) { + return nil, nil +} +func (m *mockSessionLauncher) StartCreatedSession(context.Context, string, string, string, string, bool, bool, []v1.MessageAttachment) (*executor.TaskExecution, error) { + return nil, nil +} +func (m *mockSessionLauncher) ResumeTaskSession(context.Context, string, string) (*executor.TaskExecution, error) { + return nil, nil +} +func (m *mockSessionLauncher) GetMessageQueue() *messagequeue.Service { return nil } + func TestAutoStartTask_DefaultsToWorktreeExecutor(t *testing.T) { launcher := newMockSessionLauncher() log := testLogger(t) diff --git a/apps/backend/internal/mcp/handlers/message_task_test.go b/apps/backend/internal/mcp/handlers/message_task_test.go new file mode 100644 index 0000000000..5ce35fa69d --- /dev/null +++ b/apps/backend/internal/mcp/handlers/message_task_test.go @@ -0,0 +1,330 @@ +package handlers + +import ( + "context" + "encoding/json" + "sync" + "testing" + "testing/synctest" + + "github.com/kandev/kandev/internal/orchestrator" + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/task/service" + v1 "github.com/kandev/kandev/pkg/api/v1" + ws "github.com/kandev/kandev/pkg/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeOrchestrator records calls to the SessionLauncher methods exercised by +// handleMessageTask. PromptTask returns a configurable error so the auto-resume +// path can be tested. +type fakeOrchestrator struct { + mu sync.Mutex + + queue *messagequeue.Service + + promptCalls []promptCall + startCreatedCalls []startCreatedCall + resumeCalls int + + // Configurable: error returned by PromptTask. Cleared after first call so + // the retry-after-resume path can succeed on the second call. + promptErrFirst error +} + +type promptCall struct { + taskID, sessionID, prompt string +} +type startCreatedCall struct { + taskID, sessionID, agentProfileID, prompt string + skipMessageRecord bool +} + +func (f *fakeOrchestrator) LaunchSession(context.Context, *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) { + return nil, nil +} + +func (f *fakeOrchestrator) PromptTask(_ context.Context, taskID, sessionID, prompt, _ string, _ bool, _ []v1.MessageAttachment) (*orchestrator.PromptResult, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.promptCalls = append(f.promptCalls, promptCall{taskID, sessionID, prompt}) + if f.promptErrFirst != nil { + err := f.promptErrFirst + f.promptErrFirst = nil + return nil, err + } + return &orchestrator.PromptResult{}, nil +} + +func (f *fakeOrchestrator) StartCreatedSession(_ context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, _ bool, _ []v1.MessageAttachment) (*executor.TaskExecution, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.startCreatedCalls = append(f.startCreatedCalls, startCreatedCall{ + taskID: taskID, + sessionID: sessionID, + agentProfileID: agentProfileID, + prompt: prompt, + skipMessageRecord: skipMessageRecord, + }) + return &executor.TaskExecution{SessionID: sessionID}, nil +} + +func (f *fakeOrchestrator) ResumeTaskSession(_ context.Context, _, _ string) (*executor.TaskExecution, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.resumeCalls++ + return &executor.TaskExecution{}, nil +} + +func (f *fakeOrchestrator) GetMessageQueue() *messagequeue.Service { return f.queue } + +func newMessageTaskHandler(t *testing.T, svc *service.Service) (*Handlers, *fakeOrchestrator) { + t.Helper() + log := testLogger(t) + orch := &fakeOrchestrator{queue: messagequeue.NewService(log)} + h := &Handlers{ + taskSvc: svc, + sessionLauncher: orch, + logger: log.WithFields(), + } + return h, orch +} + +// seedTaskWithSession creates a workspace, workflow, task, and primary session +// in the given state. Returns the task and session models. +func seedTaskWithSession(t *testing.T, svc *service.Service, repo interface { + CreateWorkspace(context.Context, *models.Workspace) error + CreateWorkflow(context.Context, *models.Workflow) error + CreateTaskSession(context.Context, *models.TaskSession) error + UpdateTaskSessionState(context.Context, string, models.TaskSessionState, string) error +}, state models.TaskSessionState) (*models.Task, *models.TaskSession) { + t.Helper() + ctx := context.Background() + require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Test"})) + require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "Board"})) + task, err := svc.CreateTask(ctx, &service.CreateTaskRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf-1", + Title: "Target task", + }) + require.NoError(t, err) + + sess := &models.TaskSession{ + ID: "sess-1", + TaskID: task.ID, + AgentProfileID: "agent-profile-1", + IsPrimary: true, + State: models.TaskSessionStateCreated, + } + require.NoError(t, repo.CreateTaskSession(ctx, sess)) + if state != models.TaskSessionStateCreated { + require.NoError(t, repo.UpdateTaskSessionState(ctx, sess.ID, state, "")) + } + loaded, err := svc.GetTaskSession(ctx, sess.ID) + require.NoError(t, err) + return task, loaded +} + +func TestHandleMessageTask_MissingTaskID(t *testing.T) { + h := &Handlers{} + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "prompt": "hello", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeValidation) +} + +func TestHandleMessageTask_MissingPrompt(t *testing.T) { + h := &Handlers{} + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": "task-1", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeValidation) +} + +func TestHandleMessageTask_BadPayload(t *testing.T) { + h := &Handlers{} + msg := &ws.Message{ + ID: "test-id", + Type: ws.MessageTypeRequest, + Action: ws.ActionMCPMessageTask, + Payload: json.RawMessage(`{not-json`), + } + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeBadRequest) +} + +func TestHandleMessageTask_RunningSession_Queues(t *testing.T) { + svc, repo := newTestTaskService(t) + task, sess := seedTaskWithSession(t, svc, repo, models.TaskSessionStateRunning) + + h, orch := newMessageTaskHandler(t, svc) + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": task.ID, + "prompt": "follow-up message", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, "queued", payload["status"]) + assert.Equal(t, sess.ID, payload["session_id"]) + + // Message landed in the queue. + status := orch.queue.GetStatus(context.Background(), sess.ID) + require.True(t, status.IsQueued) + assert.Equal(t, "follow-up message", status.Message.Content) + assert.Empty(t, orch.promptCalls) + assert.Empty(t, orch.startCreatedCalls) +} + +func TestHandleMessageTask_WaitingForInput_PromptsAgent(t *testing.T) { + svc, repo := newTestTaskService(t) + task, sess := seedTaskWithSession(t, svc, repo, models.TaskSessionStateWaitingForInput) + + h, orch := newMessageTaskHandler(t, svc) + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": task.ID, + "prompt": "next instruction", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, "sent", payload["status"]) + + require.Len(t, orch.promptCalls, 1) + assert.Equal(t, task.ID, orch.promptCalls[0].taskID) + assert.Equal(t, sess.ID, orch.promptCalls[0].sessionID) + assert.Equal(t, "next instruction", orch.promptCalls[0].prompt) + assert.Zero(t, orch.resumeCalls) + + // Prompt is recorded as a user message so it shows in the receiving task's chat. + messages, err := svc.ListMessages(context.Background(), sess.ID) + require.NoError(t, err) + require.Len(t, messages, 1) + assert.Equal(t, "next instruction", messages[0].Content) + assert.Equal(t, models.MessageAuthorUser, messages[0].AuthorType) +} + +func TestHandleMessageTask_PromptFailsWithExecutionNotFound_AutoResumes(t *testing.T) { + // Wrapped in synctest so the WaitForSessionReady poll's time.After advances + // virtually instead of blocking the test for ~1s of real time. Matches + // CLAUDE.md guidance to prefer synctest over time.Sleep-based waits. + synctest.Test(t, func(t *testing.T) { + svc, repo := newTestTaskService(t) + task, _ := seedTaskWithSession(t, svc, repo, models.TaskSessionStateWaitingForInput) + + h, orch := newMessageTaskHandler(t, svc) + orch.promptErrFirst = executor.ErrExecutionNotFound + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": task.ID, + "prompt": "retry me", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + assert.Len(t, orch.promptCalls, 2, "should retry prompt after resume") + assert.Equal(t, 1, orch.resumeCalls) + }) +} + +func TestHandleMessageTask_CreatedSession_StartsAgent(t *testing.T) { + svc, repo := newTestTaskService(t) + task, sess := seedTaskWithSession(t, svc, repo, models.TaskSessionStateCreated) + + h, orch := newMessageTaskHandler(t, svc) + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": task.ID, + "prompt": "kick off the work", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, "started", payload["status"]) + + require.Len(t, orch.startCreatedCalls, 1) + c := orch.startCreatedCalls[0] + assert.Equal(t, task.ID, c.taskID) + assert.Equal(t, sess.ID, c.sessionID) + assert.Equal(t, "agent-profile-1", c.agentProfileID) + assert.Equal(t, "kick off the work", c.prompt) + // skipMessageRecord must be false so postLaunchCreated → recordInitialMessage + // writes the prompt to the receiving task's chat. + assert.False(t, c.skipMessageRecord, "skipMessageRecord must be false so the prompt is recorded in chat") +} + +func TestHandleMessageTask_FailedSession_Rejects(t *testing.T) { + svc, repo := newTestTaskService(t) + task, _ := seedTaskWithSession(t, svc, repo, models.TaskSessionStateFailed) + + h, _ := newMessageTaskHandler(t, svc) + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": task.ID, + "prompt": "hello", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeInternalError) +} + +func TestHandleMessageTask_CancelledSession_Rejects(t *testing.T) { + svc, repo := newTestTaskService(t) + task, _ := seedTaskWithSession(t, svc, repo, models.TaskSessionStateCancelled) + + h, _ := newMessageTaskHandler(t, svc) + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": task.ID, + "prompt": "hello", + }) + resp, err := h.handleMessageTask(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeInternalError) +} + +func TestHandleMessageTask_NoPrimarySession_Rejects(t *testing.T) { + svc, repo := newTestTaskService(t) + ctx := context.Background() + require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Test"})) + require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "Board"})) + task, err := svc.CreateTask(ctx, &service.CreateTaskRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf-1", + Title: "Sessionless task", + }) + require.NoError(t, err) + + h, _ := newMessageTaskHandler(t, svc) + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{ + "task_id": task.ID, + "prompt": "hello", + }) + resp, err := h.handleMessageTask(ctx, msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeNotFound) +} diff --git a/apps/backend/internal/mcp/server/handlers.go b/apps/backend/internal/mcp/server/handlers.go index 417399fb07..8637866602 100644 --- a/apps/backend/internal/mcp/server/handlers.go +++ b/apps/backend/internal/mcp/server/handlers.go @@ -173,6 +173,29 @@ func (s *Server) updateTaskHandler() server.ToolHandlerFunc { } } +func (s *Server) messageTaskHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + taskID, err := req.RequireString("task_id") + if err != nil { + return mcp.NewToolResultError("task_id is required"), nil + } + prompt, err := req.RequireString("prompt") + if err != nil { + return mcp.NewToolResultError("prompt is required"), nil + } + payload := map[string]interface{}{ + "task_id": taskID, + "prompt": prompt, + } + var result map[string]interface{} + if err := s.backend.RequestPayload(ctx, ws.ActionMCPMessageTask, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + func (s *Server) askUserQuestionHandler() server.ToolHandlerFunc { return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { prompt, err := req.RequireString("prompt") diff --git a/apps/backend/internal/mcp/server/handlers_test.go b/apps/backend/internal/mcp/server/handlers_test.go index a1f3e9a9ba..7f584da197 100644 --- a/apps/backend/internal/mcp/server/handlers_test.go +++ b/apps/backend/internal/mcp/server/handlers_test.go @@ -264,3 +264,49 @@ func TestCreateTask_RepositoryURL_RejectedForSubtasks(t *testing.T) { assert.True(t, result.IsError, "repository_url should be rejected for subtasks") } + +func TestMessageTask_ForwardsToBackend(t *testing.T) { + backend := &testBackend{ + response: map[string]interface{}{ + "task_id": "task-target", + "session_id": "sess-1", + "status": "queued", + }, + } + s := newTaskModeServer(t, backend, "task-current") + + result := callTool(t, s, "message_task_kandev", map[string]interface{}{ + "task_id": "task-target", + "prompt": "follow up", + }) + + assert.False(t, result.IsError) + assert.Equal(t, ws.ActionMCPMessageTask, backend.lastAction) + + payload, ok := backend.lastPayload.(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, "task-target", payload["task_id"]) + assert.Equal(t, "follow up", payload["prompt"]) +} + +func TestMessageTask_MissingTaskID_ReturnsError(t *testing.T) { + backend := &testBackend{} + s := newTaskModeServer(t, backend, "task-current") + + result := callTool(t, s, "message_task_kandev", map[string]interface{}{ + "prompt": "follow up", + }) + + assert.True(t, result.IsError) +} + +func TestMessageTask_MissingPrompt_ReturnsError(t *testing.T) { + backend := &testBackend{} + s := newTaskModeServer(t, backend, "task-current") + + result := callTool(t, s, "message_task_kandev", map[string]interface{}{ + "task_id": "task-target", + }) + + assert.True(t, result.IsError) +} diff --git a/apps/backend/internal/mcp/server/server.go b/apps/backend/internal/mcp/server/server.go index f30b3bb29c..98fe6829e6 100644 --- a/apps/backend/internal/mcp/server/server.go +++ b/apps/backend/internal/mcp/server/server.go @@ -307,7 +307,7 @@ func (s *Server) registerTools() { count++ default: // ModeTask s.registerKanbanTools() - count += 9 + count += 10 if !s.disableAskQuestion { s.registerInteractionTools() count++ @@ -390,6 +390,24 @@ func (s *Server) registerKanbanTools() { ), s.wrapHandler("move_task_kandev", s.moveTaskHandler()), ) + s.mcpServer.AddTool( + mcp.NewTool("message_task_kandev", + mcp.WithDescription(`Send a follow-up prompt (message) to an existing task's primary session. + +Use this to communicate with a sibling task, a parent task, or any task you know the ID of — for example to ask a delegated subtask for clarification, hand it new context, or nudge a paused task forward. + +Behaviour by session state: +- Running/starting: the message is queued and delivered when the current turn ends. +- Idle (waiting for input or completed): the message is sent immediately as a new turn. +- Created (not yet started): the agent is started with this message as its first prompt. +- Failed/cancelled: an error is returned (use create_task_kandev to start fresh). + +Returns the dispatch status: "queued", "sent", or "started".`), + mcp.WithString("task_id", mcp.Required(), mcp.Description("The target task ID")), + mcp.WithString("prompt", mcp.Required(), mcp.Description("The message to deliver to the task's agent")), + ), + s.wrapHandler("message_task_kandev", s.messageTaskHandler()), + ) } // registerCreateTaskTool registers the create_task_kandev tool. Shared between diff --git a/apps/backend/internal/mcp/server/server_test.go b/apps/backend/internal/mcp/server/server_test.go index 65c5984d73..f645e967e5 100644 --- a/apps/backend/internal/mcp/server/server_test.go +++ b/apps/backend/internal/mcp/server/server_test.go @@ -43,6 +43,7 @@ func TestServerModeTask_RegistersCorrectTools(t *testing.T) { assert.Contains(t, tools, "create_task_kandev") assert.Contains(t, tools, "update_task_kandev") assert.Contains(t, tools, "move_task_kandev") + assert.Contains(t, tools, "message_task_kandev") // Task mode should have plan tools assert.Contains(t, tools, "create_task_plan_kandev") @@ -193,8 +194,8 @@ func TestServerModeTask_ToolCount(t *testing.T) { s := New(backend, "test-session", "test-task", 10005, log, "", false, ModeTask) tools := getRegisteredToolNames(s) - // 9 kanban + 1 interaction + 4 plan = 14 - assert.Equal(t, 14, len(tools)) + // 10 kanban + 1 interaction + 4 plan = 15 + assert.Equal(t, 15, len(tools)) } func TestServerModeConfig_ToolCount(t *testing.T) { @@ -258,6 +259,9 @@ func TestServerModeExternal_RegistersCorrectTools(t *testing.T) { // External mode does NOT include kanban update_task_kandev (config has its own update_task_state) assert.NotContains(t, tools, "update_task_kandev") + + // External mode does NOT include message_task_kandev (no live session context) + assert.NotContains(t, tools, "message_task_kandev") } func TestServerModeExternal_ToolCount(t *testing.T) { diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index 0fce164cdb..9d3caa9cc1 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -3,11 +3,9 @@ package handlers import ( "context" "errors" - "fmt" "net/http" "strconv" "strings" - "time" "unicode" "github.com/gin-gonic/gin" @@ -456,43 +454,10 @@ func (h *MessageHandlers) createPromptErrorMessage(ctx context.Context, taskID, } } -// waitForSessionReady polls the session state after a resume operation until the agent -// is ready to accept prompts. It returns nil when the session reaches WAITING_FOR_INPUT -// state, or an error if the session transitions to FAILED or the timeout is exceeded. +// waitForSessionReady delegates to the shared service.WaitForSessionReady helper. +// Kept as a thin wrapper so existing tests on this method continue to pass. func (h *MessageHandlers) waitForSessionReady(ctx context.Context, sessionID string) error { - const ( - pollInterval = 1 * time.Second - maxWait = 90 * time.Second - ) - deadline := time.Now().Add(maxWait) - for { - if time.Now().After(deadline) { - return fmt.Errorf("timeout waiting for session to become ready after resume") - } - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(pollInterval): - } - session, err := h.service.GetTaskSession(ctx, sessionID) - if err != nil { - return fmt.Errorf("failed to check session state: %w", err) - } - switch session.State { - case models.TaskSessionStateWaitingForInput: - return nil - case models.TaskSessionStateFailed: - errMsg := session.ErrorMessage - if errMsg == "" { - errMsg = "session failed during resume" - } - return fmt.Errorf("session failed after resume: %s", errMsg) - case models.TaskSessionStateCancelled, models.TaskSessionStateCompleted: - return fmt.Errorf("session in unexpected state after resume: %s", session.State) - default: - // STARTING or RUNNING — keep polling - } - } + return h.service.WaitForSessionReady(ctx, sessionID) } type wsListMessagesRequest struct { diff --git a/apps/backend/internal/task/service/service_sessions.go b/apps/backend/internal/task/service/service_sessions.go index 28fab9d33b..87d05fed4f 100644 --- a/apps/backend/internal/task/service/service_sessions.go +++ b/apps/backend/internal/task/service/service_sessions.go @@ -2,12 +2,57 @@ package service import ( "context" + "fmt" + "time" "go.uber.org/zap" "github.com/kandev/kandev/internal/task/models" ) +// SessionReadyPollInterval is how often WaitForSessionReady polls session state. +const SessionReadyPollInterval = 1 * time.Second + +// SessionReadyMaxWait is the maximum time WaitForSessionReady waits before timing out. +const SessionReadyMaxWait = 90 * time.Second + +// WaitForSessionReady polls the session state until the agent is ready to accept +// prompts. Returns nil when the session reaches WAITING_FOR_INPUT, or an error if +// it transitions to FAILED/CANCELLED/COMPLETED, the context is cancelled, or the +// timeout is exceeded. Used after ResumeTaskSession to gate the prompt retry until +// the agent has actually finished booting. +func (s *Service) WaitForSessionReady(ctx context.Context, sessionID string) error { + deadline := time.Now().Add(SessionReadyMaxWait) + for { + if time.Now().After(deadline) { + return fmt.Errorf("timeout waiting for session to become ready after resume") + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(SessionReadyPollInterval): + } + session, err := s.sessions.GetTaskSession(ctx, sessionID) + if err != nil { + return fmt.Errorf("failed to check session state: %w", err) + } + switch session.State { + case models.TaskSessionStateWaitingForInput: + return nil + case models.TaskSessionStateFailed: + errMsg := session.ErrorMessage + if errMsg == "" { + errMsg = "session failed during resume" + } + return fmt.Errorf("session failed after resume: %s", errMsg) + case models.TaskSessionStateCancelled, models.TaskSessionStateCompleted: + return fmt.Errorf("session in unexpected state after resume: %s", session.State) + default: + // STARTING or RUNNING — keep polling + } + } +} + // ListTaskSessions returns all sessions for a task. func (s *Service) ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) { return s.sessions.ListTaskSessions(ctx, taskID) diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index d17acc0c9e..b6c26c0076 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -350,6 +350,7 @@ const ( ActionMCPDeleteTask = "mcp.delete_task" ActionMCPArchiveTask = "mcp.archive_task" ActionMCPUpdateTaskState = "mcp.update_task_state" + ActionMCPMessageTask = "mcp.message_task" ) // GitHub integration actions