diff --git a/apps/backend/internal/mcp/handlers/get_task_conversation_test.go b/apps/backend/internal/mcp/handlers/get_task_conversation_test.go new file mode 100644 index 0000000000..8185d48fa2 --- /dev/null +++ b/apps/backend/internal/mcp/handlers/get_task_conversation_test.go @@ -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"]) +} diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index ee63d1c8b4..33af08415d 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -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) { diff --git a/apps/backend/internal/mcp/server/config_handlers.go b/apps/backend/internal/mcp/server/config_handlers.go index e6bc4a26c7..32e27057d7 100644 --- a/apps/backend/internal/mcp/server/config_handlers.go +++ b/apps/backend/internal/mcp/server/config_handlers.go @@ -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 --- diff --git a/apps/backend/internal/mcp/server/handlers.go b/apps/backend/internal/mcp/server/handlers.go index 8637866602..d3bd785553 100644 --- a/apps/backend/internal/mcp/server/handlers.go +++ b/apps/backend/internal/mcp/server/handlers.go @@ -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") diff --git a/apps/backend/internal/mcp/server/handlers_test.go b/apps/backend/internal/mcp/server/handlers_test.go index 7f584da197..67b03f6783 100644 --- a/apps/backend/internal/mcp/server/handlers_test.go +++ b/apps/backend/internal/mcp/server/handlers_test.go @@ -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) +} diff --git a/apps/backend/internal/mcp/server/server.go b/apps/backend/internal/mcp/server/server.go index 98fe6829e6..68d2aa9836 100644 --- a/apps/backend/internal/mcp/server/server.go +++ b/apps/backend/internal/mcp/server/server.go @@ -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 diff --git a/apps/backend/internal/mcp/server/server_test.go b/apps/backend/internal/mcp/server/server_test.go index f645e967e5..64a81c8e77 100644 --- a/apps/backend/internal/mcp/server/server_test.go +++ b/apps/backend/internal/mcp/server/server_test.go @@ -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) { diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 727cbc0edb..78f5aa4f64 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -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