feat(mcp): add message_task_kandev tool (#745)

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.

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)

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)
+}

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")

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)
+}

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

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) {

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 {

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)

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