From 4560901d352de7553af337028f7d2e4923defe0b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 15 Jan 2026 23:39:07 +0000 Subject: [PATCH] feat: flaky resume sessions --- apps/backend/cmd/kandev/main.go | 127 +++++++------ apps/backend/dev/seed.sql | 2 +- apps/backend/internal/agent/agentctl/acp.go | 40 ++++ .../internal/agent/lifecycle/manager.go | 139 +++++++++----- apps/backend/internal/agentctl/api/acp.go | 58 +++++- .../internal/agentctl/api/control_server.go | 54 +++++- apps/backend/internal/agentctl/api/server.go | 1 + apps/backend/internal/events/types.go | 13 +- .../orchestrator/controller/controller.go | 25 +++ apps/backend/internal/orchestrator/dto/dto.go | 21 ++- .../orchestrator/executor/executor.go | 155 ++++++++++++++- .../orchestrator/handlers/handlers.go | 32 ++++ apps/backend/internal/orchestrator/service.go | 143 ++++++++++++-- .../internal/orchestrator/watcher/watcher.go | 65 ++++++- apps/backend/internal/task/controller/task.go | 15 ++ apps/backend/internal/task/dto/dto.go | 60 ++++++ apps/backend/internal/task/dto/requests.go | 4 + .../task/handlers/message_handlers.go | 30 ++- .../internal/task/handlers/task_handlers.go | 32 ++++ apps/backend/internal/task/service/service.go | 13 +- apps/backend/pkg/api/v1/task.go | 8 +- apps/backend/pkg/websocket/actions.go | 2 + apps/web/app/task/[id]/[sessionId]/page.tsx | 33 ++++ apps/web/app/task/[id]/page-client.tsx | 19 +- apps/web/app/task/[id]/page.tsx | 7 +- apps/web/components/kanban-board.tsx | 61 +++++- apps/web/components/kanban-card.tsx | 3 +- apps/web/components/kanban-with-preview.tsx | 68 ++++++- apps/web/components/task-preview-panel.tsx | 178 ++++++------------ .../components/task/chat/message-renderer.tsx | 6 +- .../task/chat/messages/chat-message.tsx | 2 +- .../task/chat/messages/status-message.tsx | 4 +- .../task/chat/messages/thinking-message.tsx | 2 +- .../task/chat/messages/todo-message.tsx | 2 +- .../task/chat/messages/tool-call-message.tsx | 2 +- .../web/components/task/sessions-dropdown.tsx | 118 +++++++++--- apps/web/components/task/task-top-bar.tsx | 11 +- apps/web/hooks/use-task-chat-session.ts | 74 ++++++++ apps/web/lib/config.ts | 3 +- apps/web/lib/http/index.ts | 5 + apps/web/lib/types/http.ts | 31 +++ 41 files changed, 1331 insertions(+), 337 deletions(-) create mode 100644 apps/web/app/task/[id]/[sessionId]/page.tsx create mode 100644 apps/web/hooks/use-task-chat-session.ts diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 5f34f4e6bc..b18f376b99 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -441,15 +441,15 @@ func main() { for _, message := range messages { action := ws.ActionMessageAdded payload := map[string]interface{}{ - "message_id": message.ID, + "message_id": message.ID, "task_session_id": message.TaskSessionID, - "task_id": message.TaskID, - "author_type": string(message.AuthorType), - "author_id": message.AuthorID, - "content": message.Content, - "type": string(message.Type), - "requests_input": message.RequestsInput, - "created_at": message.CreatedAt.Format(time.RFC3339), + "task_id": message.TaskID, + "author_type": string(message.AuthorType), + "author_id": message.AuthorID, + "content": message.Content, + "type": string(message.Type), + "requests_input": message.RequestsInput, + "created_at": message.CreatedAt.Format(time.RFC3339), } if message.Metadata != nil { payload["metadata"] = message.Metadata @@ -532,11 +532,11 @@ func main() { // Create a message from the agent messageRecord, err := taskSvc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ TaskSessionID: session.ID, - TaskID: taskID, - Content: message, - AuthorType: "agent", - AuthorID: agentID, - RequestsInput: true, + TaskID: taskID, + Content: message, + AuthorType: "agent", + AuthorID: agentID, + RequestsInput: true, }) if err != nil { log.Error("failed to create agent message", @@ -547,10 +547,10 @@ func main() { // Broadcast message.added notification to task subscribers notification, _ := ws.NewNotification(ws.ActionMessageAdded, map[string]interface{}{ - "task_id": taskID, + "task_id": taskID, "task_session_id": session.ID, - "message": messageRecord.ToAPI(), - "requests_input": true, + "message": messageRecord.ToAPI(), + "requests_input": true, }) gateway.Hub.BroadcastToTask(taskID, notification) @@ -573,22 +573,22 @@ func main() { // Subscribe to message.added events and broadcast to WebSocket subscribers _, err = eventBus.Subscribe(events.MessageAdded, func(ctx context.Context, event *bus.Event) error { data := event.Data - agentSessionID, _ := data["agent_session_id"].(string) + taskSessionID, _ := data["task_session_id"].(string) taskID, _ := data["task_id"].(string) - if agentSessionID == "" { + if taskSessionID == "" { return nil } payload := map[string]interface{}{ - "task_id": taskID, - "task_session_id": agentSessionID, - "message_id": data["message_id"], - "author_type": data["author_type"], - "author_id": data["author_id"], - "content": data["content"], - "type": data["type"], - "requests_input": data["requests_input"], - "created_at": data["created_at"], + "task_id": taskID, + "task_session_id": taskSessionID, + "message_id": data["message_id"], + "author_type": data["author_type"], + "author_id": data["author_id"], + "content": data["content"], + "type": data["type"], + "requests_input": data["requests_input"], + "created_at": data["created_at"], } if metadata, ok := data["metadata"]; ok && metadata != nil { payload["metadata"] = metadata @@ -610,22 +610,22 @@ func main() { // Subscribe to message.updated events and broadcast to WebSocket subscribers _, err = eventBus.Subscribe(events.MessageUpdated, func(ctx context.Context, event *bus.Event) error { data := event.Data - agentSessionID, _ := data["agent_session_id"].(string) + taskSessionID, _ := data["task_session_id"].(string) taskID, _ := data["task_id"].(string) - if agentSessionID == "" { - log.Warn("message.updated event has no agent_session_id, skipping") + if taskSessionID == "" { + log.Warn("message.updated event has no task_session_id, skipping") return nil } payload := map[string]interface{}{ - "message_id": data["message_id"], - "task_session_id": agentSessionID, - "task_id": taskID, - "author_type": data["author_type"], - "author_id": data["author_id"], - "content": data["content"], - "type": data["type"], - "requests_input": data["requests_input"], - "created_at": data["created_at"], + "message_id": data["message_id"], + "task_session_id": taskSessionID, + "task_id": taskID, + "author_type": data["author_type"], + "author_id": data["author_id"], + "content": data["content"], + "type": data["type"], + "requests_input": data["requests_input"], + "created_at": data["created_at"], } if metadata, ok := data["metadata"]; ok && metadata != nil { payload["metadata"] = metadata @@ -644,7 +644,7 @@ func main() { log.Info("Subscribed to message.updated events for WebSocket broadcasting") } - // Subscribe to agent_session.state_changed events and broadcast to task subscribers + // Subscribe to task_session.state_changed events and broadcast to task subscribers _, err = eventBus.Subscribe(events.TaskSessionStateChanged, func(ctx context.Context, event *bus.Event) error { data := event.Data taskID, _ := data["task_id"].(string) @@ -653,16 +653,16 @@ func main() { } notification, err := ws.NewNotification(ws.ActionTaskSessionStateChanged, data) if err != nil { - log.Error("Failed to create agent_session.state_changed notification", zap.Error(err)) + log.Error("Failed to create task_session.state_changed notification", zap.Error(err)) return nil } gateway.Hub.BroadcastToTask(taskID, notification) return nil }) if err != nil { - log.Error("Failed to subscribe to agent_session.state_changed events", zap.Error(err)) + log.Error("Failed to subscribe to task_session.state_changed events", zap.Error(err)) } else { - log.Info("Subscribed to agent_session.state_changed events for WebSocket broadcasting") + log.Info("Subscribed to task_session.state_changed events for WebSocket broadcasting") } // Subscribe to git status events and broadcast to WebSocket subscribers @@ -862,6 +862,8 @@ func (a *lifecycleAdapter) LaunchAgent(ctx context.Context, req *executor.Launch AgentProfileID: req.AgentProfileID, WorkspacePath: req.RepositoryURL, // May be empty - lifecycle manager handles this TaskDescription: req.TaskDescription, + Env: req.Env, + ACPSessionID: req.ACPSessionID, Metadata: req.Metadata, // Worktree configuration for concurrent agent execution UseWorktree: req.UseWorktree, @@ -1015,6 +1017,11 @@ func (a *orchestratorAdapter) PromptTask(ctx context.Context, taskID string, pro }, nil } +func (a *orchestratorAdapter) ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error { + _, err := a.svc.ResumeTaskSession(ctx, taskID, taskSessionID) + return err +} + // messageCreatorAdapter adapts the task service to the orchestrator.MessageCreator interface type messageCreatorAdapter struct { svc *taskservice.Service @@ -1024,9 +1031,9 @@ type messageCreatorAdapter struct { func (a *messageCreatorAdapter) CreateAgentMessage(ctx context.Context, taskID, content, agentSessionID string) error { _, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ TaskSessionID: agentSessionID, - TaskID: taskID, - Content: content, - AuthorType: "agent", + TaskID: taskID, + Content: content, + AuthorType: "agent", }) return err } @@ -1035,9 +1042,9 @@ func (a *messageCreatorAdapter) CreateAgentMessage(ctx context.Context, taskID, func (a *messageCreatorAdapter) CreateUserMessage(ctx context.Context, taskID, content, agentSessionID string) error { _, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ TaskSessionID: agentSessionID, - TaskID: taskID, - Content: content, - AuthorType: "user", + TaskID: taskID, + Content: content, + AuthorType: "user", }) return err } @@ -1062,11 +1069,11 @@ func (a *messageCreatorAdapter) CreateToolCallMessage(ctx context.Context, taskI _, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ TaskSessionID: agentSessionID, - TaskID: taskID, - Content: title, - AuthorType: "agent", - Type: "tool_call", - Metadata: metadata, + TaskID: taskID, + Content: title, + AuthorType: "agent", + Type: "tool_call", + Metadata: metadata, }) return err } @@ -1080,12 +1087,12 @@ func (a *messageCreatorAdapter) UpdateToolCallMessage(ctx context.Context, taskI func (a *messageCreatorAdapter) CreateSessionMessage(ctx context.Context, taskID, content, agentSessionID, messageType string, metadata map[string]interface{}, requestsInput bool) error { _, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ TaskSessionID: agentSessionID, - TaskID: taskID, - Content: content, - AuthorType: "agent", - Type: messageType, - Metadata: metadata, - RequestsInput: requestsInput, + TaskID: taskID, + Content: content, + AuthorType: "agent", + Type: messageType, + Metadata: metadata, + RequestsInput: requestsInput, }) return err } diff --git a/apps/backend/dev/seed.sql b/apps/backend/dev/seed.sql index 5ddef587d3..1fc4edb244 100644 --- a/apps/backend/dev/seed.sql +++ b/apps/backend/dev/seed.sql @@ -66,7 +66,7 @@ VALUES ('task-103', 'dev-board-2', 'col-planned', 'Agent type marketplace', 'Browse and install different agent types', 'TODO', 6, 0, '', '$KANDEV_REPO_PATH', '', '{}', datetime('now'), datetime('now')); -- Sample comments on tasks -INSERT OR REPLACE INTO task_comments (id, task_id, author_type, author_id, content, type, metadata, requests_input, agent_session_id, created_at) +INSERT OR REPLACE INTO task_comments (id, task_id, author_type, author_id, content, type, metadata, requests_input, task_session_id, created_at) VALUES -- Comments on task-005 (Agent session persistence) ('comment-001', 'task-005', 'user', 'dev-user', 'Starting work on database schema for agent sessions', 'message', '{}', 0, '', datetime('now', '-2 hours')), diff --git a/apps/backend/internal/agent/agentctl/acp.go b/apps/backend/internal/agent/agentctl/acp.go index cdfd6de47d..6b8e811a47 100644 --- a/apps/backend/internal/agent/agentctl/acp.go +++ b/apps/backend/internal/agent/agentctl/acp.go @@ -111,6 +111,46 @@ func (c *Client) NewSession(ctx context.Context, cwd string) (string, error) { return result.SessionID, nil } +// LoadSession resumes an existing ACP session +func (c *Client) LoadSession(ctx context.Context, sessionID string) error { + reqBody := struct { + SessionID string `json:"session_id"` + }{SessionID: sessionID} + + body, err := json.Marshal(reqBody) + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/acp/session/load", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer func() { + if err := resp.Body.Close(); err != nil { + c.logger.Debug("failed to close load session response body", zap.Error(err)) + } + }() + + var result struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return err + } + if !result.Success { + return fmt.Errorf("load session failed: %s", result.Error) + } + return nil +} + // PromptResponse contains the response from a prompt call type PromptResponse struct { Success bool `json:"success"` diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index d360aeb9ad..95081d8b30 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -34,15 +34,16 @@ type AgentInstance struct { TaskID string AgentProfileID string ContainerID string - ContainerIP string // IP address of the container for agentctl communication - WorkspacePath string // Path to the workspace (worktree or repository path) - Status v1.AgentStatus - StartedAt time.Time - FinishedAt *time.Time - ExitCode *int - ErrorMessage string - Progress int - Metadata map[string]interface{} + ContainerIP string // IP address of the container for agentctl communication + WorkspacePath string // Path to the workspace (worktree or repository path) + ACPSessionID string // ACP session ID to resume, if available + Status v1.AgentStatus + StartedAt time.Time + FinishedAt *time.Time + ExitCode *int + ErrorMessage string + Progress int + Metadata map[string]interface{} // agentctl client for this instance agentctl *agentctl.Client @@ -66,11 +67,12 @@ func (ai *AgentInstance) GetAgentCtlClient() *agentctl.Client { // LaunchRequest contains parameters for launching an agent type LaunchRequest struct { TaskID string - TaskTitle string // Human-readable task title for semantic worktree naming + TaskTitle string // Human-readable task title for semantic worktree naming AgentProfileID string - WorkspacePath string // Host path to workspace (original repository path) - TaskDescription string // Task description to send via ACP prompt - Env map[string]string // Additional env vars + WorkspacePath string // Host path to workspace (original repository path) + TaskDescription string // Task description to send via ACP prompt + Env map[string]string // Additional env vars + ACPSessionID string // ACP session ID to resume, if available Metadata map[string]interface{} // Worktree configuration @@ -294,9 +296,9 @@ func (m *Manager) reconcileWithDocker(ctx context.Context) ([]RecoveredInstance, TaskID: taskID, AgentProfileID: agentProfileID, ContainerID: ctr.ID, - ContainerIP: containerIP, - Status: v1.AgentStatusRunning, - StartedAt: time.Now(), // We don't know exact start time + ContainerIP: containerIP, + Status: v1.AgentStatusRunning, + StartedAt: time.Now(), // We don't know exact start time } // Create agentctl client @@ -555,6 +557,7 @@ func (m *Manager) launchDocker(ctx context.Context, instanceID string, req *Laun ContainerID: containerID, ContainerIP: containerIP, WorkspacePath: "/workspace", // Docker mounts workspace to /workspace + ACPSessionID: req.ACPSessionID, Status: v1.AgentStatusRunning, StartedAt: time.Now(), Progress: 0, @@ -578,6 +581,9 @@ func (m *Manager) launchStandalone(ctx context.Context, instanceID string, req * if agentConfig.ModelFlag != "" && profileInfo != nil && profileInfo.Model != "" { agentCommand = agentCommand + " " + agentConfig.ModelFlag + " " + profileInfo.Model } + if req.ACPSessionID != "" && profileInfo != nil && profileInfo.AgentName == "auggie" { + agentCommand = agentCommand + " --resume " + req.ACPSessionID + } // If empty, agentctl will use its default (auggie --acp) // Create instance via control API @@ -585,8 +591,8 @@ func (m *Manager) launchStandalone(ctx context.Context, instanceID string, req * ID: instanceID, WorkspacePath: req.WorkspacePath, AgentCommand: agentCommand, - Protocol: string(agentConfig.Protocol), // Pass protocol from registry - WorkspaceFlag: agentConfig.WorkspaceFlag, // CLI flag for workspace path + Protocol: string(agentConfig.Protocol), // Pass protocol from registry + WorkspaceFlag: agentConfig.WorkspaceFlag, // CLI flag for workspace path Env: env, AutoStart: false, // We'll start via ACP initialize flow } @@ -622,6 +628,7 @@ func (m *Manager) launchStandalone(ctx context.Context, instanceID string, req * TaskID: req.TaskID, AgentProfileID: req.AgentProfileID, WorkspacePath: req.WorkspacePath, + ACPSessionID: req.ACPSessionID, Status: v1.AgentStatusRunning, StartedAt: time.Now(), Progress: 0, @@ -772,21 +779,48 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta zap.String("agent_name", agentName), zap.String("agent_version", agentVersion)) - // 2. Create new session - m.logger.Info("sending ACP session/new request", - zap.String("instance_id", instance.ID), - zap.String("workspace_path", instance.WorkspacePath)) - - sessionID, err := instance.agentctl.NewSession(ctx, instance.WorkspacePath) - if err != nil { - m.logger.Error("ACP session/new failed", + // 2. Create or resume ACP session + sessionID := "" + if instance.ACPSessionID != "" && agentName != "auggie" { + m.logger.Info("sending ACP session/load request", zap.String("instance_id", instance.ID), - zap.Error(err)) - return fmt.Errorf("session/new failed: %w", err) + zap.String("session_id", instance.ACPSessionID)) + + if err := instance.agentctl.LoadSession(ctx, instance.ACPSessionID); err != nil { + m.logger.Error("ACP session/load failed", + zap.String("instance_id", instance.ID), + zap.Error(err)) + return fmt.Errorf("session/load failed: %w", err) + } + sessionID = instance.ACPSessionID + m.logger.Info("ACP session loaded", + zap.String("instance_id", instance.ID), + zap.String("session_id", sessionID)) + } else { + if instance.ACPSessionID != "" && agentName == "auggie" { + m.logger.Info("skipping ACP session/load for auggie (resume handled via CLI)", + zap.String("instance_id", instance.ID), + zap.String("session_id", instance.ACPSessionID)) + } + m.logger.Info("sending ACP session/new request", + zap.String("instance_id", instance.ID), + zap.String("workspace_path", instance.WorkspacePath)) + + newSessionID, err := instance.agentctl.NewSession(ctx, instance.WorkspacePath) + if err != nil { + m.logger.Error("ACP session/new failed", + zap.String("instance_id", instance.ID), + zap.Error(err)) + return fmt.Errorf("session/new failed: %w", err) + } + sessionID = newSessionID + m.logger.Info("ACP session created", + zap.String("instance_id", instance.ID), + zap.String("session_id", sessionID)) } - m.logger.Info("ACP session created", - zap.String("instance_id", instance.ID), - zap.String("session_id", sessionID)) + + instance.ACPSessionID = sessionID + m.publishACPSessionCreated(instance, sessionID) // 3. Set up updates stream to receive session notifications // Use a ready channel to signal when the stream is connected @@ -1007,16 +1041,16 @@ func (m *Manager) publishPermissionRequest(instance *AgentInstance, notification } data := map[string]interface{}{ - "type": "permission_request", - "task_id": instance.TaskID, + "type": "permission_request", + "task_id": instance.TaskID, "agent_instance_id": instance.ID, - "pending_id": notification.PendingID, - "session_id": notification.SessionID, - "tool_call_id": notification.ToolCallID, - "title": notification.Title, - "options": options, - "created_at": notification.CreatedAt, - "timestamp": time.Now().UTC().Format(time.RFC3339), + "pending_id": notification.PendingID, + "session_id": notification.SessionID, + "tool_call_id": notification.ToolCallID, + "title": notification.Title, + "options": options, + "created_at": notification.CreatedAt, + "timestamp": time.Now().UTC().Format(time.RFC3339), } event := bus.NewEvent(events.ACPMessage, "agent-manager", data) @@ -1426,8 +1460,6 @@ func (m *Manager) updateInstanceProgress(instanceID string, progress int) { instance.Progress = progress } - - // updateInstanceError updates an instance with an error func (m *Manager) updateInstanceError(instanceID, errorMsg string) { m.mu.Lock() @@ -1541,6 +1573,9 @@ func (m *Manager) buildContainerConfig(instanceID string, req *LaunchRequest, ag if agentConfig.ModelFlag != "" && profileInfo != nil && profileInfo.Model != "" { cmd = append(cmd, agentConfig.ModelFlag, profileInfo.Model) } + if req.ACPSessionID != "" && profileInfo != nil && profileInfo.AgentName == "auggie" { + cmd = append(cmd, "--resume", req.ACPSessionID) + } return docker.ContainerConfig{ Name: containerName, @@ -1987,6 +2022,26 @@ func (m *Manager) publishEvent(ctx context.Context, eventType string, instance * } } +func (m *Manager) publishACPSessionCreated(instance *AgentInstance, sessionID string) { + if m.eventBus == nil || sessionID == "" { + return + } + + data := map[string]interface{}{ + "task_id": instance.TaskID, + "agent_instance_id": instance.ID, + "acp_session_id": sessionID, + } + + event := bus.NewEvent(events.AgentACPSessionCreated, "agent-manager", data) + if err := m.eventBus.Publish(context.Background(), events.AgentACPSessionCreated, event); err != nil { + m.logger.Error("failed to publish ACP session event", + zap.String("event_type", events.AgentACPSessionCreated), + zap.String("instance_id", instance.ID), + zap.Error(err)) + } +} + // cleanupLoop runs periodic cleanup of stale containers func (m *Manager) cleanupLoop(ctx context.Context) { defer m.wg.Done() diff --git a/apps/backend/internal/agentctl/api/acp.go b/apps/backend/internal/agentctl/api/acp.go index 9a14456385..0a5c0bd3c1 100644 --- a/apps/backend/internal/agentctl/api/acp.go +++ b/apps/backend/internal/agentctl/api/acp.go @@ -128,6 +128,62 @@ func (s *Server) handleACPNewSession(c *gin.Context) { }) } +// LoadSessionRequest is a request to load an existing ACP session +type LoadSessionRequest struct { + SessionID string `json:"session_id"` +} + +// LoadSessionResponse is the response to a load session call +type LoadSessionResponse struct { + Success bool `json:"success"` + SessionID string `json:"session_id,omitempty"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handleACPLoadSession(c *gin.Context) { + var req LoadSessionRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, LoadSessionResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + if req.SessionID == "" { + c.JSON(http.StatusBadRequest, LoadSessionResponse{ + Success: false, + Error: "session_id is required", + }) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + adapter := s.procMgr.GetAdapter() + if adapter == nil { + c.JSON(http.StatusServiceUnavailable, LoadSessionResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + if err := adapter.LoadSession(ctx, req.SessionID); err != nil { + s.logger.Error("load session failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, LoadSessionResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, LoadSessionResponse{ + Success: true, + SessionID: req.SessionID, + }) +} + // PromptRequest is a request to send a prompt to the agent type PromptRequest struct { Text string `json:"text"` // Simple text prompt @@ -221,8 +277,6 @@ func (s *Server) handleACPStreamWS(c *gin.Context) { } } - - // PermissionRespondRequest is a request to respond to a permission request type PermissionRespondRequest struct { PendingID string `json:"pending_id" binding:"required"` diff --git a/apps/backend/internal/agentctl/api/control_server.go b/apps/backend/internal/agentctl/api/control_server.go index 470942e154..aae84f02dc 100644 --- a/apps/backend/internal/agentctl/api/control_server.go +++ b/apps/backend/internal/agentctl/api/control_server.go @@ -25,10 +25,10 @@ type ControlServer struct { router *gin.Engine // Single-instance mode: embedded process manager and config - singleMode bool - singleCfg *config.Config + singleMode bool + singleCfg *config.Config singleProcMgr *process.Manager - upgrader websocket.Upgrader + upgrader websocket.Upgrader } // NewControlServer creates a new ControlServer for multi-instance management. @@ -106,6 +106,7 @@ func (m *ControlServer) setupSingleInstanceRoutes(api *gin.RouterGroup) { // ACP high-level methods api.POST("/acp/initialize", m.handleACPInitialize) api.POST("/acp/session/new", m.handleACPNewSession) + api.POST("/acp/session/load", m.handleACPLoadSession) api.POST("/acp/prompt", m.handleACPPrompt) api.GET("/acp/stream", m.handleACPStreamWS) @@ -259,7 +260,6 @@ func (m *ControlServer) handleStop(c *gin.Context) { }) } - // ACP handlers for single-instance mode func (m *ControlServer) handleACPInitialize(c *gin.Context) { @@ -346,6 +346,50 @@ func (m *ControlServer) handleACPNewSession(c *gin.Context) { }) } +func (m *ControlServer) handleACPLoadSession(c *gin.Context) { + var req LoadSessionRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, LoadSessionResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + if req.SessionID == "" { + c.JSON(http.StatusBadRequest, LoadSessionResponse{ + Success: false, + Error: "session_id is required", + }) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + adapter := m.singleProcMgr.GetAdapter() + if adapter == nil { + c.JSON(http.StatusServiceUnavailable, LoadSessionResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + if err := adapter.LoadSession(ctx, req.SessionID); err != nil { + m.logger.Error("load session failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, LoadSessionResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, LoadSessionResponse{ + Success: true, + SessionID: req.SessionID, + }) +} + func (m *ControlServer) handleACPPrompt(c *gin.Context) { var req PromptRequest if err := c.ShouldBindJSON(&req); err != nil { @@ -393,7 +437,6 @@ func (m *ControlServer) handleACPPrompt(c *gin.Context) { }) } - func (m *ControlServer) handleACPStreamWS(c *gin.Context) { conn, err := m.upgrader.Upgrade(c.Writer, c.Request, nil) if err != nil { @@ -508,7 +551,6 @@ func (m *ControlServer) handlePermissionRespond(c *gin.Context) { }) } - // Workspace handlers for single-instance mode func (m *ControlServer) handleGitStatusStreamWS(c *gin.Context) { diff --git a/apps/backend/internal/agentctl/api/server.go b/apps/backend/internal/agentctl/api/server.go index ae552a8c0a..1833ab7a89 100644 --- a/apps/backend/internal/agentctl/api/server.go +++ b/apps/backend/internal/agentctl/api/server.go @@ -66,6 +66,7 @@ func (s *Server) setupRoutes() { // ACP high-level methods (using acp-go-sdk) api.POST("/acp/initialize", s.handleACPInitialize) api.POST("/acp/session/new", s.handleACPNewSession) + api.POST("/acp/session/load", s.handleACPLoadSession) api.POST("/acp/prompt", s.handleACPPrompt) api.GET("/acp/stream", s.handleACPStreamWS) diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 655f1058ce..408d036f43 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -76,12 +76,13 @@ const ( // Event types for agents const ( - AgentStarted = "agent.started" - AgentRunning = "agent.running" - AgentReady = "agent.ready" // Agent finished prompt, ready for follow-up - AgentCompleted = "agent.completed" - AgentFailed = "agent.failed" - AgentStopped = "agent.stopped" + AgentStarted = "agent.started" + AgentRunning = "agent.running" + AgentReady = "agent.ready" // Agent finished prompt, ready for follow-up + AgentCompleted = "agent.completed" + AgentFailed = "agent.failed" + AgentStopped = "agent.stopped" + AgentACPSessionCreated = "agent.acp_session_created" ) // Event types for ACP messages diff --git a/apps/backend/internal/orchestrator/controller/controller.go b/apps/backend/internal/orchestrator/controller/controller.go index 10bdd5ac1c..f62193f034 100644 --- a/apps/backend/internal/orchestrator/controller/controller.go +++ b/apps/backend/internal/orchestrator/controller/controller.go @@ -86,6 +86,31 @@ func (c *Controller) StartTask(ctx context.Context, req dto.StartTaskRequest) (d return resp, nil } +// ResumeTaskSession resumes a specific task session execution. +func (c *Controller) ResumeTaskSession(ctx context.Context, req dto.ResumeTaskSessionRequest) (dto.ResumeTaskSessionResponse, error) { + execution, err := c.service.ResumeTaskSession(ctx, req.TaskID, req.TaskSessionID) + if err != nil { + return dto.ResumeTaskSessionResponse{}, err + } + + resp := dto.ResumeTaskSessionResponse{ + Success: true, + TaskID: execution.TaskID, + AgentInstanceID: execution.AgentInstanceID, + TaskSessionID: execution.SessionID, + State: string(execution.SessionState), + } + + if execution.WorktreePath != "" { + resp.WorktreePath = &execution.WorktreePath + } + if execution.WorktreeBranch != "" { + resp.WorktreeBranch = &execution.WorktreeBranch + } + + return resp, nil +} + // StopTask stops task execution func (c *Controller) StopTask(ctx context.Context, req dto.StopTaskRequest) (dto.SuccessResponse, error) { reason := req.Reason diff --git a/apps/backend/internal/orchestrator/dto/dto.go b/apps/backend/internal/orchestrator/dto/dto.go index 7b228f764a..d248326cd8 100644 --- a/apps/backend/internal/orchestrator/dto/dto.go +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -52,7 +52,24 @@ type StartTaskResponse struct { Success bool `json:"success"` TaskID string `json:"task_id"` AgentInstanceID string `json:"agent_instance_id"` - TaskSessionID string `json:"agent_session_id,omitempty"` + TaskSessionID string `json:"task_session_id,omitempty"` + State string `json:"state"` + WorktreePath *string `json:"worktree_path,omitempty"` + WorktreeBranch *string `json:"worktree_branch,omitempty"` +} + +// ResumeTaskSessionRequest is the payload for task.session.resume +type ResumeTaskSessionRequest struct { + TaskID string `json:"task_id"` + TaskSessionID string `json:"task_session_id"` +} + +// ResumeTaskSessionResponse is the response for task.session.resume +type ResumeTaskSessionResponse struct { + Success bool `json:"success"` + TaskID string `json:"task_id"` + AgentInstanceID string `json:"agent_instance_id"` + TaskSessionID string `json:"task_session_id,omitempty"` State string `json:"state"` WorktreePath *string `json:"worktree_path,omitempty"` WorktreeBranch *string `json:"worktree_branch,omitempty"` @@ -119,7 +136,7 @@ type TaskExecutionResponse struct { TaskID string `json:"task_id"` AgentInstanceID string `json:"agent_instance_id,omitempty"` AgentProfileID string `json:"agent_profile_id,omitempty"` - TaskSessionID string `json:"agent_session_id,omitempty"` + TaskSessionID string `json:"task_session_id,omitempty"` State string `json:"state,omitempty"` Progress int `json:"progress,omitempty"` StartedAt string `json:"started_at,omitempty"` diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 0dc32d5de2..563417d4c5 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -18,9 +18,10 @@ import ( // Common errors var ( - ErrMaxConcurrentReached = errors.New("maximum concurrent executions reached") - ErrNoAgentProfileID = errors.New("task has no agent_profile_id configured") - ErrExecutionNotFound = errors.New("execution not found") + ErrMaxConcurrentReached = errors.New("maximum concurrent executions reached") + ErrNoAgentProfileID = errors.New("task has no agent_profile_id configured") + ErrExecutionNotFound = errors.New("execution not found") + ErrExecutionAlreadyRunning = errors.New("execution already running") ) // PromptResult contains the result of a prompt operation @@ -69,6 +70,8 @@ type LaunchAgentRequest struct { TaskDescription string // Task description to send via ACP prompt Priority int Metadata map[string]interface{} + Env map[string]string + ACPSessionID string // ACP session ID to resume, if available // Worktree configuration for concurrent agent execution UseWorktree bool // Whether to use a Git worktree for isolation @@ -366,6 +369,152 @@ func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentP return execution, nil } +// ResumeSession restarts an existing task session using its stored worktree. +func (e *Executor) ResumeSession(ctx context.Context, task *v1.Task, session *models.TaskSession) (*TaskExecution, error) { + if session == nil { + return nil, ErrExecutionNotFound + } + + if !e.CanExecute() { + e.logger.Warn("max concurrent executions reached", + zap.Int("max", e.maxConcurrent), + zap.Int("current", e.ActiveCount())) + return nil, ErrMaxConcurrentReached + } + + if session.AgentProfileID == "" { + e.logger.Error("task session has no agent_profile_id configured", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID)) + return nil, ErrNoAgentProfileID + } + + if existing, ok := e.GetExecutionWithContext(ctx, task.ID); ok && existing != nil { + return nil, ErrExecutionAlreadyRunning + } + + req := &LaunchAgentRequest{ + TaskID: task.ID, + TaskTitle: task.Title, + AgentProfileID: session.AgentProfileID, + TaskDescription: "", + Priority: task.Priority, + } + + metadata := map[string]interface{}{} + if session.Metadata != nil { + for key, value := range session.Metadata { + metadata[key] = value + } + } + if session.WorktreeID != "" { + metadata["worktree_id"] = session.WorktreeID + } + if len(metadata) > 0 { + req.Metadata = metadata + } + + repositoryID := session.RepositoryID + var repositoryPath string + if repositoryID == "" && task.RepositoryID != nil { + repositoryID = *task.RepositoryID + } + if repositoryID != "" { + repository, err := e.repo.GetRepository(ctx, repositoryID) + if err != nil { + e.logger.Error("failed to load repository for task session resume", + zap.String("task_id", task.ID), + zap.String("repository_id", repositoryID), + zap.Error(err)) + return nil, err + } + repositoryPath = repository.LocalPath + if repositoryPath != "" { + req.RepositoryURL = repositoryPath + } + } + + baseBranch := session.BaseBranch + if baseBranch == "" && task.BaseBranch != nil && *task.BaseBranch != "" { + baseBranch = *task.BaseBranch + } + if baseBranch != "" { + req.Branch = baseBranch + } + + if e.worktreeEnabled && repositoryPath != "" { + req.UseWorktree = true + req.RepositoryPath = repositoryPath + req.RepositoryID = repositoryID + if baseBranch != "" { + req.BaseBranch = baseBranch + } else { + req.BaseBranch = "main" + } + } + + if session.Metadata != nil { + if acpSessionID, ok := session.Metadata["acp_session_id"].(string); ok && acpSessionID != "" { + req.ACPSessionID = acpSessionID + } + } + + e.logger.Info("resuming agent session", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID), + zap.String("agent_profile_id", session.AgentProfileID), + zap.Bool("use_worktree", req.UseWorktree)) + + resp, err := e.agentManager.LaunchAgent(ctx, req) + if err != nil { + e.logger.Error("failed to relaunch agent for session", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID), + zap.Error(err)) + return nil, err + } + + session.AgentInstanceID = resp.AgentInstanceID + session.ContainerID = resp.ContainerID + if resp.WorktreePath != "" { + session.WorktreePath = resp.WorktreePath + } + if resp.WorktreeBranch != "" { + session.WorktreeBranch = resp.WorktreeBranch + } + session.Progress = 0 + session.ErrorMessage = "" + session.State = models.TaskSessionStateStarting + session.CompletedAt = nil + + if err := e.repo.UpdateTaskSession(ctx, session); err != nil { + e.logger.Error("failed to update task session for resume", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID), + zap.Error(err)) + } + + now := time.Now().UTC() + execution := &TaskExecution{ + TaskID: task.ID, + AgentInstanceID: resp.AgentInstanceID, + AgentProfileID: session.AgentProfileID, + StartedAt: now, + SessionState: v1.TaskSessionStateStarting, + Progress: 0, + LastUpdate: now, + SessionID: session.ID, + WorktreePath: session.WorktreePath, + WorktreeBranch: session.WorktreeBranch, + } + + e.mu.Lock() + e.executions[task.ID] = execution + e.mu.Unlock() + + return execution, nil +} + // Stop stops an active execution func (e *Executor) Stop(ctx context.Context, taskID string, reason string, force bool) error { execution, err := e.getOrLoadExecution(ctx, taskID) diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index 4425b7805b..d2e5ed78a5 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -36,6 +36,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionOrchestratorComplete, h.wsCompleteTask) d.RegisterFunc(ws.ActionPermissionRespond, h.wsRespondToPermission) d.RegisterFunc(ws.ActionTaskExecution, h.wsGetTaskExecution) + d.RegisterFunc(ws.ActionTaskSessionResume, h.wsResumeTaskSession) } // WS handlers @@ -109,6 +110,37 @@ func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Messag return ws.NewResponse(msg.ID, msg.Action, resp) } +type wsResumeTaskSessionRequest struct { + TaskID string `json:"task_id"` + TaskSessionID string `json:"task_session_id"` +} + +func (h *Handlers) wsResumeTaskSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsResumeTaskSessionRequest + if err := msg.ParsePayload(&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.TaskSessionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_session_id is required", nil) + } + + resp, err := h.controller.ResumeTaskSession(ctx, dto.ResumeTaskSessionRequest{ + TaskID: req.TaskID, + TaskSessionID: req.TaskSessionID, + }) + if err != nil { + h.logger.Error("failed to resume task session", + zap.String("task_id", req.TaskID), + zap.String("task_session_id", req.TaskSessionID), + zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to resume task session: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + type wsStopTaskRequest struct { TaskID string `json:"task_id"` Reason string `json:"reason,omitempty"` diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 215a244be5..3f801930f5 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "os" "sync" "time" @@ -92,7 +93,7 @@ type Service struct { } type readyState struct { - readySeen bool + readySeen bool promptCompleteSeen bool } @@ -148,15 +149,16 @@ func NewService( // Create the watcher with event handlers that wire everything together handlers := watcher.EventHandlers{ - OnTaskStateChanged: s.handleTaskStateChanged, - OnAgentReady: s.handleAgentReady, - OnAgentCompleted: s.handleAgentCompleted, - OnAgentFailed: s.handleAgentFailed, - OnACPMessage: s.handleACPMessage, - OnPromptComplete: s.handlePromptComplete, - OnToolCallStarted: s.handleToolCallStarted, - OnToolCallComplete: s.handleToolCallComplete, - OnGitStatusUpdated: s.handleGitStatusUpdated, + OnTaskStateChanged: s.handleTaskStateChanged, + OnAgentReady: s.handleAgentReady, + OnAgentCompleted: s.handleAgentCompleted, + OnAgentFailed: s.handleAgentFailed, + OnACPMessage: s.handleACPMessage, + OnACPSessionCreated: s.handleACPSessionCreated, + OnPromptComplete: s.handlePromptComplete, + OnToolCallStarted: s.handleToolCallStarted, + OnToolCallComplete: s.handleToolCallComplete, + OnGitStatusUpdated: s.handleGitStatusUpdated, } s.watcher = watcher.NewWatcher(eventBus, handlers, log) @@ -342,6 +344,62 @@ func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID s return execution, nil } +// ResumeTaskSession restarts a specific task session using its stored worktree. +func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID string) (*executor.TaskExecution, error) { + s.logger.Info("resuming task session", + zap.String("task_id", taskID), + zap.String("task_session_id", sessionID)) + + if exec, ok := s.executor.GetExecutionWithContext(ctx, taskID); ok && exec != nil { + return nil, executor.ErrExecutionAlreadyRunning + } + + if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateScheduling); err != nil { + s.logger.Warn("failed to update task state to SCHEDULING", + zap.String("task_id", taskID), + zap.Error(err)) + } + + task, err := s.scheduler.GetTask(ctx, taskID) + if err != nil { + s.logger.Error("failed to fetch task for resume", + zap.String("task_id", taskID), + zap.Error(err)) + return nil, err + } + + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + return nil, err + } + if session.TaskID != taskID { + return nil, fmt.Errorf("task session does not belong to task") + } + if session.WorktreePath == "" { + return nil, fmt.Errorf("task session has no worktree path") + } + if _, err := os.Stat(session.WorktreePath); err != nil { + return nil, fmt.Errorf("worktree path not found: %w", err) + } + + s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateStarting, "", true) + + execution, err := s.executor.ResumeSession(ctx, task, session) + if err != nil { + return nil, err + } + + s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true) + + if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { + s.logger.Warn("failed to update task state to IN_PROGRESS", + zap.String("task_id", taskID), + zap.Error(err)) + } + + return execution, nil +} + // StopTask stops agent execution for a task func (s *Service) StopTask(ctx context.Context, taskID string, reason string, force bool) error { s.logger.Info("stopping task execution", @@ -483,6 +541,57 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD } } +func (s *Service) handleACPSessionCreated(ctx context.Context, data watcher.ACPSessionEventData) { + if data.TaskID == "" || data.ACPSessionID == "" { + return + } + + sessionID := "" + if exec, ok := s.executor.GetExecution(data.TaskID); ok && exec != nil { + sessionID = exec.SessionID + } + if sessionID == "" { + if session, err := s.repo.GetTaskSessionByTaskID(ctx, data.TaskID); err == nil && session != nil { + sessionID = session.ID + } + } + if sessionID == "" { + s.logger.Warn("no task session found to store ACP session id", + zap.String("task_id", data.TaskID)) + return + } + + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + s.logger.Warn("failed to load task session for ACP session update", + zap.String("task_id", data.TaskID), + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + + if session.Metadata == nil { + session.Metadata = make(map[string]interface{}) + } + if existing, ok := session.Metadata["acp_session_id"].(string); ok && existing == data.ACPSessionID { + return + } + session.Metadata["acp_session_id"] = data.ACPSessionID + + if err := s.repo.UpdateTaskSession(ctx, session); err != nil { + s.logger.Warn("failed to persist ACP session id", + zap.String("task_id", data.TaskID), + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + + s.logger.Info("stored ACP session id for task session", + zap.String("task_id", data.TaskID), + zap.String("task_session_id", sessionID), + zap.String("acp_session_id", data.ACPSessionID)) +} + // handleAgentCompleted handles agent completion events func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEventData) { s.logger.Info("handling agent completed", @@ -665,22 +774,22 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID return } if err := s.repo.UpdateTaskSessionState(ctx, sessionID, nextState, errorMessage); err != nil { - s.logger.Error("failed to update agent session state", + s.logger.Error("failed to update task session state", zap.String("session_id", sessionID), zap.String("state", string(nextState)), zap.Error(err)) } - s.logger.Info("agent session state updated", + s.logger.Info("task session state updated", zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.String("old_state", string(oldState)), zap.String("new_state", string(nextState))) if s.eventBus != nil { - _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "agent-session", map[string]interface{}{ - "task_id": taskID, - "agent_session_id": sessionID, - "old_state": string(oldState), - "new_state": string(nextState), + _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "task-session", map[string]interface{}{ + "task_id": taskID, + "task_session_id": sessionID, + "old_state": string(oldState), + "new_state": string(nextState), })) } if taskID != "" { diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index 6a98174831..347f163837 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -12,8 +12,8 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" - v1 "github.com/kandev/kandev/pkg/api/v1" "github.com/kandev/kandev/pkg/acp/protocol" + v1 "github.com/kandev/kandev/pkg/api/v1" ) // TaskEventData contains data from task events @@ -33,6 +33,13 @@ type AgentEventData struct { ErrorMessage string `json:"error_message,omitempty"` } +// ACPSessionEventData contains data from ACP session events +type ACPSessionEventData struct { + TaskID string `json:"task_id"` + AgentInstanceID string `json:"agent_instance_id"` + ACPSessionID string `json:"acp_session_id"` +} + // PromptCompleteData contains data from prompt_complete events type PromptCompleteData struct { TaskID string `json:"task_id"` @@ -85,11 +92,12 @@ type EventHandlers struct { OnTaskDeleted func(ctx context.Context, data TaskEventData) // Agent events - OnAgentStarted func(ctx context.Context, data AgentEventData) - OnAgentReady func(ctx context.Context, data AgentEventData) - OnAgentCompleted func(ctx context.Context, data AgentEventData) - OnAgentFailed func(ctx context.Context, data AgentEventData) - OnAgentStopped func(ctx context.Context, data AgentEventData) + OnAgentStarted func(ctx context.Context, data AgentEventData) + OnAgentReady func(ctx context.Context, data AgentEventData) + OnAgentCompleted func(ctx context.Context, data AgentEventData) + OnAgentFailed func(ctx context.Context, data AgentEventData) + OnAgentStopped func(ctx context.Context, data AgentEventData) + OnACPSessionCreated func(ctx context.Context, data ACPSessionEventData) // ACP messages OnACPMessage func(ctx context.Context, taskID string, msg *protocol.Message) @@ -151,6 +159,12 @@ func (w *Watcher) Start(ctx context.Context) error { return err } + // Subscribe to ACP session events + if err := w.subscribeToACPSessionEvents(); err != nil { + w.unsubscribeAll() + return err + } + // Subscribe to ACP messages if err := w.subscribeToACPMessages(); err != nil { w.unsubscribeAll() @@ -272,6 +286,23 @@ func (w *Watcher) subscribeToAgentEvents() error { return nil } +// subscribeToACPSessionEvents subscribes to ACP session lifecycle events +func (w *Watcher) subscribeToACPSessionEvents() error { + if w.handlers.OnACPSessionCreated == nil { + return nil + } + + sub, err := w.eventBus.QueueSubscribe(events.AgentACPSessionCreated, queueName, w.createACPSessionEventHandler(w.handlers.OnACPSessionCreated)) + if err != nil { + w.logger.Error("Failed to subscribe to ACP session event", + zap.String("subject", events.AgentACPSessionCreated), + zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) + return nil +} + // subscribeToACPMessages subscribes to ACP messages using wildcard func (w *Watcher) subscribeToACPMessages() error { if w.handlers.OnACPMessage == nil { @@ -336,6 +367,28 @@ func (w *Watcher) createAgentEventHandler(handler func(ctx context.Context, data } } +// createACPSessionEventHandler creates a bus.EventHandler for ACP session events +func (w *Watcher) createACPSessionEventHandler(handler func(ctx context.Context, data ACPSessionEventData)) bus.EventHandler { + return func(ctx context.Context, event *bus.Event) error { + var data ACPSessionEventData + if err := w.parseEventData(event.Data, &data); err != nil { + w.logger.Error("Failed to parse ACP session event data", + zap.String("event_type", event.Type), + zap.String("event_id", event.ID), + zap.Error(err)) + return nil + } + + w.logger.Debug("Handling ACP session event", + zap.String("event_type", event.Type), + zap.String("task_id", data.TaskID), + zap.String("acp_session_id", data.ACPSessionID)) + + handler(ctx, data) + return nil + } +} + // createACPMessageHandler creates a bus.EventHandler for ACP messages func (w *Watcher) createACPMessageHandler() bus.EventHandler { return func(ctx context.Context, event *bus.Event) error { diff --git a/apps/backend/internal/task/controller/task.go b/apps/backend/internal/task/controller/task.go index 2621536c2a..5aa60c1c79 100644 --- a/apps/backend/internal/task/controller/task.go +++ b/apps/backend/internal/task/controller/task.go @@ -54,6 +54,21 @@ func (c *TaskController) ListTasks(ctx context.Context, req dto.ListTasksRequest return resp, nil } +func (c *TaskController) ListTaskSessions(ctx context.Context, req dto.ListTaskSessionsRequest) (dto.ListTaskSessionsResponse, error) { + sessions, err := c.service.ListTaskSessions(ctx, req.TaskID) + if err != nil { + return dto.ListTaskSessionsResponse{}, err + } + resp := dto.ListTaskSessionsResponse{ + Sessions: make([]dto.TaskSessionDTO, 0, len(sessions)), + Total: len(sessions), + } + for _, session := range sessions { + resp.Sessions = append(resp.Sessions, dto.FromTaskSession(session)) + } + return resp, nil +} + func (c *TaskController) GetTask(ctx context.Context, req dto.GetTaskRequest) (dto.TaskDTO, error) { task, err := c.service.GetTask(ctx, req.ID) if err != nil { diff --git a/apps/backend/internal/task/dto/dto.go b/apps/backend/internal/task/dto/dto.go index af498fd443..ffefac52c8 100644 --- a/apps/backend/internal/task/dto/dto.go +++ b/apps/backend/internal/task/dto/dto.go @@ -112,6 +112,37 @@ type TaskDTO struct { WorktreeBranch *string `json:"worktree_branch,omitempty"` } +type TaskSessionDTO struct { + ID string `json:"id"` + TaskID string `json:"task_id"` + AgentInstanceID string `json:"agent_instance_id,omitempty"` + ContainerID string `json:"container_id,omitempty"` + AgentProfileID string `json:"agent_profile_id,omitempty"` + ExecutorID string `json:"executor_id,omitempty"` + EnvironmentID string `json:"environment_id,omitempty"` + RepositoryID string `json:"repository_id,omitempty"` + BaseBranch string `json:"base_branch,omitempty"` + WorktreeID string `json:"worktree_id,omitempty"` + WorktreePath string `json:"worktree_path,omitempty"` + WorktreeBranch string `json:"worktree_branch,omitempty"` + State models.TaskSessionState `json:"state"` + Progress int `json:"progress"` + ErrorMessage string `json:"error_message,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` + AgentProfileSnapshot map[string]interface{} `json:"agent_profile_snapshot,omitempty"` + ExecutorSnapshot map[string]interface{} `json:"executor_snapshot,omitempty"` + EnvironmentSnapshot map[string]interface{} `json:"environment_snapshot,omitempty"` + RepositorySnapshot map[string]interface{} `json:"repository_snapshot,omitempty"` + StartedAt time.Time `json:"started_at"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + UpdatedAt time.Time `json:"updated_at"` +} + +type ListTaskSessionsResponse struct { + Sessions []TaskSessionDTO `json:"sessions"` + Total int `json:"total"` +} + type BoardSnapshotDTO struct { Board BoardDTO `json:"board"` Columns []ColumnDTO `json:"columns"` @@ -348,3 +379,32 @@ func FromTask(task *models.Task) TaskDTO { Metadata: task.Metadata, } } + +func FromTaskSession(session *models.TaskSession) TaskSessionDTO { + result := TaskSessionDTO{ + ID: session.ID, + TaskID: session.TaskID, + AgentInstanceID: session.AgentInstanceID, + ContainerID: session.ContainerID, + AgentProfileID: session.AgentProfileID, + ExecutorID: session.ExecutorID, + EnvironmentID: session.EnvironmentID, + RepositoryID: session.RepositoryID, + BaseBranch: session.BaseBranch, + WorktreeID: session.WorktreeID, + WorktreePath: session.WorktreePath, + WorktreeBranch: session.WorktreeBranch, + State: session.State, + Progress: session.Progress, + ErrorMessage: session.ErrorMessage, + Metadata: session.Metadata, + AgentProfileSnapshot: session.AgentProfileSnapshot, + ExecutorSnapshot: session.ExecutorSnapshot, + EnvironmentSnapshot: session.EnvironmentSnapshot, + RepositorySnapshot: session.RepositorySnapshot, + StartedAt: session.StartedAt, + CompletedAt: session.CompletedAt, + UpdatedAt: session.UpdatedAt, + } + return result +} diff --git a/apps/backend/internal/task/dto/requests.go b/apps/backend/internal/task/dto/requests.go index 0b7a27dbef..238322fb94 100644 --- a/apps/backend/internal/task/dto/requests.go +++ b/apps/backend/internal/task/dto/requests.go @@ -229,6 +229,10 @@ type ListTasksRequest struct { BoardID string } +type ListTaskSessionsRequest struct { + TaskID string +} + type GetTaskRequest struct { ID string } diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index e3206f1859..e3b8d0d720 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -2,9 +2,11 @@ package handlers import ( "context" + "errors" "net/http" "strconv" "strings" + "time" "github.com/gin-gonic/gin" "go.uber.org/zap" @@ -12,6 +14,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/task/controller" "github.com/kandev/kandev/internal/task/dto" + "github.com/kandev/kandev/internal/orchestrator/executor" v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" ) @@ -25,6 +28,7 @@ type PromptResult struct { // OrchestratorService defines the interface for orchestrator operations type OrchestratorService interface { PromptTask(ctx context.Context, taskID string, prompt string) (*PromptResult, error) + ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error } // MessageHandlers handles WebSocket requests for messages @@ -65,7 +69,7 @@ func (h *MessageHandlers) registerWS(dispatcher *ws.Dispatcher) { func (h *MessageHandlers) httpListMessages(c *gin.Context) { sessionID := c.Param("id") if sessionID == "" { - c.JSON(http.StatusBadRequest, gin.H{"error": "agent session id is required"}) + c.JSON(http.StatusBadRequest, gin.H{"error": "task session id is required"}) return } before := c.Query("before") @@ -115,7 +119,7 @@ func (h *MessageHandlers) httpListMessages(c *gin.Context) { type wsAddMessageRequest struct { TaskID string `json:"task_id"` - TaskSessionID string `json:"agent_session_id"` + TaskSessionID string `json:"task_session_id"` Content string `json:"content"` AuthorID string `json:"author_id,omitempty"` } @@ -126,7 +130,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) } if req.TaskSessionID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_session_id is required", nil) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_session_id is required", nil) } if req.TaskID == "" { return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) @@ -171,6 +175,22 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w if h.orchestrator != nil { _, err := h.orchestrator.PromptTask(ctx, req.TaskID, req.Content) if err != nil { + if errors.Is(err, executor.ErrExecutionNotFound) { + if resumeErr := h.orchestrator.ResumeTaskSession(ctx, req.TaskID, req.TaskSessionID); resumeErr != nil { + h.logger.Warn("failed to resume task session for prompt", + zap.String("task_id", req.TaskID), + zap.String("task_session_id", req.TaskSessionID), + zap.Error(resumeErr)) + } else { + for attempt := 0; attempt < 3; attempt++ { + time.Sleep(500 * time.Millisecond) + _, err = h.orchestrator.PromptTask(ctx, req.TaskID, req.Content) + if err == nil { + break + } + } + } + } h.logger.Warn("failed to forward message as prompt to agent", zap.String("task_id", req.TaskID), zap.Error(err)) @@ -181,7 +201,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w } type wsListMessagesRequest struct { - TaskSessionID string `json:"agent_session_id"` + TaskSessionID string `json:"task_session_id"` Limit int `json:"limit"` Before string `json:"before"` After string `json:"after"` @@ -194,7 +214,7 @@ func (h *MessageHandlers) wsListMessages(ctx context.Context, msg *ws.Message) ( return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) } if req.TaskSessionID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_session_id is required", nil) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_session_id is required", nil) } if req.Before != "" && req.After != "" { return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "only one of before or after can be set", nil) diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index 26e913c392..fe2afd582f 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -36,6 +36,7 @@ func (h *TaskHandlers) registerHTTP(router *gin.Engine) { api := router.Group("/api/v1") api.GET("/boards/:id/tasks", h.httpListTasks) api.GET("/tasks/:id", h.httpGetTask) + api.GET("/tasks/:id/sessions", h.httpListTaskSessions) api.POST("/tasks", h.httpCreateTask) api.PATCH("/tasks/:id", h.httpUpdateTask) api.POST("/tasks/:id/move", h.httpMoveTask) @@ -50,6 +51,7 @@ func (h *TaskHandlers) registerWS(dispatcher *ws.Dispatcher) { dispatcher.RegisterFunc(ws.ActionTaskDelete, h.wsDeleteTask) dispatcher.RegisterFunc(ws.ActionTaskMove, h.wsMoveTask) dispatcher.RegisterFunc(ws.ActionTaskState, h.wsUpdateTaskState) + dispatcher.RegisterFunc(ws.ActionTaskSessionList, h.wsListTaskSessions) } // HTTP handlers @@ -72,6 +74,15 @@ func (h *TaskHandlers) httpGetTask(c *gin.Context) { c.JSON(http.StatusOK, resp) } +func (h *TaskHandlers) httpListTaskSessions(c *gin.Context) { + resp, err := h.controller.ListTaskSessions(c.Request.Context(), dto.ListTaskSessionsRequest{TaskID: c.Param("id")}) + if err != nil { + handleNotFound(c, h.logger, err, "task sessions not found") + return + } + c.JSON(http.StatusOK, resp) +} + type httpCreateTaskRequest struct { WorkspaceID string `json:"workspace_id"` BoardID string `json:"board_id"` @@ -199,6 +210,27 @@ func (h *TaskHandlers) httpDeleteTask(c *gin.Context) { // WS handlers +type wsListTaskSessionsRequest struct { + TaskID string `json:"task_id"` +} + +func (h *TaskHandlers) wsListTaskSessions(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsListTaskSessionsRequest + if err := msg.ParsePayload(&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) + } + + resp, err := h.controller.ListTaskSessions(ctx, dto.ListTaskSessionsRequest{TaskID: req.TaskID}) + if err != nil { + h.logger.Error("failed to list task sessions", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list task sessions", nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + type wsListTasksRequest struct { BoardID string `json:"board_id"` } diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 38fae0dd22..2dda65c5ec 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -369,6 +369,11 @@ func (s *Service) ListTasks(ctx context.Context, boardID string) ([]*models.Task return s.repo.ListTasks(ctx, boardID) } +// ListTaskSessions returns all sessions for a task. +func (s *Service) ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) { + return s.repo.ListTaskSessions(ctx, taskID) +} + // UpdateTaskState updates the state of a task, moves it to the matching column, // and publishes a task.state_changed event func (s *Service) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) (*models.Task, error) { @@ -1276,7 +1281,7 @@ func (s *Service) publishEnvironmentEvent(ctx context.Context, eventType string, // CreateMessageRequest contains the data for creating a new message type CreateMessageRequest struct { - TaskSessionID string `json:"agent_session_id"` + TaskSessionID string `json:"task_session_id"` TaskID string `json:"task_id,omitempty"` Content string `json:"content"` AuthorType string `json:"author_type,omitempty"` // "user" or "agent", defaults to "user" @@ -1331,7 +1336,7 @@ func (s *Service) CreateMessage(ctx context.Context, req *CreateMessageRequest) s.logger.Info("message created", zap.String("message_id", message.ID), - zap.String("agent_session_id", message.TaskSessionID), + zap.String("task_session_id", message.TaskSessionID), zap.String("author_type", string(message.AuthorType))) return message, nil @@ -1380,7 +1385,7 @@ func (s *Service) UpdateToolCallMessage(ctx context.Context, sessionID, toolCall message, err := s.repo.GetMessageByToolCallID(ctx, sessionID, toolCallID) if err != nil { s.logger.Warn("tool call message not found for update", - zap.String("agent_session_id", sessionID), + zap.String("task_session_id", sessionID), zap.String("tool_call_id", toolCallID), zap.Error(err)) return err @@ -1427,7 +1432,7 @@ func (s *Service) publishMessageEvent(ctx context.Context, eventType string, mes data := map[string]interface{}{ "message_id": message.ID, - "agent_session_id": message.TaskSessionID, + "task_session_id": message.TaskSessionID, "task_id": message.TaskID, "author_type": string(message.AuthorType), "author_id": message.AuthorID, diff --git a/apps/backend/pkg/api/v1/task.go b/apps/backend/pkg/api/v1/task.go index de0d1779f0..19a1579d70 100644 --- a/apps/backend/pkg/api/v1/task.go +++ b/apps/backend/pkg/api/v1/task.go @@ -102,10 +102,10 @@ type TaskEvent struct { CreatedAt time.Time `json:"created_at"` } -// Message represents a message in an agent session (user or agent) +// Message represents a message in a task session (user or agent) type Message struct { ID string `json:"id"` - TaskSessionID string `json:"agent_session_id"` + TaskSessionID string `json:"task_session_id"` TaskID string `json:"task_id,omitempty"` AuthorType string `json:"author_type"` // "user" or "agent" Type string `json:"type,omitempty"` @@ -116,9 +116,9 @@ type Message struct { CreatedAt time.Time `json:"created_at"` } -// CreateMessageRequest for adding a message to an agent session +// CreateMessageRequest for adding a message to a task session type CreateMessageRequest struct { - TaskSessionID string `json:"agent_session_id" binding:"required"` + TaskSessionID string `json:"task_session_id" binding:"required"` Content string `json:"content" binding:"required"` AuthorType string `json:"author_type,omitempty"` // Defaults to "user" if not specified Type string `json:"type,omitempty"` diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 822810acea..83cf1dad3e 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -63,6 +63,8 @@ const ( ActionTaskMove = "task.move" ActionTaskState = "task.state" ActionTaskExecution = "task.execution" + ActionTaskSessionList = "task.session.list" + ActionTaskSessionResume = "task.session.resume" // Agent actions ActionAgentList = "agent.list" diff --git a/apps/web/app/task/[id]/[sessionId]/page.tsx b/apps/web/app/task/[id]/[sessionId]/page.tsx new file mode 100644 index 0000000000..a1f8eeeece --- /dev/null +++ b/apps/web/app/task/[id]/[sessionId]/page.tsx @@ -0,0 +1,33 @@ +import { StateHydrator } from '@/components/state-hydrator'; +import { fetchTask } from '@/lib/http'; +import type { Task } from '@/lib/types/http'; +import { taskToState } from '@/lib/ssr/mapper'; +import TaskPageClient from '../page-client'; + +export default async function TaskSessionPage({ + params, +}: { + params: Promise<{ id: string; sessionId: string }>; +}) { + let initialState: ReturnType | null = null; + let task: Task | null = null; + let sessionId: string | null = null; + + try { + const { id, sessionId: paramSessionId } = await params; + sessionId = paramSessionId; + task = await fetchTask(id, { cache: 'no-store' }); + initialState = taskToState(task); + } catch { + initialState = null; + task = null; + sessionId = null; + } + + return ( + <> + {initialState ? : null} + + + ); +} diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index 865d34619d..546014af93 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -16,9 +16,10 @@ import { useAppStore } from '@/components/state-provider'; type TaskPageClientProps = { task: Task | null; + sessionId?: string | null; }; -export default function TaskPage({ task: initialTask }: TaskPageClientProps) { +export default function TaskPage({ task: initialTask, sessionId = null }: TaskPageClientProps) { const [isMounted, setIsMounted] = useState(false); const kanbanTask = useAppStore((state) => initialTask?.id ? state.kanban.tasks.find((item) => item.id === initialTask.id) ?? null : null @@ -49,16 +50,18 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { handleStartAgent, handleStopAgent, } = useTaskAgent(task); + const activeSessionId = sessionId ?? taskSessionId; const isAgentWorking = taskSessionState !== null ? taskSessionState === 'STARTING' || taskSessionState === 'RUNNING' : isAgentRunning && (task?.state === 'IN_PROGRESS' || task?.state === 'SCHEDULING'); - const { isLoading: isLoadingMessages } = useTaskMessages(task?.id ?? null, taskSessionId); + const { isLoading: isLoadingMessages } = useTaskMessages(task?.id ?? null, activeSessionId); useEffect(() => { queueMicrotask(() => setIsMounted(true)); }, []); + const { repositories } = useRepositories(task?.workspace_id ?? null, Boolean(task?.workspace_id)); const repository = useMemo( () => repositories.find((item) => item.id === task?.repository_id) ?? null, @@ -71,7 +74,7 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { const client = getWebSocketClient(); if (!client) return; - if (!taskSessionId) { + if (!activeSessionId) { console.error('No active agent session. Start an agent before sending a message.'); return; } @@ -79,13 +82,13 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { try { await client.request( 'message.add', - { task_id: task.id, task_session_id: taskSessionId, content }, + { task_id: task.id, task_session_id: activeSessionId, content }, 10000 ); } catch (error) { console.error('Failed to send message:', error); } - }, [taskSessionId, task]); + }, [activeSessionId, task]); if (!isMounted) { return
; @@ -100,14 +103,16 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { entries={{ ws_status: connectionStatus, task_id: task?.id ?? null, - task_session_id: taskSessionId ?? null, + task_session_id: activeSessionId ?? null, task_state: task?.state ?? null, - agent_session_state: taskSessionState ?? null, + task_session_state: taskSessionState ?? null, is_agent_working: isAgentWorking, }} /> )} }) { let initialState: ReturnType | null = null; @@ -11,6 +12,10 @@ export default async function TaskPage({ params }: { params: Promise<{ id: strin try { const { id } = await params; task = await fetchTask(id, { cache: 'no-store' }); + const sessions = await listTaskSessions(id, { cache: 'no-store' }); + if (sessions.sessions.length > 0) { + redirect(`/task/${id}/${sessions.sessions[0].id}`); + } initialState = taskToState(task); } catch { initialState = null; diff --git a/apps/web/components/kanban-board.tsx b/apps/web/components/kanban-board.tsx index fd3be48383..f461d8e1a2 100644 --- a/apps/web/components/kanban-board.tsx +++ b/apps/web/components/kanban-board.tsx @@ -15,10 +15,11 @@ import { useBoardSnapshot } from '@/hooks/use-board-snapshot'; import { useTaskActions } from '@/hooks/use-task-actions'; import { KanbanBoardHeader } from './kanban-board-header'; import { KanbanBoardGrid } from './kanban-board-grid'; +import { getWebSocketClient } from '@/lib/ws/connection'; interface KanbanBoardProps { onPreviewTask?: (task: Task) => void; - onOpenTask?: (task: Task) => void; + onOpenTask?: (task: Task, sessionId: string) => void; } export function KanbanBoard({ onPreviewTask, onOpenTask }: KanbanBoardProps = {}) { @@ -26,6 +27,7 @@ export function KanbanBoard({ onPreviewTask, onOpenTask }: KanbanBoardProps = {} const [activeTaskId, setActiveTaskId] = useState(null); const [isMovingTask, setIsMovingTask] = useState(false); const [editingTask, setEditingTask] = useState(null); + const [taskSessionAvailability, setTaskSessionAvailability] = useState>({}); const router = useRouter(); const kanban = useAppStore((state) => state.kanban); const workspaceState = useAppStore((state) => state.workspaces); @@ -100,9 +102,16 @@ export function KanbanBoard({ onPreviewTask, onOpenTask }: KanbanBoardProps = {} () => filterTasksByRepositories(backendTasks, selectedRepositoryIds), [backendTasks, selectedRepositoryIds] ); + const visibleTasksWithSessions = useMemo( + () => visibleTasks.map((task) => ({ + ...task, + hasSession: taskSessionAvailability[task.id], + })), + [visibleTasks, taskSessionAvailability] + ); const activeTask = useMemo( - () => visibleTasks.find((task) => task.id === activeTaskId) ?? null, - [visibleTasks, activeTaskId] + () => visibleTasksWithSessions.find((task) => task.id === activeTaskId) ?? null, + [visibleTasksWithSessions, activeTaskId] ); const handleDragStart = (event: DragStartEvent) => { @@ -196,15 +205,47 @@ export function KanbanBoard({ onPreviewTask, onOpenTask }: KanbanBoardProps = {} }); }; - const handleOpenTask = (task: Task) => { - if (onOpenTask) { - onOpenTask(task); - } else { - router.push(`/task/${task.id}`); + const fetchLatestSessionId = async (taskId: string) => { + const client = getWebSocketClient(); + if (!client) return null; + try { + const response = await client.request<{ sessions: Array<{ id: string }> }>( + 'task.session.list', + { task_id: taskId }, + 10000 + ); + setTaskSessionAvailability((prev) => ({ + ...prev, + [taskId]: response.sessions.length > 0, + })); + return response.sessions[0]?.id ?? null; + } catch (error) { + console.error('Failed to load task sessions:', error); + return null; } }; - const handlePreviewTask = (task: Task) => { + const handleOpenTask = async (task: Task) => { + const latestSessionId = await fetchLatestSessionId(task.id); + if (!latestSessionId) { + setEditingTask(task); + setIsDialogOpen(true); + return; + } + if (onOpenTask) { + onOpenTask(task, latestSessionId); + } else { + router.push(`/task/${task.id}/${latestSessionId}`); + } + }; + + const handlePreviewTask = async (task: Task) => { + const latestSessionId = await fetchLatestSessionId(task.id); + if (!latestSessionId) { + setEditingTask(task); + setIsDialogOpen(true); + return; + } if (onPreviewTask) { onPreviewTask(task); } else { @@ -375,7 +416,7 @@ export function KanbanBoard({ onPreviewTask, onOpenTask }: KanbanBoardProps = {} /> {statusIcon} - {onOpenFullPage && ( + {onOpenFullPage && task.hasSession !== false && (
)} diff --git a/apps/web/components/task-preview-panel.tsx b/apps/web/components/task-preview-panel.tsx index 7622940f92..9350dab7ac 100644 --- a/apps/web/components/task-preview-panel.tsx +++ b/apps/web/components/task-preview-panel.tsx @@ -1,113 +1,67 @@ "use client"; -import { IconX, IconCircleCheck, IconCircleX, IconLoader2, IconAlertTriangle, IconArrowsMaximize } from "@tabler/icons-react"; +import { IconArrowsMaximize, IconX } from "@tabler/icons-react"; import { Button } from "@kandev/ui/button"; -import { Badge } from "@kandev/ui/badge"; -import { getRepositoryDisplayName } from "@/lib/utils"; -import { useAppStore } from "@/components/state-provider"; -import { useMemo } from "react"; import type { Task } from "./kanban-card"; +import { TaskChatPanel } from "./task/task-chat-panel"; +import { useTaskMessages } from "@/hooks/use-task-messages"; +import { useTaskChatSession } from "@/hooks/use-task-chat-session"; +import { getWebSocketClient } from "@/lib/ws/connection"; interface TaskPreviewPanelProps { task: Task | null; + sessionId?: string | null; onClose: () => void; - onMaximize?: (task: Task) => void; + onMaximize?: (task: Task, sessionId: string) => void; } -function getStateDisplay(state?: string) { - switch (state) { - case "IN_PROGRESS": - case "SCHEDULING": - return { - label: state === "SCHEDULING" ? "Scheduling" : "In Progress", - icon: , - variant: "default" as const, - }; - case "COMPLETED": - return { - label: "Completed", - icon: , - variant: "default" as const, - className: "bg-emerald-500/10 text-emerald-500 border-emerald-500/20", - }; - case "FAILED": - case "CANCELLED": - return { - label: state === "FAILED" ? "Failed" : "Cancelled", - icon: , - variant: "destructive" as const, - }; - case "BLOCKED": - case "WAITING_FOR_INPUT": - return { - label: state === "BLOCKED" ? "Blocked" : "Waiting for Input", - icon: , - variant: "default" as const, - className: "bg-yellow-500/10 text-yellow-500 border-yellow-500/20", - }; - case "TODO": - return { - label: "To Do", - icon: null, - variant: "secondary" as const, - }; - case "REVIEW": - return { - label: "Review", - icon: null, - variant: "default" as const, - }; - case "CREATED": - default: - return { - label: "Created", - icon: null, - variant: "secondary" as const, - }; - } -} +const AGENTS = [ + { id: "codex", label: "Codex" }, + { id: "claude", label: "Claude Code" }, +]; -export function TaskPreviewPanel({ task, onClose, onMaximize }: TaskPreviewPanelProps) { - const repositoriesByWorkspace = useAppStore((state) => state.repositories.itemsByWorkspaceId); - const repository = useMemo(() => { - if (!task?.repositoryId) return null; - return Object.values(repositoriesByWorkspace) - .flat() - .find((repo) => repo.id === task.repositoryId) ?? null; - }, [repositoriesByWorkspace, task]); - const repoName = getRepositoryDisplayName(repository?.local_path); +export function TaskPreviewPanel({ task, sessionId = null, onClose, onMaximize }: TaskPreviewPanelProps) { + const { taskSessionId, isTaskSessionWorking } = useTaskChatSession(task?.id ?? null); + const activeSessionId = sessionId ?? taskSessionId; + const isActiveSessionWorking = sessionId ? isTaskSessionWorking && sessionId === taskSessionId : isTaskSessionWorking; + const { isLoading: isLoadingMessages } = useTaskMessages(task?.id ?? null, activeSessionId); - if (!task) { - return ( -
-
-

Task Preview

- -
-
-

Select a task to preview

-
-
- ); - } + const handleSendMessage = async (content: string) => { + if (!task?.id) return; - const stateDisplay = getStateDisplay(task.state); + const client = getWebSocketClient(); + if (!client) return; + + if (!activeSessionId) { + console.error("No active task session. Start an agent before sending a message."); + return; + } + + try { + await client.request( + "message.add", + { task_id: task.id, task_session_id: activeSessionId, content }, + 10000 + ); + } catch (error) { + console.error("Failed to send message:", error); + } + }; return (
{/* Header */}
-

Task Preview

+

+ {task?.title ?? "Task Chat"} +

- {onMaximize && ( + {onMaximize && activeSessionId && task && (
{/* Content */} -
-
- {/* Repository */} - {repoName && ( -
-

Repository

-

{repoName}

-
- )} - - {/* Title */} -
-

Title

-

{task.title}

+
+ {task ? ( + + ) : ( +
+ Select a task to start chatting
- - {/* State */} - {task.state && ( -
-

Status

- - {stateDisplay.icon && {stateDisplay.icon}} - {stateDisplay.label} - -
- )} - - {/* Description */} - {task.description && ( -
-

Description

-

- {task.description} -

-
- )} -
+ )}
); diff --git a/apps/web/components/task/chat/message-renderer.tsx b/apps/web/components/task/chat/message-renderer.tsx index fa38503636..dcf1b2c55d 100644 --- a/apps/web/components/task/chat/message-renderer.tsx +++ b/apps/web/components/task/chat/message-renderer.tsx @@ -42,7 +42,7 @@ const adapters: MessageAdapter[] = [ ); @@ -52,7 +52,7 @@ const adapters: MessageAdapter[] = [ ); } @@ -60,7 +60,7 @@ const adapters: MessageAdapter[] = [ ); diff --git a/apps/web/components/task/chat/messages/chat-message.tsx b/apps/web/components/task/chat/messages/chat-message.tsx index 9d263d0dca..542c781797 100644 --- a/apps/web/components/task/chat/messages/chat-message.tsx +++ b/apps/web/components/task/chat/messages/chat-message.tsx @@ -18,7 +18,7 @@ export function ChatMessage({ comment, label, className, showRichBlocks }: ChatM const isUser = comment.author_type === 'user'; return ( -
+

{label} {comment.requests_input ? ( diff --git a/apps/web/components/task/chat/messages/status-message.tsx b/apps/web/components/task/chat/messages/status-message.tsx index b066632093..ce2320f85c 100644 --- a/apps/web/components/task/chat/messages/status-message.tsx +++ b/apps/web/components/task/chat/messages/status-message.tsx @@ -17,10 +17,10 @@ export function StatusMessage({ comment }: { comment: Message }) { return (

diff --git a/apps/web/components/task/chat/messages/thinking-message.tsx b/apps/web/components/task/chat/messages/thinking-message.tsx index 7917730377..dbc8abb40e 100644 --- a/apps/web/components/task/chat/messages/thinking-message.tsx +++ b/apps/web/components/task/chat/messages/thinking-message.tsx @@ -9,7 +9,7 @@ export function ThinkingMessage({ comment }: { comment: Message }) { const text = metadata?.thinking ?? comment.content; if (!text) return null; return ( -
+
Thinking diff --git a/apps/web/components/task/chat/messages/todo-message.tsx b/apps/web/components/task/chat/messages/todo-message.tsx index f077aaf405..ee5cdd7d7e 100644 --- a/apps/web/components/task/chat/messages/todo-message.tsx +++ b/apps/web/components/task/chat/messages/todo-message.tsx @@ -20,7 +20,7 @@ export function TodoMessage({ comment }: { comment: Message }) { const progress = Math.round((completed / todoItems.length) * 100); return ( -
+
Todos diff --git a/apps/web/components/task/chat/messages/tool-call-message.tsx b/apps/web/components/task/chat/messages/tool-call-message.tsx index 336cd3e782..161b55c54d 100644 --- a/apps/web/components/task/chat/messages/tool-call-message.tsx +++ b/apps/web/components/task/chat/messages/tool-call-message.tsx @@ -77,7 +77,7 @@ export function ToolCallMessage({ comment }: { comment: Message }) { } return ( -
+
- Agent Sessions + Task Sessions
- {sessions.length === 0 ? ( + {visibleSessions.length === 0 ? (
No sessions yet
) : (
- {sessions.map((session) => { + {visibleSessions.map((session) => { const duration = formatDuration(session.startedAt, session.status === 'running'); // Use currentTime to force re-render when timer updates void currentTime; @@ -172,7 +247,7 @@ export function SessionsDropdown({ taskTitle = '', taskDescription = '' }: Sessi key={session.id} onClick={() => handleSelectSession(session.id)} className={`w-full flex items-center gap-3 px-2 py-1.5 hover:bg-muted/50 rounded-sm cursor-pointer transition-colors ${ - selectedSessionId === session.id ? 'bg-muted/50' : '' + activeSessionId === session.id ? 'bg-muted/50' : '' }`} > {/* Session Number - Fixed width */} @@ -245,4 +320,3 @@ export function SessionsDropdown({ taskTitle = '', taskDescription = '' }: Sessi ); } - diff --git a/apps/web/components/task/task-top-bar.tsx b/apps/web/components/task/task-top-bar.tsx index d80d7e8090..0a5a997b8e 100644 --- a/apps/web/components/task/task-top-bar.tsx +++ b/apps/web/components/task/task-top-bar.tsx @@ -41,6 +41,8 @@ import { useAppStore } from '@/components/state-provider'; import { formatUserHomePath } from '@/lib/utils'; type TaskTopBarProps = { + taskId?: string | null; + activeSessionId?: string | null; taskTitle?: string; taskDescription?: string; baseBranch?: string; @@ -55,6 +57,8 @@ type TaskTopBarProps = { }; const TaskTopBar = memo(function TaskTopBar({ + taskId, + activeSessionId, taskTitle, taskDescription, baseBranch, @@ -148,7 +152,12 @@ const TaskTopBar = memo(function TaskTopBar({ )} {taskTitle ?? 'Task details'} - + {displayBranch && ( diff --git a/apps/web/hooks/use-task-chat-session.ts b/apps/web/hooks/use-task-chat-session.ts new file mode 100644 index 0000000000..65eb2baf41 --- /dev/null +++ b/apps/web/hooks/use-task-chat-session.ts @@ -0,0 +1,74 @@ +import { useEffect, useMemo, useState } from 'react'; +import { getWebSocketClient } from '@/lib/ws/connection'; +import { useAppStore } from '@/components/state-provider'; +import type { TaskSessionState } from '@/lib/types/http'; + +type UseTaskChatSessionReturn = { + taskSessionId: string | null; + taskSessionState: TaskSessionState | null; + isTaskSessionWorking: boolean; +}; + +export function useTaskChatSession(taskId: string | null): UseTaskChatSessionReturn { + const [taskSessionIdsByTaskId, setTaskSessionIdsByTaskId] = useState>({}); + const connectionStatus = useAppStore((state) => state.connection.status); + const taskSessionState = useAppStore((state) => + taskId ? state.taskSessionStatesByTaskId?.[taskId] ?? null : null + ); + const taskSessionId = useMemo( + () => (taskId ? taskSessionIdsByTaskId[taskId] ?? null : null), + [taskId, taskSessionIdsByTaskId] + ); + + useEffect(() => { + if (!taskId) return; + if (connectionStatus !== 'connected') return; + + const checkExecution = async () => { + const client = getWebSocketClient(); + if (!client) return; + + try { + const response = await client.request<{ + has_execution: boolean; + task_id: string; + state?: string; + task_session_id?: string; + }>('task.execution', { task_id: taskId }); + + setTaskSessionIdsByTaskId((prev) => { + if (!taskId) return prev; + if (!response.has_execution || !response.task_session_id) { + if (!prev[taskId]) return prev; + const next = { ...prev }; + delete next[taskId]; + return next; + } + if (prev[taskId] === response.task_session_id) { + return prev; + } + return { ...prev, [taskId]: response.task_session_id }; + }); + } catch (err) { + console.error('[useTaskChatSession] Failed to check task execution:', err); + } + }; + + checkExecution(); + const interval = setInterval(() => { + if (connectionStatus === 'connected') { + checkExecution(); + } + }, 2000); + + return () => clearInterval(interval); + }, [connectionStatus, taskId]); + + const isTaskSessionWorking = taskSessionState === 'STARTING' || taskSessionState === 'RUNNING'; + + return { + taskSessionId, + taskSessionState, + isTaskSessionWorking, + }; +} diff --git a/apps/web/lib/config.ts b/apps/web/lib/config.ts index cef6f34a79..1bb9b92a68 100644 --- a/apps/web/lib/config.ts +++ b/apps/web/lib/config.ts @@ -3,7 +3,8 @@ export type AppConfig = { }; const DEFAULT_API_BASE_URL = 'http://localhost:8080'; -export const DEBUG_UI = process.env.NEXT_PUBLIC_KANDEV_DEBUG === 'true'; +export const DEBUG_UI = true +// export const DEBUG_UI = process.env.NEXT_PUBLIC_KANDEV_DEBUG === 'true'; export function getBackendConfig(): AppConfig { if (typeof window === 'undefined') { diff --git a/apps/web/lib/http/index.ts b/apps/web/lib/http/index.ts index 973d4c873f..3d18d67f8b 100644 --- a/apps/web/lib/http/index.ts +++ b/apps/web/lib/http/index.ts @@ -11,6 +11,7 @@ import type { ListMessagesResponse, RepositoryBranchesResponse, Task, + TaskSessionsResponse, UserSettingsResponse, Workspace, } from '@/lib/types/http'; @@ -159,6 +160,10 @@ export async function fetchTask(taskId: string, options?: ApiRequestOptions) { return fetchJson(`/api/v1/tasks/${taskId}`, options); } +export async function listTaskSessions(taskId: string, options?: ApiRequestOptions) { + return fetchJson(`/api/v1/tasks/${taskId}/sessions`, options); +} + export async function listTaskSessionMessages( taskSessionId: string, params?: { limit?: number; before?: string; after?: string; sort?: 'asc' | 'desc' }, diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index 8208444204..e6cc9b4690 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -99,6 +99,37 @@ export type Task = { worktree_branch?: string | null; }; +export type TaskSession = { + id: string; + task_id: string; + agent_instance_id?: string; + container_id?: string; + agent_profile_id?: string; + executor_id?: string; + environment_id?: string; + repository_id?: string; + base_branch?: string; + worktree_id?: string; + worktree_path?: string; + worktree_branch?: string; + state: TaskSessionState; + progress: number; + error_message?: string; + metadata?: Record | null; + agent_profile_snapshot?: Record | null; + executor_snapshot?: Record | null; + environment_snapshot?: Record | null; + repository_snapshot?: Record | null; + started_at: string; + completed_at?: string | null; + updated_at: string; +}; + +export type TaskSessionsResponse = { + sessions: TaskSession[]; + total: number; +}; + export type User = { id: string; email: string;