diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 78b1eb75bc..e41fed4233 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -379,6 +379,7 @@ func main() { if worktreeMgr != nil { taskController.SetWorktreeLookup(worktreeMgr) // Enable worktree info in task responses } + commentController := taskcontroller.NewCommentController(taskSvc) repositoryController := taskcontroller.NewRepositoryController(taskSvc) executorController := taskcontroller.NewExecutorController(taskSvc) environmentController := taskcontroller.NewEnvironmentController(taskSvc) @@ -658,7 +659,14 @@ func main() { taskhandlers.RegisterRepositoryRoutes(router, gateway.Dispatcher, repositoryController, log) taskhandlers.RegisterExecutorRoutes(router, gateway.Dispatcher, executorController, log) taskhandlers.RegisterEnvironmentRoutes(router, gateway.Dispatcher, environmentController, log) - taskhandlers.RegisterCommentRoutes(router, gateway.Dispatcher, taskSvc, &orchestratorAdapter{svc: orchestratorSvc}, log) + taskhandlers.RegisterCommentRoutes( + router, + gateway.Dispatcher, + commentController, + taskController, + &orchestratorAdapter{svc: orchestratorSvc}, + log, + ) log.Info("Registered Task Service handlers (HTTP + WebSocket)") agentsettingshandlers.RegisterRoutes(router, agentSettingsController, gateway.Hub, log) diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index fd7ae65a1d..4e09975fc5 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -36,6 +36,7 @@ type Client struct { subscriptions map[string]bool // Task IDs this client is subscribed to userSubscriptions map[string]bool // User IDs this client is subscribed to mu sync.RWMutex + closed bool logger *logger.Logger } @@ -262,11 +263,22 @@ func (c *Client) sendMessage(msg *ws.Message) { c.logger.Error("Failed to marshal message", zap.Error(err)) return } + c.sendBytes(data) +} + +func (c *Client) sendBytes(data []byte) bool { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return false + } select { case c.send <- data: + return true default: c.logger.Warn("Client send buffer full") + return false } } @@ -280,6 +292,16 @@ func (c *Client) sendError(id, action, code, message string, details map[string] c.sendMessage(msg) } +func (c *Client) closeSend() { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return + } + c.closed = true + close(c.send) +} + // WritePump pumps messages from the hub to the WebSocket connection func (c *Client) WritePump() { ticker := time.NewTicker(pingPeriod) diff --git a/apps/backend/internal/gateway/websocket/hub.go b/apps/backend/internal/gateway/websocket/hub.go index 0637142166..b34a12271b 100644 --- a/apps/backend/internal/gateway/websocket/hub.go +++ b/apps/backend/internal/gateway/websocket/hub.go @@ -87,7 +87,7 @@ func (h *Hub) closeAllClients() { defer h.mu.Unlock() for client := range h.clients { - close(client.send) + client.closeSend() delete(h.clients, client) } h.taskSubscribers = make(map[string]map[*Client]bool) @@ -100,7 +100,7 @@ func (h *Hub) removeClient(client *Client) { if _, ok := h.clients[client]; ok { delete(h.clients, client) - close(client.send) + client.closeSend() // Remove from all task subscriptions for taskID := range client.subscriptions { @@ -137,11 +137,7 @@ func (h *Hub) broadcastMessage(msg *ws.Message) { // For now, broadcast to all clients // TODO: Add topic-based routing for task-specific notifications for client := range h.clients { - select { - case client.send <- data: - default: - // Client buffer full, will be cleaned up by write pump - } + client.sendBytes(data) } } @@ -179,14 +175,14 @@ func (h *Hub) BroadcastToTask(taskID string, msg *ws.Message) { zap.Int("subscriber_count", numClients)) for client := range clients { - select { - case client.send <- data: + if client.sendBytes(data) { h.logger.Debug("Sent message to client", zap.String("client_id", client.ID), zap.String("action", msg.Action)) - default: - h.logger.Warn("Client buffer full, dropping message", - zap.String("client_id", client.ID)) + } else { + h.logger.Warn("Client send buffer full, dropping message", + zap.String("client_id", client.ID), + zap.String("action", msg.Action)) } } } @@ -210,14 +206,14 @@ func (h *Hub) BroadcastToUser(userID string, msg *ws.Message) { zap.Int("subscriber_count", numClients)) for client := range clients { - select { - case client.send <- data: + if client.sendBytes(data) { h.logger.Debug("Sent message to client", zap.String("client_id", client.ID), zap.String("action", msg.Action)) - default: - h.logger.Warn("Client buffer full, dropping message", - zap.String("client_id", client.ID)) + } else { + h.logger.Warn("Client send buffer full, dropping message", + zap.String("client_id", client.ID), + zap.String("action", msg.Action)) } } } diff --git a/apps/backend/internal/task/controller/comment.go b/apps/backend/internal/task/controller/comment.go new file mode 100644 index 0000000000..ce2e13dab1 --- /dev/null +++ b/apps/backend/internal/task/controller/comment.go @@ -0,0 +1,50 @@ +package controller + +import ( + "context" + + "github.com/kandev/kandev/internal/task/dto" + "github.com/kandev/kandev/internal/task/service" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +type CommentController struct { + service *service.Service +} + +func NewCommentController(svc *service.Service) *CommentController { + return &CommentController{service: svc} +} + +func (c *CommentController) ListComments(ctx context.Context, req dto.ListCommentsRequest) (dto.ListCommentsResponse, error) { + comments, err := c.service.ListComments(ctx, req.TaskID) + if err != nil { + return dto.ListCommentsResponse{}, err + } + result := make([]*v1.Comment, 0, len(comments)) + for _, comment := range comments { + result = append(result, comment.ToAPI()) + } + return dto.ListCommentsResponse{ + Comments: result, + Total: len(result), + }, nil +} + +func (c *CommentController) CreateComment(ctx context.Context, req dto.CreateCommentRequest) (*v1.Comment, error) { + comment, err := c.service.CreateComment(ctx, &service.CreateCommentRequest{ + TaskID: req.TaskID, + Content: req.Content, + AuthorType: req.AuthorType, + AuthorID: req.AuthorID, + Type: req.Type, + RequestsInput: req.RequestsInput, + Metadata: req.Metadata, + ACPSessionID: req.ACPSessionID, + AgentSessionID: req.AgentSessionID, + }) + if err != nil { + return nil, err + } + return comment.ToAPI(), nil +} diff --git a/apps/backend/internal/task/dto/dto.go b/apps/backend/internal/task/dto/dto.go index 9420aa2a52..41965a57d7 100644 --- a/apps/backend/internal/task/dto/dto.go +++ b/apps/backend/internal/task/dto/dto.go @@ -119,6 +119,11 @@ type BoardSnapshotDTO struct { Tasks []TaskDTO `json:"tasks"` } +type ListCommentsResponse struct { + Comments []*v1.Comment `json:"comments"` + Total int `json:"total"` +} + type ListBoardsResponse struct { Boards []BoardDTO `json:"boards"` Total int `json:"total"` diff --git a/apps/backend/internal/task/dto/requests.go b/apps/backend/internal/task/dto/requests.go index e505ead270..82f6a18dd8 100644 --- a/apps/backend/internal/task/dto/requests.go +++ b/apps/backend/internal/task/dto/requests.go @@ -260,6 +260,22 @@ type DeleteTaskRequest struct { ID string } +type ListCommentsRequest struct { + TaskID string +} + +type CreateCommentRequest struct { + TaskID string + Content string + AuthorType string + AuthorID string + Type string + RequestsInput bool + Metadata map[string]interface{} + ACPSessionID string + AgentSessionID string +} + type MoveTaskRequest struct { ID string BoardID string diff --git a/apps/backend/internal/task/handlers/comment_handlers.go b/apps/backend/internal/task/handlers/comment_handlers.go index 90098b6ba1..ddc42e8911 100644 --- a/apps/backend/internal/task/handlers/comment_handlers.go +++ b/apps/backend/internal/task/handlers/comment_handlers.go @@ -2,12 +2,14 @@ package handlers import ( "context" + "net/http" "github.com/gin-gonic/gin" "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/task/service" + "github.com/kandev/kandev/internal/task/controller" + "github.com/kandev/kandev/internal/task/dto" v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" ) @@ -25,31 +27,54 @@ type OrchestratorService interface { // CommentHandlers handles WebSocket requests for comments type CommentHandlers struct { - svc *service.Service + commentController *controller.CommentController + taskController *controller.TaskController orchestrator OrchestratorService logger *logger.Logger } // NewCommentHandlers creates a new CommentHandlers instance -func NewCommentHandlers(svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) *CommentHandlers { +func NewCommentHandlers(commentCtrl *controller.CommentController, taskCtrl *controller.TaskController, orchestrator OrchestratorService, log *logger.Logger) *CommentHandlers { return &CommentHandlers{ - svc: svc, + commentController: commentCtrl, + taskController: taskCtrl, orchestrator: orchestrator, logger: log.WithFields(zap.String("component", "task-comment-handlers")), } } // RegisterCommentRoutes registers comment WebSocket handlers -func RegisterCommentRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) { - handlers := NewCommentHandlers(svc, orchestrator, log) +func RegisterCommentRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, commentCtrl *controller.CommentController, taskCtrl *controller.TaskController, orchestrator OrchestratorService, log *logger.Logger) { + handlers := NewCommentHandlers(commentCtrl, taskCtrl, orchestrator, log) + handlers.registerHTTP(router) handlers.registerWS(dispatcher) } +func (h *CommentHandlers) registerHTTP(router *gin.Engine) { + api := router.Group("/api/v1") + api.GET("/tasks/:id/comments", h.httpListComments) +} + func (h *CommentHandlers) registerWS(dispatcher *ws.Dispatcher) { dispatcher.RegisterFunc(ws.ActionCommentAdd, h.wsAddComment) dispatcher.RegisterFunc(ws.ActionCommentList, h.wsListComments) } +func (h *CommentHandlers) httpListComments(c *gin.Context) { + taskID := c.Param("id") + if taskID == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "task id is required"}) + return + } + resp, err := h.commentController.ListComments(c.Request.Context(), dto.ListCommentsRequest{TaskID: taskID}) + if err != nil { + h.logger.Error("failed to list comments", zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list comments"}) + return + } + c.JSON(http.StatusOK, resp) +} + // WS handlers type wsAddCommentRequest struct { @@ -71,7 +96,7 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w } // Get the current task state to determine if we need to transition - task, err := h.svc.GetTask(ctx, req.TaskID) + task, err := h.taskController.GetTask(ctx, dto.GetTaskRequest{ID: req.TaskID}) if err != nil { h.logger.Error("failed to get task", zap.String("task_id", req.TaskID), zap.Error(err)) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get task", nil) @@ -82,7 +107,8 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w // REVIEW → (user sends follow-up) → IN_PROGRESS // The transition back to REVIEW happens in handleAgentReady when the agent finishes if task.State == v1.TaskStateReview { - if _, err := h.svc.UpdateTaskState(ctx, req.TaskID, v1.TaskStateInProgress); err != nil { + nextState := v1.TaskStateInProgress + if _, err := h.taskController.UpdateTask(ctx, dto.UpdateTaskRequest{ID: req.TaskID, State: &nextState}); err != nil { h.logger.Error("failed to transition task from REVIEW to IN_PROGRESS", zap.String("task_id", req.TaskID), zap.Error(err)) @@ -93,7 +119,7 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w } } - comment, err := h.svc.CreateComment(ctx, &service.CreateCommentRequest{ + comment, err := h.commentController.CreateComment(ctx, dto.CreateCommentRequest{ TaskID: req.TaskID, Content: req.Content, AuthorType: "user", @@ -132,12 +158,11 @@ func (h *CommentHandlers) wsListComments(ctx context.Context, msg *ws.Message) ( return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) } - comments, err := h.svc.ListComments(ctx, req.TaskID) + resp, err := h.commentController.ListComments(ctx, dto.ListCommentsRequest{TaskID: req.TaskID}) if err != nil { h.logger.Error("failed to list comments", zap.Error(err)) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list comments", nil) } - return ws.NewResponse(msg.ID, msg.Action, comments) + return ws.NewResponse(msg.ID, msg.Action, resp) } - diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index 1d6fa7d880..eec56f5715 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -46,6 +46,7 @@ const DEFAULT_HORIZONTAL_LAYOUT: [number, number] = [75, 25]; export default function TaskPage({ task: initialTask }: TaskPageClientProps) { const store = useAppStoreApi(); + const commentsState = useAppStore((state) => state.comments); const [isMounted, setIsMounted] = useState(false); const [horizontalLayout, setHorizontalLayout] = useState<[number, number]>( getLocalStorage('task-layout-horizontal', DEFAULT_HORIZONTAL_LAYOUT) @@ -118,6 +119,10 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { // Clear git status when switching tasks to avoid showing stale data store.getState().clearGitStatus(); + if (commentsState.taskId === task.id) { + return; + } + const fetchComments = async () => { const client = getWebSocketClient(); if (!client) return; @@ -126,9 +131,9 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { store.getState().setCommentsLoading(true); try { - const response = await client.request('comment.list', { task_id: task.id }, 10000); + const response = await client.request<{ comments: Comment[] }>('comment.list', { task_id: task.id }, 10000); console.log('[API] comment.list response:', JSON.stringify(response, null, 2)); - store.getState().setComments(task.id, response); + store.getState().setComments(task.id, response.comments ?? []); } catch (error) { console.error('Failed to fetch comments:', error); store.getState().setComments(task.id, []); @@ -138,7 +143,7 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { }; fetchComments(); - }, [task?.id, store]); + }, [commentsState.taskId, task?.id, store]); // Subscribe to task for real-time updates useEffect(() => { diff --git a/apps/web/app/task/[id]/page.tsx b/apps/web/app/task/[id]/page.tsx index fbf8deafcb..0a6c219adc 100644 --- a/apps/web/app/task/[id]/page.tsx +++ b/apps/web/app/task/[id]/page.tsx @@ -1,5 +1,5 @@ import { StateHydrator } from '@/components/state-hydrator'; -import { fetchTask } from '@/lib/http'; +import { fetchTask, listTaskComments } from '@/lib/http'; import type { Task } from '@/lib/types/http'; import { taskToState } from '@/lib/ssr/mapper'; import TaskPageClient from './page-client'; @@ -7,11 +7,13 @@ import TaskPageClient from './page-client'; export default async function TaskPage({ params }: { params: Promise<{ id: string }> }) { let initialState: ReturnType | null = null; let task: Task | null = null; + let comments: Awaited> | null = null; try { const { id } = await params; task = await fetchTask(id, { cache: 'no-store' }); - initialState = taskToState(task); + comments = await listTaskComments(id, { cache: 'no-store' }); + initialState = taskToState(task, comments.comments); } catch { initialState = null; task = null; diff --git a/apps/web/lib/http/index.ts b/apps/web/lib/http/index.ts index a8a76c6955..3809920b22 100644 --- a/apps/web/lib/http/index.ts +++ b/apps/web/lib/http/index.ts @@ -8,6 +8,8 @@ import type { ListExecutorsResponse, ListRepositoriesResponse, ListWorkspacesResponse, + Comment, + ListCommentsResponse, RepositoryBranchesResponse, Task, UserSettingsResponse, @@ -160,6 +162,10 @@ export async function fetchTask(taskId: string, options?: ApiRequestOptions) { return fetchJson(`/api/v1/tasks/${taskId}`, options); } +export async function listTaskComments(taskId: string, options?: ApiRequestOptions) { + return fetchJson(`/api/v1/tasks/${taskId}/comments`, options); +} + export async function listExecutors(options?: ApiRequestOptions): Promise { return fetchJson('/api/v1/executors', options); } diff --git a/apps/web/lib/ssr/mapper.ts b/apps/web/lib/ssr/mapper.ts index cc839de28e..a19f64d7a8 100644 --- a/apps/web/lib/ssr/mapper.ts +++ b/apps/web/lib/ssr/mapper.ts @@ -1,5 +1,5 @@ import type { AppState, KanbanState } from '@/lib/state/store'; -import type { BoardSnapshot, Task } from '@/lib/types/http'; +import type { BoardSnapshot, Comment, Task } from '@/lib/types/http'; type KanbanTask = KanbanState['tasks'][number]; @@ -34,10 +34,17 @@ export function snapshotToState(snapshot: BoardSnapshot): Partial { }; } -export function taskToState(task: Task): Partial { +export function taskToState(task: Task, comments?: Comment[]): Partial { return { tasks: { activeTaskId: task.id, }, + comments: comments + ? { + taskId: task.id, + items: comments, + isLoading: false, + } + : undefined, }; } diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index 58aaa67c7f..cbf4e07a09 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -212,6 +212,11 @@ export type ListEnvironmentsResponse = { total: number; }; +export type ListCommentsResponse = { + comments: Comment[]; + total: number; +}; + export type CommentAuthorType = 'user' | 'agent'; export type CommentType = 'message' | 'content' | 'tool_call' | 'progress' | 'error' | 'status';