feat: flaky resume sessions

This commit is contained in:
Carlos Florêncio
2026-01-15 23:39:07 +00:00
parent 712610e428
commit 4560901d35
41 changed files with 1331 additions and 337 deletions
+67 -60
View File
@@ -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
}
+1 -1
View File
@@ -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')),
@@ -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"`
@@ -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()
+56 -2
View File
@@ -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"`
@@ -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) {
@@ -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)
+7 -6
View File
@@ -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
@@ -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
+19 -2
View File
@@ -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"`
@@ -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)
@@ -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"`
+126 -17
View File
@@ -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 != "" {
@@ -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 {
@@ -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 {
+60
View File
@@ -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
}
@@ -229,6 +229,10 @@ type ListTasksRequest struct {
BoardID string
}
type ListTaskSessionsRequest struct {
TaskID string
}
type GetTaskRequest struct {
ID string
}
@@ -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)
@@ -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"`
}
@@ -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,
+4 -4
View File
@@ -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"`
+2
View File
@@ -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"
@@ -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<typeof taskToState> | 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 ? <StateHydrator initialState={initialState} /> : null}
<TaskPageClient task={task} sessionId={sessionId} />
</>
);
}
+12 -7
View File
@@ -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 <div className="h-screen w-full bg-background" />;
@@ -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,
}}
/>
)}
<TaskTopBar
taskId={task?.id ?? null}
activeSessionId={activeSessionId ?? null}
taskTitle={task?.title}
taskDescription={task?.description}
baseBranch={task?.base_branch ?? undefined}
+6 -1
View File
@@ -1,8 +1,9 @@
import { StateHydrator } from '@/components/state-hydrator';
import { fetchTask } from '@/lib/http';
import { fetchTask, listTaskSessions } from '@/lib/http';
import type { Task } from '@/lib/types/http';
import { taskToState } from '@/lib/ssr/mapper';
import TaskPageClient from './page-client';
import { redirect } from 'next/navigation';
export default async function TaskPage({ params }: { params: Promise<{ id: string }> }) {
let initialState: ReturnType<typeof taskToState> | 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;
+51 -10
View File
@@ -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<string | null>(null);
const [isMovingTask, setIsMovingTask] = useState(false);
const [editingTask, setEditingTask] = useState<Task | null>(null);
const [taskSessionAvailability, setTaskSessionAvailability] = useState<Record<string, boolean>>({});
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 = {}
/>
<KanbanBoardGrid
columns={activeColumns}
tasks={visibleTasks}
tasks={visibleTasksWithSessions}
onPreviewTask={handlePreviewTask}
onOpenTask={handleOpenTask}
onEditTask={handleEditTask}
+2 -1
View File
@@ -22,6 +22,7 @@ export interface Task {
description?: string;
position?: number;
repositoryId?: string;
hasSession?: boolean;
}
interface KanbanCardProps {
@@ -138,7 +139,7 @@ export function KanbanCard({ task, onClick, onEdit, onDelete, onOpenFullPage }:
actions={
<div className="flex items-center gap-2">
{statusIcon}
{onOpenFullPage && (
{onOpenFullPage && task.hasSession !== false && (
<button
type="button"
className="text-muted-foreground hover:text-foreground hover:bg-accent rounded-sm p-1 -m-1 transition-colors cursor-pointer"
+64 -4
View File
@@ -8,6 +8,7 @@ import { useKanbanPreview } from "@/hooks/use-kanban-preview";
import { useAppStore } from "@/components/state-provider";
import { Task } from "./kanban-card";
import { PREVIEW_PANEL } from "@/lib/settings/constants";
import { getWebSocketClient } from "@/lib/ws/connection";
export function KanbanWithPreview() {
const router = useRouter();
@@ -30,6 +31,7 @@ export function KanbanWithPreview() {
// Cleanup handled by close
},
});
const [selectedTaskSessionId, setSelectedTaskSessionId] = useState<string | null>(null);
// Compute selected task from kanbanTasks and selectedTaskId
const selectedTask = useMemo(() => {
@@ -49,6 +51,44 @@ export function KanbanWithPreview() {
};
}, [selectedTaskId, kanbanTasks]);
useEffect(() => {
if (!selectedTaskId) {
return;
}
let isActive = true;
const loadLatestSession = async () => {
const client = getWebSocketClient();
if (!client) {
if (isActive) {
setSelectedTaskSessionId(null);
}
return;
}
try {
const response = await client.request<{ sessions: Array<{ id: string }> }>(
"task.session.list",
{ task_id: selectedTaskId },
10000
);
if (isActive) {
setSelectedTaskSessionId(response.sessions[0]?.id ?? null);
}
} catch (error) {
console.error("Failed to load task sessions:", error);
if (isActive) {
setSelectedTaskSessionId(null);
}
}
};
loadLatestSession();
return () => {
isActive = false;
};
}, [selectedTaskId]);
// Measure container width on mount and resize
useEffect(() => {
if (!containerRef.current) return;
@@ -82,9 +122,9 @@ export function KanbanWithPreview() {
}, [isOpen, selectedTaskId, selectedTask, close]);
const handleNavigateToTask = useCallback(
(task: Task) => {
(task: Task, sessionId: string) => {
close();
router.push(`/task/${task.id}`);
router.push(`/task/${task.id}/${sessionId}`);
},
[close, router]
);
@@ -149,6 +189,8 @@ export function KanbanWithPreview() {
? minKanbanWidthPx
: (isOpen ? containerWidth - previewWidthPx : containerWidth);
const activeSessionId = selectedTaskId ? selectedTaskSessionId : null;
return (
<div ref={containerRef} className="h-screen w-full flex flex-col bg-background relative">
{shouldFloat ? (
@@ -192,7 +234,16 @@ export function KanbanWithPreview() {
{/* Panel content */}
<div className="flex-1 min-w-0 overflow-hidden">
<TaskPreviewPanel task={selectedTask} onClose={close} onMaximize={handleNavigateToTask} />
<TaskPreviewPanel
task={selectedTask}
sessionId={activeSessionId}
onClose={close}
onMaximize={
activeSessionId
? (task) => handleNavigateToTask(task, activeSessionId)
: undefined
}
/>
</div>
</div>
</>
@@ -231,7 +282,16 @@ export function KanbanWithPreview() {
{/* Panel content */}
<div className="flex-1 min-w-0 overflow-hidden">
<TaskPreviewPanel task={selectedTask} onClose={close} onMaximize={handleNavigateToTask} />
<TaskPreviewPanel
task={selectedTask}
sessionId={activeSessionId}
onClose={close}
onMaximize={
activeSessionId
? (task) => handleNavigateToTask(task, activeSessionId)
: undefined
}
/>
</div>
</div>
)}
+55 -123
View File
@@ -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: <IconLoader2 className="h-3 w-3 animate-spin" />,
variant: "default" as const,
};
case "COMPLETED":
return {
label: "Completed",
icon: <IconCircleCheck className="h-3 w-3" />,
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: <IconCircleX className="h-3 w-3" />,
variant: "destructive" as const,
};
case "BLOCKED":
case "WAITING_FOR_INPUT":
return {
label: state === "BLOCKED" ? "Blocked" : "Waiting for Input",
icon: <IconAlertTriangle className="h-3 w-3" />,
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 (
<div className="flex h-full w-full flex-col border-l bg-background">
<div className="flex items-center justify-between border-b px-4 py-3">
<h2 className="text-sm font-semibold">Task Preview</h2>
<Button variant="ghost" size="icon" className="h-8 w-8 cursor-pointer" onClick={onClose}>
<IconX className="h-4 w-4" />
<span className="sr-only">Close preview</span>
</Button>
</div>
<div className="flex flex-1 items-center justify-center p-8">
<p className="text-sm text-muted-foreground">Select a task to preview</p>
</div>
</div>
);
}
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 (
<div className="flex h-full w-full flex-col border-l bg-background">
{/* Header */}
<div className="flex items-center justify-between border-b px-4 py-3">
<h2 className="text-sm font-semibold">Task Preview</h2>
<h2 className="text-sm font-semibold truncate">
{task?.title ?? "Task Chat"}
</h2>
<div className="flex items-center gap-1">
{onMaximize && (
{onMaximize && activeSessionId && task && (
<Button
variant="ghost"
size="icon"
className="h-8 w-8 cursor-pointer"
onClick={() => onMaximize(task)}
onClick={() => onMaximize(task, activeSessionId)}
title="Open full page"
>
<IconArrowsMaximize className="h-4 w-4" />
@@ -122,43 +76,21 @@ export function TaskPreviewPanel({ task, onClose, onMaximize }: TaskPreviewPanel
</div>
{/* Content */}
<div className="flex-1 overflow-y-auto p-4">
<div className="space-y-4">
{/* Repository */}
{repoName && (
<div>
<p className="text-xs font-medium text-muted-foreground mb-1">Repository</p>
<p className="text-sm">{repoName}</p>
</div>
)}
{/* Title */}
<div>
<p className="text-xs font-medium text-muted-foreground mb-1">Title</p>
<h3 className="text-base font-semibold">{task.title}</h3>
<div className="flex-1 min-h-0 p-4 flex flex-col">
{task ? (
<TaskChatPanel
agents={AGENTS}
onSend={handleSendMessage}
isLoading={isLoadingMessages}
isAgentWorking={isActiveSessionWorking}
taskId={task.id}
taskDescription={task.description}
/>
) : (
<div className="flex flex-1 items-center justify-center text-sm text-muted-foreground">
Select a task to start chatting
</div>
{/* State */}
{task.state && (
<div>
<p className="text-xs font-medium text-muted-foreground mb-1">Status</p>
<Badge variant={stateDisplay.variant} className={stateDisplay.className}>
{stateDisplay.icon && <span className="mr-1">{stateDisplay.icon}</span>}
{stateDisplay.label}
</Badge>
</div>
)}
{/* Description */}
{task.description && (
<div>
<p className="text-xs font-medium text-muted-foreground mb-1">Description</p>
<p className="text-sm text-muted-foreground whitespace-pre-wrap">
{task.description}
</p>
</div>
)}
</div>
)}
</div>
</div>
);
@@ -42,7 +42,7 @@ const adapters: MessageAdapter[] = [
<ChatMessage
comment={comment}
label="Task"
className="ml-auto bg-primary/20 text-foreground border border-primary/40"
className="bg-amber-500/10 text-foreground border-amber-500/30"
showRichBlocks
/>
);
@@ -52,7 +52,7 @@ const adapters: MessageAdapter[] = [
<ChatMessage
comment={comment}
label="You"
className="ml-auto bg-primary text-primary-foreground"
className="bg-primary/10 text-foreground border-primary/30"
/>
);
}
@@ -60,7 +60,7 @@ const adapters: MessageAdapter[] = [
<ChatMessage
comment={comment}
label="Agent"
className="bg-muted text-foreground"
className="bg-muted/40 text-foreground border-border/60"
showRichBlocks={comment.type === 'message' || comment.type === 'content' || !comment.type}
/>
);
@@ -18,7 +18,7 @@ export function ChatMessage({ comment, label, className, showRichBlocks }: ChatM
const isUser = comment.author_type === 'user';
return (
<div className={cn('max-w-[85%] rounded-lg px-4 py-3 text-sm', className)}>
<div className={cn('w-full rounded-lg border px-4 py-3 text-sm', className)}>
<p className="text-[11px] uppercase tracking-wide mb-2 opacity-70">
{label}
{comment.requests_input ? (
@@ -17,10 +17,10 @@ export function StatusMessage({ comment }: { comment: Message }) {
return (
<div
className={cn(
'max-w-[85%] rounded-lg border px-4 py-3 text-sm',
'w-full rounded-lg border px-4 py-3 text-sm',
comment.type === 'error'
? 'border-red-500/40 bg-red-500/10 text-foreground'
: 'border-border/50 bg-muted/40 text-foreground'
: 'border-border/60 bg-muted/30 text-foreground'
)}
>
<div className="flex items-center gap-2 mb-2 text-[11px] uppercase tracking-wide opacity-70">
@@ -9,7 +9,7 @@ export function ThinkingMessage({ comment }: { comment: Message }) {
const text = metadata?.thinking ?? comment.content;
if (!text) return null;
return (
<div className="max-w-[85%] rounded-lg border border-border/50 bg-background/60 px-4 py-3 text-sm">
<div className="w-full rounded-lg border border-border/60 bg-background/60 px-4 py-3 text-sm">
<div className="flex items-center gap-2 text-muted-foreground mb-2 uppercase tracking-wide text-[11px]">
<IconBrain className="h-3.5 w-3.5" />
<span>Thinking</span>
@@ -20,7 +20,7 @@ export function TodoMessage({ comment }: { comment: Message }) {
const progress = Math.round((completed / todoItems.length) * 100);
return (
<div className="max-w-[85%] rounded-lg border border-border/50 bg-background/60 px-4 py-3 text-sm">
<div className="w-full rounded-lg border border-border/60 bg-background/60 px-4 py-3 text-sm">
<div className="flex items-center gap-2 text-muted-foreground mb-2 uppercase tracking-wide text-[11px]">
<IconListCheck className="h-3.5 w-3.5" />
<span>Todos</span>
@@ -77,7 +77,7 @@ export function ToolCallMessage({ comment }: { comment: Message }) {
}
return (
<div className="rounded-md border border-border/40 bg-muted/20 overflow-hidden max-w-[85%]">
<div className="w-full rounded-md border border-border/50 bg-muted/20 overflow-hidden">
<button
type="button"
onClick={() => hasDetails && setIsExpanded(!isExpanded)}
+96 -22
View File
@@ -1,6 +1,7 @@
'use client';
import { useState, useEffect } from 'react';
import { useCallback, useState, useEffect } from 'react';
import { useRouter } from 'next/navigation';
import {
IconStack2,
IconLoader2,
@@ -20,8 +21,10 @@ import {
import { Badge } from '@kandev/ui/badge';
import { Tooltip, TooltipContent, TooltipTrigger } from '@kandev/ui/tooltip';
import { TaskCreateDialog } from '../task-create-dialog';
import { useAppStore } from '@/components/state-provider';
import { getWebSocketClient } from '@/lib/ws/connection';
type SessionStatus = 'running' | 'waiting_input' | 'complete' | 'failed';
type SessionStatus = 'running' | 'waiting_input' | 'complete' | 'failed' | 'cancelled';
type Session = {
id: string;
@@ -31,12 +34,17 @@ type Session = {
status: SessionStatus;
};
// Dummy data - will be replaced with real data later
const DUMMY_SESSIONS: Session[] = [
{ id: '1', number: 1, agentProfile: 'Gemini Flash 2.0', startedAt: new Date(Date.now() - 114000), status: 'running' },
{ id: '2', number: 2, agentProfile: 'Auggie Sonnet 4.5', startedAt: new Date(Date.now() - 192000), status: 'complete' },
{ id: '3', number: 3, agentProfile: 'Gemini Flash 2.0', startedAt: new Date(Date.now() - 45000), status: 'waiting_input' },
];
type TaskSessionResponse = {
id: string;
agent_profile_id?: string;
started_at: string;
state: string;
};
type ListTaskSessionsResponse = {
sessions: TaskSessionResponse[];
total: number;
};
// Format duration from start time
function formatDuration(startedAt: Date, isRunning: boolean): string {
@@ -65,6 +73,7 @@ function getStatusIcon(status: SessionStatus) {
case 'waiting_input':
return <IconAlertCircle className="h-3.5 w-3.5 text-yellow-500" />;
case 'failed':
case 'cancelled':
return <IconX className="h-3.5 w-3.5 text-red-500" />;
}
}
@@ -79,19 +88,74 @@ function getStatusLabel(status: SessionStatus) {
return 'Waiting for input';
case 'failed':
return 'Failed';
case 'cancelled':
return 'Cancelled';
}
}
function mapSessionStatus(state: string): SessionStatus {
switch (state) {
case 'RUNNING':
case 'STARTING':
return 'running';
case 'WAITING_FOR_INPUT':
return 'waiting_input';
case 'COMPLETED':
return 'complete';
case 'FAILED':
return 'failed';
case 'CANCELLED':
return 'cancelled';
default:
return 'running';
}
}
type SessionsDropdownProps = {
taskId: string | null;
activeSessionId?: string | null;
taskTitle?: string;
taskDescription?: string;
};
export function SessionsDropdown({ taskTitle = '', taskDescription = '' }: SessionsDropdownProps) {
const [sessions] = useState<Session[]>(DUMMY_SESSIONS);
const [selectedSessionId, setSelectedSessionId] = useState<string | null>(null);
export function SessionsDropdown({ taskId, activeSessionId = null, taskTitle = '', taskDescription = '' }: SessionsDropdownProps) {
const [sessions, setSessions] = useState<Session[]>([]);
const [currentTime, setCurrentTime] = useState(() => Date.now());
const [showNewSessionDialog, setShowNewSessionDialog] = useState(false);
const agentProfiles = useAppStore((state) => state.agentProfiles.items);
const router = useRouter();
const visibleSessions = taskId ? sessions : [];
const fetchSessions = useCallback(async (targetTaskId: string) => {
const client = getWebSocketClient();
if (!client) return;
try {
const response = await client.request<ListTaskSessionsResponse>('task.session.list', {
task_id: targetTaskId,
});
const total = response.sessions.length;
const mapped = response.sessions.map((session, index) => {
const label = agentProfiles.find((profile) => profile.id === session.agent_profile_id)?.label
?? session.agent_profile_id
?? 'Unknown agent';
return {
id: session.id,
number: total - index,
agentProfile: label,
startedAt: new Date(session.started_at),
status: mapSessionStatus(session.state),
};
});
setSessions(mapped);
} catch (error) {
console.error('Failed to load task sessions:', error);
}
}, [agentProfiles]);
const handleOpenChange = useCallback((open: boolean) => {
if (!open || !taskId) return;
fetchSessions(taskId);
}, [fetchSessions, taskId]);
// Update timer every second for running sessions
useEffect(() => {
@@ -111,10 +175,21 @@ export function SessionsDropdown({ taskTitle = '', taskDescription = '' }: Sessi
console.log('Cancel session:', sessionId);
};
const handleSelectSession = (sessionId: string) => {
setSelectedSessionId(sessionId);
// TODO: Implement session switching
console.log('Select session:', sessionId);
const handleSelectSession = async (sessionId: string) => {
if (!taskId) return;
const client = getWebSocketClient();
if (!client) return;
try {
await client.request('task.session.resume', {
task_id: taskId,
task_session_id: sessionId,
}, 15000);
await fetchSessions(taskId);
router.push(`/task/${taskId}/${sessionId}`);
} catch (error) {
console.error('Failed to resume task session:', error);
}
};
const handleCreateSession = (data: {
@@ -129,7 +204,7 @@ export function SessionsDropdown({ taskTitle = '', taskDescription = '' }: Sessi
return (
<>
<DropdownMenu>
<DropdownMenu onOpenChange={handleOpenChange}>
<DropdownMenuTrigger asChild>
<Button
variant="ghost"
@@ -138,13 +213,13 @@ export function SessionsDropdown({ taskTitle = '', taskDescription = '' }: Sessi
>
<IconStack2 className="h-4 w-4 text-muted-foreground" />
<Badge variant="secondary" className="h-5 px-1.5 text-xs font-normal">
{sessions.length}
{visibleSessions.length}
</Badge>
</Button>
</DropdownMenuTrigger>
<DropdownMenuContent align="start" className="w-[480px]">
<div className="flex items-center justify-between px-2 py-1.5">
<span className="text-xs font-medium text-muted-foreground">Agent Sessions</span>
<span className="text-xs font-medium text-muted-foreground">Task Sessions</span>
<button
type="button"
onClick={() => setShowNewSessionDialog(true)}
@@ -156,13 +231,13 @@ export function SessionsDropdown({ taskTitle = '', taskDescription = '' }: Sessi
</div>
<DropdownMenuSeparator />
<div className="max-h-[300px] overflow-y-auto">
{sessions.length === 0 ? (
{visibleSessions.length === 0 ? (
<div className="px-2 py-6 text-center text-sm text-muted-foreground">
No sessions yet
</div>
) : (
<div className="space-y-0.5">
{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
</>
);
}
+10 -1
View File
@@ -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({
</>
)}
<span className="text-sm font-medium">{taskTitle ?? 'Task details'}</span>
<SessionsDropdown taskTitle={taskTitle} taskDescription={taskDescription} />
<SessionsDropdown
taskId={taskId ?? null}
activeSessionId={activeSessionId ?? null}
taskTitle={taskTitle}
taskDescription={taskDescription}
/>
{displayBranch && (
<Popover open={popoverOpen} onOpenChange={setPopoverOpen}>
<Tooltip>
+74
View File
@@ -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<Record<string, string>>({});
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,
};
}
+2 -1
View File
@@ -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') {
+5
View File
@@ -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<Task>(`/api/v1/tasks/${taskId}`, options);
}
export async function listTaskSessions(taskId: string, options?: ApiRequestOptions) {
return fetchJson<TaskSessionsResponse>(`/api/v1/tasks/${taskId}/sessions`, options);
}
export async function listTaskSessionMessages(
taskSessionId: string,
params?: { limit?: number; before?: string; after?: string; sort?: 'asc' | 'desc' },
+31
View File
@@ -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<string, unknown> | null;
agent_profile_snapshot?: Record<string, unknown> | null;
executor_snapshot?: Record<string, unknown> | null;
environment_snapshot?: Record<string, unknown> | null;
repository_snapshot?: Record<string, unknown> | 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;