feat(mcp): add get_task_conversation tool (#756)

* feat(mcp): add get_task_conversation tool * fix(mcp): correct conversation pagination cursor and limits --------- Co-authored-by: Kandev Agent <agent@kandev.dev>
2026-05-01 18:20:01 +01:00
parent e16debeaca
commit c8fcd6c5aa
8 changed files with 414 additions and 15 deletions
@@ -0,0 +1,135 @@
+package handlers
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/service"
+	ws "github.com/kandev/kandev/pkg/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestHandleGetTaskConversation_MissingTaskID(t *testing.T) {
+	h := &Handlers{}
+	msg := makeWSMessage(t, ws.ActionMCPGetTaskConversation, map[string]interface{}{})
+	resp, err := h.handleGetTaskConversation(context.Background(), msg)
+	require.NoError(t, err)
+	assertWSError(t, resp, ws.ErrorCodeValidation)
+}
+
+func TestHandleGetTaskConversation_UsesPrimarySession(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	task, sess := seedTaskWithSession(t, svc, repo, models.TaskSessionStateWaitingForInput)
+
+	_, err := svc.CreateMessage(context.Background(), &service.CreateMessageRequest{
+		TaskSessionID: sess.ID,
+		TaskID:        task.ID,
+		AuthorType:    "user",
+		Content:       "hello from task",
+	})
+	require.NoError(t, err)
+
+	h := &Handlers{taskSvc: svc, logger: testLogger(t).WithFields()}
+	msg := makeWSMessage(t, ws.ActionMCPGetTaskConversation, map[string]interface{}{
+		"task_id": task.ID,
+		"limit":   10,
+	})
+
+	resp, err := h.handleGetTaskConversation(context.Background(), msg)
+	require.NoError(t, err)
+	assert.Equal(t, ws.MessageTypeResponse, resp.Type)
+
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Payload, &payload))
+	assert.Equal(t, task.ID, payload["task_id"])
+	assert.Equal(t, sess.ID, payload["session_id"])
+	assert.Equal(t, float64(1), payload["total"])
+}
+
+func TestHandleGetTaskConversation_SessionMustBelongToTask(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	taskA, _ := seedTaskWithSession(t, svc, repo, models.TaskSessionStateWaitingForInput)
+
+	// Create another task/session in the same workflow to validate cross-task mismatch.
+	taskB, err := svc.CreateTask(context.Background(), &service.CreateTaskRequest{
+		WorkspaceID: "ws-1",
+		WorkflowID:  "wf-1",
+		Title:       "Other task",
+	})
+	require.NoError(t, err)
+	sessB := &models.TaskSession{
+		ID:             "sess-2",
+		TaskID:         taskB.ID,
+		AgentProfileID: "agent-profile-1",
+		IsPrimary:      true,
+		State:          models.TaskSessionStateWaitingForInput,
+	}
+	require.NoError(t, repo.CreateTaskSession(context.Background(), sessB))
+
+	h := &Handlers{taskSvc: svc, logger: testLogger(t).WithFields()}
+	msg := makeWSMessage(t, ws.ActionMCPGetTaskConversation, map[string]interface{}{
+		"task_id":    taskA.ID,
+		"session_id": "sess-2",
+	})
+
+	resp, err := h.handleGetTaskConversation(context.Background(), msg)
+	require.NoError(t, err)
+	assertWSError(t, resp, ws.ErrorCodeValidation)
+}
+
+func TestHandleGetTaskConversation_NegativeLimit(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	task, _ := seedTaskWithSession(t, svc, repo, models.TaskSessionStateWaitingForInput)
+
+	h := &Handlers{taskSvc: svc, logger: testLogger(t).WithFields()}
+	msg := makeWSMessage(t, ws.ActionMCPGetTaskConversation, map[string]interface{}{
+		"task_id": task.ID,
+		"limit":   -1,
+	})
+
+	resp, err := h.handleGetTaskConversation(context.Background(), msg)
+	require.NoError(t, err)
+	assertWSError(t, resp, ws.ErrorCodeValidation)
+}
+
+func TestHandleGetTaskConversation_FilteredPageStillReturnsCursor(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	task, sess := seedTaskWithSession(t, svc, repo, models.TaskSessionStateWaitingForInput)
+
+	_, err := svc.CreateMessage(context.Background(), &service.CreateMessageRequest{
+		TaskSessionID: sess.ID,
+		TaskID:        task.ID,
+		AuthorType:    "agent",
+		Type:          "tool_call",
+		Content:       "tool call 1",
+	})
+	require.NoError(t, err)
+	_, err = svc.CreateMessage(context.Background(), &service.CreateMessageRequest{
+		TaskSessionID: sess.ID,
+		TaskID:        task.ID,
+		AuthorType:    "agent",
+		Type:          "tool_call",
+		Content:       "tool call 2",
+	})
+	require.NoError(t, err)
+
+	h := &Handlers{taskSvc: svc, logger: testLogger(t).WithFields()}
+	msg := makeWSMessage(t, ws.ActionMCPGetTaskConversation, map[string]interface{}{
+		"task_id":       task.ID,
+		"limit":         1,
+		"message_types": []string{"message"},
+	})
+
+	resp, err := h.handleGetTaskConversation(context.Background(), msg)
+	require.NoError(t, err)
+	assert.Equal(t, ws.MessageTypeResponse, resp.Type)
+
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Payload, &payload))
+	assert.Equal(t, float64(0), payload["total"])
+	assert.Equal(t, true, payload["has_more"])
+	assert.NotEmpty(t, payload["cursor"])
+}
@@ -148,13 +148,14 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) {
 	d.RegisterFunc(ws.ActionMCPCreateTask, h.handleCreateTask)
 	d.RegisterFunc(ws.ActionMCPUpdateTask, h.handleUpdateTask)
 	d.RegisterFunc(ws.ActionMCPMessageTask, h.handleMessageTask)
+	d.RegisterFunc(ws.ActionMCPGetTaskConversation, h.handleGetTaskConversation)
 	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 := 13
+	count := 14

 	// Config-mode handlers (registered when config deps are set)
 	if h.workflowSvc != nil {
@@ -646,6 +647,132 @@ func (h *Handlers) handleMessageTask(ctx context.Context, msg *ws.Message) (*ws.
 	})
 }

+// handleGetTaskConversation returns paginated conversation history for a task.
+// If session_id is omitted, it uses the task's primary session.
+func (h *Handlers) handleGetTaskConversation(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
+	req, errResp := parseTaskConversationRequest(msg)
+	if errResp != nil {
+		return errResp, nil
+	}
+
+	session, errResp := h.resolveConversationSession(ctx, msg, req.TaskID, req.SessionID)
+	if errResp != nil {
+		return errResp, nil
+	}
+
+	messages, hasMore, err := h.taskSvc.ListMessagesPaginated(ctx, service.ListMessagesRequest{
+		TaskSessionID: session.ID,
+		Limit:         conversationLimit(req.Limit),
+		Before:        req.Before,
+		After:         req.After,
+		Sort:          req.Sort,
+	})
+	if err != nil {
+		h.logger.Error("failed to list task conversation", zap.Error(err))
+		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list task conversation", nil)
+	}
+
+	result := filterAndConvertMessages(messages, req.Types)
+	cursor := conversationCursor(messages)
+
+	return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
+		"task_id":    req.TaskID,
+		"session_id": session.ID,
+		"messages":   result,
+		"total":      len(result),
+		"has_more":   hasMore,
+		"cursor":     cursor,
+	})
+}
+
+type taskConversationRequest struct {
+	TaskID    string   `json:"task_id"`
+	SessionID string   `json:"session_id"`
+	Limit     int      `json:"limit"`
+	Before    string   `json:"before"`
+	After     string   `json:"after"`
+	Sort      string   `json:"sort"`
+	Types     []string `json:"message_types"`
+}
+
+func parseTaskConversationRequest(msg *ws.Message) (*taskConversationRequest, *ws.Message) {
+	var req taskConversationRequest
+	if err := json.Unmarshal(msg.Payload, &req); err != nil {
+		return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error())
+	}
+	if req.TaskID == "" {
+		return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required")
+	}
+	if req.Before != "" && req.After != "" {
+		return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeValidation, "only one of before or after can be set")
+	}
+	if req.Sort != "" && req.Sort != "asc" && req.Sort != "desc" {
+		return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeValidation, "sort must be asc or desc")
+	}
+	if req.Limit < 0 {
+		return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeValidation, "limit must be non-negative")
+	}
+	return &req, nil
+}
+
+func (h *Handlers) resolveConversationSession(ctx context.Context, msg *ws.Message, taskID, sessionID string) (*models.TaskSession, *ws.Message) {
+	if sessionID != "" {
+		session, err := h.taskSvc.GetTaskSession(ctx, sessionID)
+		if err != nil || session == nil {
+			return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "session not found")
+		}
+		if session.TaskID != taskID {
+			return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id does not belong to task_id")
+		}
+		return session, nil
+	}
+	session, err := h.taskSvc.GetPrimarySession(ctx, taskID)
+	if err != nil || session == nil {
+		return nil, wsError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "task has no session")
+	}
+	return session, nil
+}
+
+func wsError(id, action, code, message string) *ws.Message {
+	resp, _ := ws.NewError(id, action, code, message, nil)
+	return resp
+}
+
+func filterAndConvertMessages(messages []*models.Message, types []string) []*v1.Message {
+	filterTypes := make(map[string]struct{}, len(types))
+	for _, mt := range types {
+		if mt == "" {
+			continue
+		}
+		filterTypes[mt] = struct{}{}
+	}
+
+	result := make([]*v1.Message, 0, len(messages))
+	for _, message := range messages {
+		if len(filterTypes) > 0 {
+			if _, ok := filterTypes[string(message.Type)]; !ok {
+				continue
+			}
+		}
+		result = append(result, message.ToAPI())
+	}
+	return result
+}
+
+func conversationLimit(requested int) int {
+	if requested > 0 {
+		return requested
+	}
+	return service.DefaultMessagesPageSize
+}
+
+func conversationCursor(messages []*models.Message) string {
+	if len(messages) == 0 {
+		return ""
+	}
+	return messages[len(messages)-1].ID
+}
+
 // 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) {
@@ -278,6 +278,19 @@ func (s *Server) registerConfigTaskTools() {
 		),
 		s.wrapHandler("update_task_state_kandev", s.updateTaskStateHandler()),
 	)
+	s.mcpServer.AddTool(
+		mcp.NewTool("get_task_conversation_kandev",
+			mcp.WithDescription("Get conversation history for a task. If session_id is omitted, the primary session is used."),
+			mcp.WithString("task_id", mcp.Required(), mcp.Description("The task ID")),
+			mcp.WithString("session_id", mcp.Description("Optional session ID (must belong to task_id)")),
+			mcp.WithNumber("limit", mcp.Description("Optional page size (defaults to backend setting, max backend-capped)")),
+			mcp.WithString("before", mcp.Description("Optional cursor message ID to fetch messages before this ID")),
+			mcp.WithString("after", mcp.Description("Optional cursor message ID to fetch messages after this ID")),
+			mcp.WithString("sort", mcp.Description("Optional sort order: asc or desc")),
+			mcp.WithArray("message_types", mcp.Description("Optional message type filters (e.g. message, tool_call, error)"), mcp.Items(map[string]any{"type": "string"})),
+		),
+		s.wrapHandler("get_task_conversation_kandev", s.getTaskConversationHandler()),
+	)
 }

 // --- Handler implementations ---
@@ -196,6 +196,72 @@ func (s *Server) messageTaskHandler() server.ToolHandlerFunc {
 	}
 }

+func (s *Server) getTaskConversationHandler() 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
+		}
+		payload := buildTaskConversationPayload(req, taskID)
+
+		var result map[string]interface{}
+		if err := s.backend.RequestPayload(ctx, ws.ActionMCPGetTaskConversation, payload, &result); err != nil {
+			return mcp.NewToolResultError(err.Error()), nil
+		}
+		data, _ := json.MarshalIndent(result, "", "  ")
+		return mcp.NewToolResultText(string(data)), nil
+	}
+}
+
+func buildTaskConversationPayload(req mcp.CallToolRequest, taskID string) map[string]interface{} {
+	payload := map[string]interface{}{"task_id": taskID}
+	copyOptionalStringArg(payload, req, "session_id")
+	copyOptionalStringArg(payload, req, "before")
+	copyOptionalStringArg(payload, req, "after")
+	copyOptionalStringArg(payload, req, "sort")
+	copyOptionalLimitArg(payload, req)
+	copyOptionalMessageTypesArg(payload, req)
+	return payload
+}
+
+func copyOptionalStringArg(payload map[string]interface{}, req mcp.CallToolRequest, key string) {
+	if value := req.GetString(key, ""); value != "" {
+		payload[key] = value
+	}
+}
+
+func copyOptionalLimitArg(payload map[string]interface{}, req mcp.CallToolRequest) {
+	args := req.GetArguments()
+	if raw := args["limit"]; raw != nil {
+		if limit, ok := raw.(float64); ok {
+			payload["limit"] = int(limit)
+		}
+	}
+}
+
+func copyOptionalMessageTypesArg(payload map[string]interface{}, req mcp.CallToolRequest) {
+	args := req.GetArguments()
+	raw := args["message_types"]
+	if raw == nil {
+		return
+	}
+	items, ok := raw.([]interface{})
+	if !ok {
+		return
+	}
+	types := make([]string, 0, len(items))
+	for _, item := range items {
+		value, ok := item.(string)
+		if !ok || value == "" {
+			continue
+		}
+		types = append(types, value)
+	}
+	if len(types) > 0 {
+		payload["message_types"] = types
+	}
+}
+
 func (s *Server) askUserQuestionHandler() server.ToolHandlerFunc {
 	return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
 		prompt, err := req.RequireString("prompt")
@@ -310,3 +310,45 @@ func TestMessageTask_MissingPrompt_ReturnsError(t *testing.T) {

 	assert.True(t, result.IsError)
 }
+
+func TestGetTaskConversation_ForwardsToBackend(t *testing.T) {
+	backend := &testBackend{
+		response: map[string]interface{}{
+			"task_id":    "task-target",
+			"session_id": "sess-1",
+			"messages":   []interface{}{},
+			"total":      0,
+			"has_more":   false,
+			"cursor":     "",
+		},
+	}
+	s := newTaskModeServer(t, backend, "task-current")
+
+	result := callTool(t, s, "get_task_conversation_kandev", map[string]interface{}{
+		"task_id":       "task-target",
+		"session_id":    "sess-1",
+		"limit":         25,
+		"sort":          "desc",
+		"message_types": []interface{}{"message", "tool_call"},
+	})
+
+	assert.False(t, result.IsError)
+	assert.Equal(t, ws.ActionMCPGetTaskConversation, backend.lastAction)
+
+	payload, ok := backend.lastPayload.(map[string]interface{})
+	require.True(t, ok)
+	assert.Equal(t, "task-target", payload["task_id"])
+	assert.Equal(t, "sess-1", payload["session_id"])
+	assert.Equal(t, 25, payload["limit"])
+	assert.Equal(t, "desc", payload["sort"])
+	assert.Equal(t, []string{"message", "tool_call"}, payload["message_types"])
+}
+
+func TestGetTaskConversation_MissingTaskID_ReturnsError(t *testing.T) {
+	backend := &testBackend{}
+	s := newTaskModeServer(t, backend, "task-current")
+
+	result := callTool(t, s, "get_task_conversation_kandev", map[string]interface{}{})
+
+	assert.True(t, result.IsError)
+}
@@ -284,7 +284,7 @@ func (s *Server) registerTools() {
 		s.registerConfigExecutorTools()
 		count += 5
 		s.registerConfigTaskTools()
-		count += 5
+		count += 6
 		if !s.disableAskQuestion {
 			s.registerInteractionTools()
 			count++
@@ -302,12 +302,12 @@ func (s *Server) registerTools() {
 		s.registerConfigExecutorTools()
 		count += 5
 		s.registerConfigTaskTools()
-		count += 5
+		count += 6
 		s.registerCreateTaskTool()
 		count++
 	default: // ModeTask
 		s.registerKanbanTools()
-		count += 10
+		count += 11
 		if !s.disableAskQuestion {
 			s.registerInteractionTools()
 			count++
@@ -408,6 +408,19 @@ Returns the dispatch status: "queued", "sent", or "started".`),
 		),
 		s.wrapHandler("message_task_kandev", s.messageTaskHandler()),
 	)
+	s.mcpServer.AddTool(
+		mcp.NewTool("get_task_conversation_kandev",
+			mcp.WithDescription("Get conversation history for a task. If session_id is omitted, the primary session is used."),
+			mcp.WithString("task_id", mcp.Required(), mcp.Description("The task ID")),
+			mcp.WithString("session_id", mcp.Description("Optional session ID (must belong to task_id)")),
+			mcp.WithNumber("limit", mcp.Description("Optional page size (defaults to backend setting, max backend-capped)")),
+			mcp.WithString("before", mcp.Description("Optional cursor message ID to fetch messages before this ID")),
+			mcp.WithString("after", mcp.Description("Optional cursor message ID to fetch messages after this ID")),
+			mcp.WithString("sort", mcp.Description("Optional sort order: asc or desc")),
+			mcp.WithArray("message_types", mcp.Description("Optional message type filters (e.g. message, tool_call, error)"), mcp.Items(map[string]any{"type": "string"})),
+		),
+		s.wrapHandler("get_task_conversation_kandev", s.getTaskConversationHandler()),
+	)
 }

 // registerCreateTaskTool registers the create_task_kandev tool. Shared between
@@ -44,6 +44,7 @@ func TestServerModeTask_RegistersCorrectTools(t *testing.T) {
 	assert.Contains(t, tools, "update_task_kandev")
 	assert.Contains(t, tools, "move_task_kandev")
 	assert.Contains(t, tools, "message_task_kandev")
+	assert.Contains(t, tools, "get_task_conversation_kandev")

 	// Task mode should have plan tools
 	assert.Contains(t, tools, "create_task_plan_kandev")
@@ -129,6 +130,7 @@ func TestServerModeConfig_RegistersCorrectTools(t *testing.T) {
 	assert.Contains(t, tools, "delete_task_kandev")
 	assert.Contains(t, tools, "archive_task_kandev")
 	assert.Contains(t, tools, "update_task_state_kandev")
+	assert.Contains(t, tools, "get_task_conversation_kandev")

 	// Config mode should have interaction tools
 	assert.Contains(t, tools, "ask_user_question_kandev")
@@ -194,8 +196,8 @@ func TestServerModeTask_ToolCount(t *testing.T) {

 	s := New(backend, "test-session", "test-task", 10005, log, "", false, ModeTask)
 	tools := getRegisteredToolNames(s)
-	// 10 kanban + 1 interaction + 4 plan = 15
-	assert.Equal(t, 15, len(tools))
+	// 11 kanban + 1 interaction + 4 plan = 16
+	assert.Equal(t, 16, len(tools))
 }

 func TestServerModeConfig_ToolCount(t *testing.T) {
@@ -205,8 +207,8 @@ func TestServerModeConfig_ToolCount(t *testing.T) {

 	s := New(backend, "test-session", "test-task", 10005, log, "", false, ModeConfig)
 	tools := getRegisteredToolNames(s)
-	// 10 workflow + 4 agent + 4 mcp + 5 executor + 5 task + 1 interaction = 29
-	assert.Equal(t, 29, len(tools))
+	// 10 workflow + 4 agent + 4 mcp + 5 executor + 6 task + 1 interaction = 30
+	assert.Equal(t, 30, len(tools))
 }

 func TestServerModeConfig_ToolDescriptions(t *testing.T) {
@@ -271,8 +273,8 @@ func TestServerModeExternal_ToolCount(t *testing.T) {

 	s := New(backend, "", "", 0, log, "", true, ModeExternal)
 	tools := getRegisteredToolNames(s)
-	// 10 workflow + 4 agent + 4 mcp + 5 executor + 5 task + 1 create_task = 29
-	assert.Equal(t, 29, len(tools))
+	// 10 workflow + 4 agent + 4 mcp + 5 executor + 6 task + 1 create_task = 30
+	assert.Equal(t, 30, len(tools))
 }

 func TestNewExternal_Constructs(t *testing.T) {
@@ -347,11 +347,12 @@ const (
 	ActionMCPUpdateExecutorProfile = "mcp.update_executor_profile"
 	ActionMCPDeleteExecutorProfile = "mcp.delete_executor_profile"

-	ActionMCPMoveTask        = "mcp.move_task"
-	ActionMCPDeleteTask      = "mcp.delete_task"
-	ActionMCPArchiveTask     = "mcp.archive_task"
-	ActionMCPUpdateTaskState = "mcp.update_task_state"
-	ActionMCPMessageTask     = "mcp.message_task"
+	ActionMCPMoveTask            = "mcp.move_task"
+	ActionMCPDeleteTask          = "mcp.delete_task"
+	ActionMCPArchiveTask         = "mcp.archive_task"
+	ActionMCPUpdateTaskState     = "mcp.update_task_state"
+	ActionMCPMessageTask         = "mcp.message_task"
+	ActionMCPGetTaskConversation = "mcp.get_task_conversation"
 )

 // GitHub integration actions