diff --git a/.gitignore b/.gitignore index e3690f0419..86d14c34e5 100644 --- a/.gitignore +++ b/.gitignore @@ -81,3 +81,4 @@ pnpm-lock.yaml # TypeScript *.tsbuildinfo apps/landing/next-env.d.ts +package-lock.json diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index efc6a35545..8b0866a9bd 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -394,6 +394,11 @@ func main() { agentHandlers := agenthandlers.NewHandlers(agentCtrl, log) agentHandlers.RegisterHandlers(gateway.Dispatcher) log.Info("Registered Agent Manager WebSocket handlers") + + // Register workspace file handlers + workspaceFileHandlers := agenthandlers.NewWorkspaceFileHandlers(lifecycleMgr, log) + workspaceFileHandlers.RegisterHandlers(gateway.Dispatcher) + log.Info("Registered Workspace File WebSocket handlers") } // Start the WebSocket hub @@ -613,6 +618,25 @@ func main() { log.Info("Subscribed to git status events for WebSocket broadcasting") } + // Subscribe to file change events and broadcast to WebSocket subscribers + _, err = eventBus.Subscribe(events.BuildFileChangeWildcardSubject(), func(ctx context.Context, event *bus.Event) error { + data := event.Data + taskID, _ := data["task_id"].(string) + if taskID == "" { + return nil + } + + // Broadcast file change notification to task subscribers + notification, _ := ws.NewNotification("workspace.file.changes", data) + gateway.Hub.BroadcastToTask(taskID, notification) + return nil + }) + if err != nil { + log.Error("Failed to subscribe to file change events", zap.Error(err)) + } else { + log.Info("Subscribed to file change events for WebSocket broadcasting") + } + // ============================================ // HTTP SERVER (WebSocket + HTTP endpoints) // ============================================ @@ -878,6 +902,12 @@ func (a *lifecycleAdapter) GetRecoveredInstances() []executor.RecoveredInstanceI return result } +// IsAgentRunningForTask checks if an agent is actually running for a task +// This probes the actual agent (Docker container or standalone process) +func (a *lifecycleAdapter) IsAgentRunningForTask(ctx context.Context, taskID string) bool { + return a.mgr.IsAgentRunningForTask(ctx, taskID) +} + // corsMiddleware returns a CORS middleware for HTTP and WebSocket connections func corsMiddleware() gin.HandlerFunc { return func(c *gin.Context) { diff --git a/apps/backend/internal/agent/agentctl/acp.go b/apps/backend/internal/agent/agentctl/acp.go index 0666009d7c..b2a337c9e1 100644 --- a/apps/backend/internal/agent/agentctl/acp.go +++ b/apps/backend/internal/agent/agentctl/acp.go @@ -240,14 +240,6 @@ func (c *Client) CloseACPStream() { } } -// Close closes all connections -func (c *Client) Close() { - c.CloseACPStream() - c.ClosePermissionStream() - c.CloseGitStatusStream() - c.CloseFilesStream() -} - // StreamPermissions opens a WebSocket connection for streaming permission requests func (c *Client) StreamPermissions(ctx context.Context, handler func(*PermissionNotification)) error { wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/permissions/stream" diff --git a/apps/backend/internal/agent/agentctl/client.go b/apps/backend/internal/agent/agentctl/client.go index a548e73a99..7a9d87dea5 100644 --- a/apps/backend/internal/agent/agentctl/client.go +++ b/apps/backend/internal/agent/agentctl/client.go @@ -10,6 +10,7 @@ import ( "time" "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/process" "github.com/kandev/kandev/internal/common/logger" "go.uber.org/zap" ) @@ -21,11 +22,11 @@ type Client struct { logger *logger.Logger // WebSocket connections for streaming - acpConn *websocket.Conn - permissionConn *websocket.Conn - gitStatusConn *websocket.Conn - filesConn *websocket.Conn - mu sync.RWMutex + acpConn *websocket.Conn + permissionConn *websocket.Conn + gitStatusConn *websocket.Conn + fileChangesConn *websocket.Conn + mu sync.RWMutex } // StatusResponse from agentctl @@ -34,6 +35,11 @@ type StatusResponse struct { ProcessInfo map[string]interface{} `json:"process_info"` } +// IsAgentRunning returns true if the agent process is running +func (s *StatusResponse) IsAgentRunning() bool { + return s.AgentStatus == "running" +} + // NewClient creates a new agentctl client func NewClient(host string, port int, log *logger.Logger) *Client { return &Client{ @@ -164,43 +170,19 @@ func (c *Client) BaseURL() string { return c.baseURL } -// GitStatusUpdate represents a git status update -type GitStatusUpdate struct { - Timestamp time.Time `json:"timestamp"` - Modified []string `json:"modified"` - Added []string `json:"added"` - Deleted []string `json:"deleted"` - Untracked []string `json:"untracked"` - Renamed []string `json:"renamed"` - Ahead int `json:"ahead"` - Behind int `json:"behind"` - Branch string `json:"branch"` - RemoteBranch string `json:"remote_branch,omitempty"` - Files map[string]FileInfo `json:"files,omitempty"` -} - -// FileInfo represents information about a file -type FileInfo struct { - Path string `json:"path"` - Status string `json:"status"` - Additions int `json:"additions,omitempty"` - Deletions int `json:"deletions,omitempty"` - OldPath string `json:"old_path,omitempty"` - Diff string `json:"diff,omitempty"` -} - -// FileListUpdate represents a file listing update -type FileListUpdate struct { - Timestamp time.Time `json:"timestamp"` - Files []FileEntry `json:"files"` -} - -// FileEntry represents a file in the workspace -type FileEntry struct { - Path string `json:"path"` - IsDir bool `json:"is_dir"` - Size int64 `json:"size,omitempty"` -} +// Re-export types from process package for convenience +type ( + GitStatusUpdate = process.GitStatusUpdate + FileInfo = process.FileInfo + FileListUpdate = process.FileListUpdate + FileEntry = process.FileEntry + FileTreeNode = process.FileTreeNode + FileTreeRequest = process.FileTreeRequest + FileTreeResponse = process.FileTreeResponse + FileContentRequest = process.FileContentRequest + FileContentResponse = process.FileContentResponse + FileChangeNotification = process.FileChangeNotification +) // StreamGitStatus opens a WebSocket connection for streaming git status updates func (c *Client) StreamGitStatus(ctx context.Context, handler func(*GitStatusUpdate)) error { @@ -250,54 +232,6 @@ func (c *Client) StreamGitStatus(ctx context.Context, handler func(*GitStatusUpd return nil } -// StreamFiles opens a WebSocket connection for streaming file listing updates -func (c *Client) StreamFiles(ctx context.Context, handler func(*FileListUpdate)) error { - wsURL := "ws" + c.baseURL[4:] + "/api/v1/workspace/files/stream" - - conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) - if err != nil { - return fmt.Errorf("failed to connect to files stream: %w", err) - } - - c.mu.Lock() - c.filesConn = conn - c.mu.Unlock() - - c.logger.Info("connected to files stream", zap.String("url", wsURL)) - - // Read messages in a goroutine - go func() { - defer func() { - c.mu.Lock() - c.filesConn = nil - c.mu.Unlock() - conn.Close() - }() - - for { - _, message, err := conn.ReadMessage() - if err != nil { - if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { - c.logger.Info("files stream closed normally") - } else { - c.logger.Debug("files stream error", zap.Error(err)) - } - return - } - - var update FileListUpdate - if err := json.Unmarshal(message, &update); err != nil { - c.logger.Warn("failed to parse file list update", zap.Error(err)) - continue - } - - handler(&update) - } - }() - - return nil -} - // CloseGitStatusStream closes the git status stream connection func (c *Client) CloseGitStatusStream() { c.mu.Lock() @@ -309,13 +243,123 @@ func (c *Client) CloseGitStatusStream() { } } -// CloseFilesStream closes the files stream connection -func (c *Client) CloseFilesStream() { +// StreamFileChanges opens a WebSocket connection for streaming file change notifications +func (c *Client) StreamFileChanges(ctx context.Context, handler func(*FileChangeNotification)) error { + wsURL := "ws" + c.baseURL[4:] + "/api/v1/workspace/file-changes/stream" + + conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) + if err != nil { + return fmt.Errorf("failed to connect to file changes stream: %w", err) + } + + c.mu.Lock() + c.fileChangesConn = conn + c.mu.Unlock() + + c.logger.Info("connected to file changes stream", zap.String("url", wsURL)) + + // Read messages in a goroutine + go func() { + defer func() { + c.mu.Lock() + c.fileChangesConn = nil + c.mu.Unlock() + conn.Close() + }() + + for { + _, message, err := conn.ReadMessage() + if err != nil { + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + c.logger.Info("file changes stream closed normally") + } else { + c.logger.Debug("file changes stream error", zap.Error(err)) + } + return + } + + var notification FileChangeNotification + if err := json.Unmarshal(message, ¬ification); err != nil { + c.logger.Warn("failed to parse file change notification", zap.Error(err)) + continue + } + + handler(¬ification) + } + }() + + return nil +} + +// RequestFileTree requests a file tree via HTTP GET +func (c *Client) RequestFileTree(ctx context.Context, path string, depth int) (*FileTreeResponse, error) { + url := fmt.Sprintf("%s/api/v1/workspace/tree?path=%s&depth=%d", c.baseURL, path, depth) + + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to request file tree: %w", err) + } + defer resp.Body.Close() + + var response FileTreeResponse + if err := json.NewDecoder(resp.Body).Decode(&response); err != nil { + return nil, fmt.Errorf("failed to parse response: %w", err) + } + + if response.Error != "" { + return nil, fmt.Errorf("file tree error: %s", response.Error) + } + + return &response, nil +} + +// RequestFileContent requests file content via HTTP GET +func (c *Client) RequestFileContent(ctx context.Context, path string) (*FileContentResponse, error) { + url := fmt.Sprintf("%s/api/v1/workspace/file/content?path=%s", c.baseURL, path) + + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to request file content: %w", err) + } + defer resp.Body.Close() + + var response FileContentResponse + if err := json.NewDecoder(resp.Body).Decode(&response); err != nil { + return nil, fmt.Errorf("failed to parse response: %w", err) + } + + if response.Error != "" { + return nil, fmt.Errorf("file content error: %s", response.Error) + } + + return &response, nil +} + +// CloseFileChangesStream closes the file changes stream connection +func (c *Client) CloseFileChangesStream() { c.mu.Lock() defer c.mu.Unlock() - if c.filesConn != nil { - c.filesConn.Close() - c.filesConn = nil + if c.fileChangesConn != nil { + c.fileChangesConn.Close() + c.fileChangesConn = nil } } + +// Close closes all connections (ACP, permissions, git status, file changes) +func (c *Client) Close() { + c.CloseACPStream() + c.ClosePermissionStream() + c.CloseGitStatusStream() + c.CloseFileChangesStream() +} diff --git a/apps/backend/internal/agent/docker/client.go b/apps/backend/internal/agent/docker/client.go index 3db21451d5..b9331ccfbd 100644 --- a/apps/backend/internal/agent/docker/client.go +++ b/apps/backend/internal/agent/docker/client.go @@ -284,6 +284,15 @@ func (c *Client) GetContainerInfo(ctx context.Context, containerID string) (*Con return info, nil } +// IsContainerRunning returns true if the container exists and is in "running" state +func (c *Client) IsContainerRunning(ctx context.Context, containerID string) (bool, error) { + info, err := c.GetContainerInfo(ctx, containerID) + if err != nil { + return false, err + } + return info.State == "running", nil +} + // GetContainerIP returns the IP address of a container func (c *Client) GetContainerIP(ctx context.Context, containerID string) (string, error) { c.logger.Debug("Getting container IP", zap.String("container_id", containerID)) diff --git a/apps/backend/internal/agent/handlers/workspace_file_handlers.go b/apps/backend/internal/agent/handlers/workspace_file_handlers.go new file mode 100644 index 0000000000..6228024c97 --- /dev/null +++ b/apps/backend/internal/agent/handlers/workspace_file_handlers.go @@ -0,0 +1,110 @@ +package handlers + +import ( + "context" + "fmt" + + "github.com/kandev/kandev/internal/agent/lifecycle" + "github.com/kandev/kandev/internal/common/logger" + ws "github.com/kandev/kandev/pkg/websocket" + "go.uber.org/zap" +) + +// WorkspaceFileHandlers handles workspace file operations +type WorkspaceFileHandlers struct { + lifecycle *lifecycle.Manager + logger *logger.Logger +} + +// NewWorkspaceFileHandlers creates new workspace file handlers +func NewWorkspaceFileHandlers(lm *lifecycle.Manager, log *logger.Logger) *WorkspaceFileHandlers { + return &WorkspaceFileHandlers{ + lifecycle: lm, + logger: log.WithFields(zap.String("component", "workspace-file-handlers")), + } +} + +// RegisterHandlers registers workspace file handlers with the dispatcher +func (h *WorkspaceFileHandlers) RegisterHandlers(d *ws.Dispatcher) { + d.RegisterFunc(ws.ActionWorkspaceFileTreeGet, h.wsGetFileTree) + d.RegisterFunc(ws.ActionWorkspaceFileContentGet, h.wsGetFileContent) +} + +// wsGetFileTree handles workspace.tree.get action +func (h *WorkspaceFileHandlers) wsGetFileTree(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + TaskID string `json:"task_id"` + Path string `json:"path"` + Depth int `json:"depth"` + } + + 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) + } + + // Get agent instance for this task + instance, found := h.lifecycle.GetInstanceByTaskID(req.TaskID) + if !found { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "No agent found for task", nil) + } + + // Get agentctl client + client := instance.GetAgentCtlClient() + if client == nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Agent client not available", nil) + } + + // Request file tree from agentctl + response, err := client.RequestFileTree(ctx, req.Path, req.Depth) + if err != nil { + h.logger.Error("failed to get file tree", zap.Error(err), zap.String("task_id", req.TaskID)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, fmt.Sprintf("Failed to get file tree: %v", err), nil) + } + + return ws.NewResponse(msg.ID, msg.Action, response) +} + +// wsGetFileContent handles workspace.file.get action +func (h *WorkspaceFileHandlers) wsGetFileContent(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + TaskID string `json:"task_id"` + Path string `json:"path"` + } + + 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.Path == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "path is required", nil) + } + + // Get agent instance for this task + instance, found := h.lifecycle.GetInstanceByTaskID(req.TaskID) + if !found { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "No agent found for task", nil) + } + + // Get agentctl client + client := instance.GetAgentCtlClient() + if client == nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Agent client not available", nil) + } + + // Request file content from agentctl + response, err := client.RequestFileContent(ctx, req.Path) + if err != nil { + h.logger.Error("failed to get file content", zap.Error(err), zap.String("task_id", req.TaskID), zap.String("path", req.Path)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, fmt.Sprintf("Failed to get file content: %v", err), nil) + } + + return ws.NewResponse(msg.ID, msg.Action, response) +} diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index acff450786..10c65bb5b0 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -57,6 +57,11 @@ type AgentInstance struct { messageMu sync.Mutex } +// GetAgentCtlClient returns the agentctl client for this instance +func (ai *AgentInstance) GetAgentCtlClient() *agentctl.Client { + return ai.agentctl +} + // LaunchRequest contains parameters for launching an agent type LaunchRequest struct { TaskID string @@ -740,7 +745,10 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta // 5. Set up git status stream to track workspace changes go m.handleGitStatusStream(instance) - // 5. Send the task prompt if provided + // 6. Set up file changes stream to track filesystem changes + go m.handleFileChangesStream(instance) + + // 7. Send the task prompt if provided if taskDescription != "" { m.logger.Info("sending ACP prompt", zap.String("instance_id", instance.ID), @@ -885,6 +893,7 @@ func (m *Manager) reconnectStreams(instance *AgentInstance) { go m.handleUpdatesStream(instance) go m.handlePermissionStream(instance) go m.handleGitStatusStream(instance) + go m.handleFileChangesStream(instance) // Mark the instance as READY so it can accept prompts m.mu.Lock() @@ -1184,6 +1193,60 @@ func (m *Manager) publishGitStatus(instance *AgentInstance, update *agentctl.Git } } +// handleFileChangesStream handles streaming file change notifications from the agent workspace +func (m *Manager) handleFileChangesStream(instance *AgentInstance) { + ctx := context.Background() + + // Retry connection with exponential backoff + maxRetries := 5 + backoff := 1 * time.Second + + for attempt := 1; attempt <= maxRetries; attempt++ { + err := instance.agentctl.StreamFileChanges(ctx, func(notification *agentctl.FileChangeNotification) { + m.publishFileChange(instance, notification) + }) + + if err == nil { + // Connection closed normally + return + } + + if attempt < maxRetries { + time.Sleep(backoff) + backoff *= 2 // Exponential backoff + } + } + + m.logger.Error("failed to connect to file changes stream after retries", + zap.String("instance_id", instance.ID), + zap.Int("max_retries", maxRetries)) +} + +// publishFileChange publishes a file change notification to the event bus +func (m *Manager) publishFileChange(instance *AgentInstance, notification *agentctl.FileChangeNotification) { + if m.eventBus == nil { + return + } + + data := map[string]interface{}{ + "task_id": instance.TaskID, + "agent_id": instance.ID, + "path": notification.Path, + "operation": notification.Operation, + "timestamp": notification.Timestamp.Format(time.RFC3339Nano), + } + + event := bus.NewEvent(events.FileChangeNotified, "agent-manager", data) + subject := events.BuildFileChangeSubject(instance.TaskID) + + if err := m.eventBus.Publish(context.Background(), subject, event); err != nil { + m.logger.Error("failed to publish file change event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } +} + // publishPromptComplete publishes a prompt_complete event when an agent finishes responding // This is used to save the agent's response as a comment on the task func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage string) { @@ -1714,6 +1777,31 @@ func (m *Manager) ListInstances() []*AgentInstance { return result } +// IsAgentRunningForTask checks if an agent is actually running for a task +// This probes agentctl's status endpoint to verify the agent process is running +func (m *Manager) IsAgentRunningForTask(ctx context.Context, taskID string) bool { + // First check if we have an instance tracked for this task + instance, exists := m.GetInstanceByTaskID(taskID) + if !exists { + return false + } + + // Probe agentctl status to verify the agent process is running + if instance.agentctl == nil { + return false + } + + status, err := instance.agentctl.GetStatus(ctx) + if err != nil { + m.logger.Debug("failed to get agentctl status", + zap.String("task_id", taskID), + zap.Error(err)) + return false + } + + return status.IsAgentRunning() +} + // UpdateStatus updates the status of an instance func (m *Manager) UpdateStatus(instanceID string, status v1.AgentStatus) error { m.mu.Lock() diff --git a/apps/backend/internal/agentctl/api/server.go b/apps/backend/internal/agentctl/api/server.go index a7ba1bcaff..ae552a8c0a 100644 --- a/apps/backend/internal/agentctl/api/server.go +++ b/apps/backend/internal/agentctl/api/server.go @@ -77,6 +77,11 @@ func (s *Server) setupRoutes() { // Workspace monitoring (git status, files) api.GET("/workspace/git-status/stream", s.handleGitStatusStreamWS) api.GET("/workspace/files/stream", s.handleFilesStreamWS) + api.GET("/workspace/file-changes/stream", s.handleFileChangesStreamWS) + + // Workspace file operations (simple HTTP) + api.GET("/workspace/tree", s.handleFileTree) + api.GET("/workspace/file/content", s.handleFileContent) } } diff --git a/apps/backend/internal/agentctl/api/workspace.go b/apps/backend/internal/agentctl/api/workspace.go index 71165e6a22..44c9be2b0f 100644 --- a/apps/backend/internal/agentctl/api/workspace.go +++ b/apps/backend/internal/agentctl/api/workspace.go @@ -5,34 +5,10 @@ import ( "github.com/gin-gonic/gin" "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/process" "go.uber.org/zap" ) -// GitStatusUpdate represents a git status update -type GitStatusUpdate struct { - Timestamp string `json:"timestamp"` - Modified []string `json:"modified"` - Added []string `json:"added"` - Deleted []string `json:"deleted"` - Untracked []string `json:"untracked"` - Renamed []string `json:"renamed"` - Ahead int `json:"ahead"` - Behind int `json:"behind"` - Branch string `json:"branch"` - RemoteBranch string `json:"remote_branch,omitempty"` - Files map[string]FileInfo `json:"files,omitempty"` -} - -// FileInfo represents information about a file -type FileInfo struct { - Path string `json:"path"` - Status string `json:"status"` // modified, added, deleted, untracked, renamed - Additions int `json:"additions,omitempty"` - Deletions int `json:"deletions,omitempty"` - OldPath string `json:"old_path,omitempty"` // For renamed files - Diff string `json:"diff,omitempty"` -} - // handleGitStatusStreamWS streams git status updates via WebSocket func (s *Server) handleGitStatusStreamWS(c *gin.Context) { conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil) @@ -124,3 +100,92 @@ func (s *Server) handleFilesStreamWS(c *gin.Context) { } } } + +// handleFileChangesStreamWS streams filesystem change notifications via WebSocket +func (s *Server) handleFileChangesStreamWS(c *gin.Context) { + conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + s.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + // Subscribe to file change notifications + sub := s.procMgr.GetWorkspaceTracker().SubscribeFileChanges() + defer s.procMgr.GetWorkspaceTracker().UnsubscribeFileChanges(sub) + + // Handle WebSocket close + closeCh := make(chan struct{}) + go func() { + for { + _, _, err := conn.ReadMessage() + if err != nil { + close(closeCh) + return + } + } + }() + + // Stream file change notifications to client + for { + select { + case notification, ok := <-sub: + if !ok { + return + } + data, err := json.Marshal(notification) + if err != nil { + s.logger.Error("failed to marshal file change notification", zap.Error(err)) + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + case <-closeCh: + return + } + } +} + +// handleFileTree handles file tree requests via HTTP GET +func (s *Server) handleFileTree(c *gin.Context) { + path := c.Query("path") + depth := 1 + if d := c.Query("depth"); d != "" { + if _, err := json.Number(d).Int64(); err == nil { + depth = int(mustParseInt(d)) + } + } + + tree, err := s.procMgr.GetWorkspaceTracker().GetFileTree(path, depth) + if err != nil { + c.JSON(400, process.FileTreeResponse{Error: err.Error()}) + return + } + + c.JSON(200, process.FileTreeResponse{Root: tree}) +} + +// handleFileContent handles file content requests via HTTP GET +func (s *Server) handleFileContent(c *gin.Context) { + path := c.Query("path") + if path == "" { + c.JSON(400, process.FileContentResponse{Error: "path is required"}) + return + } + + content, size, err := s.procMgr.GetWorkspaceTracker().GetFileContent(path) + if err != nil { + c.JSON(400, process.FileContentResponse{Path: path, Error: err.Error(), Size: size}) + return + } + + c.JSON(200, process.FileContentResponse{Path: path, Content: content, Size: size}) +} + +// mustParseInt parses a string to int, returns 0 on error +func mustParseInt(s string) int { + var n int + json.Unmarshal([]byte(s), &n) + return n +} diff --git a/apps/backend/internal/agentctl/process/types.go b/apps/backend/internal/agentctl/process/types.go new file mode 100644 index 0000000000..bd2ffe54bb --- /dev/null +++ b/apps/backend/internal/agentctl/process/types.go @@ -0,0 +1,94 @@ +package process + +import "time" + +// GitStatusUpdate represents a git status update +type GitStatusUpdate struct { + Timestamp time.Time `json:"timestamp"` + Modified []string `json:"modified"` + Added []string `json:"added"` + Deleted []string `json:"deleted"` + Untracked []string `json:"untracked"` + Renamed []string `json:"renamed"` + Ahead int `json:"ahead"` + Behind int `json:"behind"` + Branch string `json:"branch"` + RemoteBranch string `json:"remote_branch,omitempty"` + Files map[string]FileInfo `json:"files,omitempty"` +} + +// FileInfo represents information about a file +type FileInfo struct { + Path string `json:"path"` + Status string `json:"status"` // modified, added, deleted, untracked, renamed + Additions int `json:"additions,omitempty"` + Deletions int `json:"deletions,omitempty"` + OldPath string `json:"old_path,omitempty"` // For renamed files + Diff string `json:"diff,omitempty"` +} + +// FileListUpdate represents a file listing update +type FileListUpdate struct { + Timestamp time.Time `json:"timestamp"` + Files []FileEntry `json:"files"` +} + +// FileEntry represents a file in the workspace +type FileEntry struct { + Path string `json:"path"` + IsDir bool `json:"is_dir"` + Size int64 `json:"size,omitempty"` +} + +// FileTreeNode represents a node in the file tree +type FileTreeNode struct { + Name string `json:"name"` + Path string `json:"path"` + IsDir bool `json:"is_dir"` + Size int64 `json:"size,omitempty"` + Children []*FileTreeNode `json:"children,omitempty"` +} + +// FileTreeRequest represents a request for file tree +type FileTreeRequest struct { + Path string `json:"path"` // Path to get tree for (relative to workspace root) + Depth int `json:"depth"` // Depth to traverse (0 = unlimited, 1 = immediate children only) +} + +// FileTreeResponse represents a response with file tree +type FileTreeResponse struct { + RequestID string `json:"request_id"` + Root *FileTreeNode `json:"root"` + Error string `json:"error,omitempty"` +} + +// FileContentRequest represents a request for file content +type FileContentRequest struct { + Path string `json:"path"` // Path to file (relative to workspace root) +} + +// FileContentResponse represents a response with file content +type FileContentResponse struct { + RequestID string `json:"request_id"` + Path string `json:"path"` + Content string `json:"content"` + Size int64 `json:"size"` + Error string `json:"error,omitempty"` +} + +// FileChangeNotification represents a filesystem change notification +type FileChangeNotification struct { + Timestamp time.Time `json:"timestamp"` + Path string `json:"path"` + Operation string `json:"operation"` // create, write, remove, rename, chmod, refresh +} + +// GitStatusSubscriber is a channel that receives git status updates +type GitStatusSubscriber chan GitStatusUpdate + +// FilesSubscriber is a channel that receives file listing updates +type FilesSubscriber chan FileListUpdate + +// FileChangeSubscriber is a channel that receives file change notifications +type FileChangeSubscriber chan FileChangeNotification + diff --git a/apps/backend/internal/agentctl/process/workspace_tracker.go b/apps/backend/internal/agentctl/process/workspace_tracker.go index fcc49574ac..0845cbccc2 100644 --- a/apps/backend/internal/agentctl/process/workspace_tracker.go +++ b/apps/backend/internal/agentctl/process/workspace_tracker.go @@ -2,60 +2,22 @@ package process import ( "context" + "fmt" + "io" + "io/fs" + "os" "os/exec" + "path/filepath" "strconv" "strings" "sync" "time" + "github.com/fsnotify/fsnotify" "github.com/kandev/kandev/internal/common/logger" "go.uber.org/zap" ) -// GitStatusUpdate represents a git status update -type GitStatusUpdate struct { - Timestamp time.Time `json:"timestamp"` - Modified []string `json:"modified"` - Added []string `json:"added"` - Deleted []string `json:"deleted"` - Untracked []string `json:"untracked"` - Renamed []string `json:"renamed"` - Ahead int `json:"ahead"` - Behind int `json:"behind"` - Branch string `json:"branch"` - RemoteBranch string `json:"remote_branch,omitempty"` - Files map[string]FileInfo `json:"files,omitempty"` -} - -// FileInfo represents information about a file -type FileInfo struct { - Path string `json:"path"` - Status string `json:"status"` // modified, added, deleted, untracked, renamed - Additions int `json:"additions,omitempty"` - Deletions int `json:"deletions,omitempty"` - OldPath string `json:"old_path,omitempty"` // For renamed files - Diff string `json:"diff,omitempty"` -} - -// FileListUpdate represents a file listing update -type FileListUpdate struct { - Timestamp time.Time `json:"timestamp"` - Files []FileEntry `json:"files"` -} - -// FileEntry represents a file in the workspace -type FileEntry struct { - Path string `json:"path"` - IsDir bool `json:"is_dir"` - Size int64 `json:"size,omitempty"` -} - -// GitStatusSubscriber is a channel that receives git status updates -type GitStatusSubscriber chan GitStatusUpdate - -// FilesSubscriber is a channel that receives file listing updates -type FilesSubscriber chan FileListUpdate - // WorkspaceTracker monitors workspace changes and provides real-time updates type WorkspaceTracker struct { workDir string @@ -67,9 +29,16 @@ type WorkspaceTracker struct { mu sync.RWMutex // Subscribers - gitStatusSubscribers map[GitStatusSubscriber]struct{} - filesSubscribers map[FilesSubscriber]struct{} - subMu sync.RWMutex + gitStatusSubscribers map[GitStatusSubscriber]struct{} + filesSubscribers map[FilesSubscriber]struct{} + fileChangeSubscribers map[FileChangeSubscriber]struct{} + subMu sync.RWMutex + + // Filesystem watcher + watcher *fsnotify.Watcher + + // Debounce channel for filesystem change events + fsChangeTrigger chan struct{} // Control stopCh chan struct{} @@ -78,12 +47,21 @@ type WorkspaceTracker struct { // NewWorkspaceTracker creates a new workspace tracker func NewWorkspaceTracker(workDir string, log *logger.Logger) *WorkspaceTracker { + watcher, err := fsnotify.NewWatcher() + if err != nil { + log.Error("failed to create filesystem watcher", zap.Error(err)) + watcher = nil + } + return &WorkspaceTracker{ - workDir: workDir, - logger: log.WithFields(zap.String("component", "workspace-tracker")), - gitStatusSubscribers: make(map[GitStatusSubscriber]struct{}), - filesSubscribers: make(map[FilesSubscriber]struct{}), - stopCh: make(chan struct{}), + workDir: workDir, + logger: log.WithFields(zap.String("component", "workspace-tracker")), + gitStatusSubscribers: make(map[GitStatusSubscriber]struct{}), + filesSubscribers: make(map[FilesSubscriber]struct{}), + fileChangeSubscribers: make(map[FileChangeSubscriber]struct{}), + watcher: watcher, + fsChangeTrigger: make(chan struct{}, 1), // Buffered to avoid blocking + stopCh: make(chan struct{}), } } @@ -91,21 +69,39 @@ func NewWorkspaceTracker(workDir string, log *logger.Logger) *WorkspaceTracker { func (wt *WorkspaceTracker) Start(ctx context.Context) { wt.wg.Add(1) go wt.monitorLoop(ctx) + + // Start filesystem watcher if available + if wt.watcher != nil { + wt.wg.Add(1) + go wt.watchFilesystem(ctx) + + // Add workspace root to watcher + if err := wt.addDirectoryRecursive(wt.workDir); err != nil { + wt.logger.Error("failed to watch workspace directory", zap.Error(err)) + } + } } // Stop stops the workspace tracker func (wt *WorkspaceTracker) Stop() { close(wt.stopCh) + if wt.watcher != nil { + wt.watcher.Close() + } wt.wg.Wait() wt.logger.Info("workspace tracker stopped") } -// monitorLoop periodically checks for workspace changes +// monitorLoop handles debounced filesystem change events +// When files change, we wait for activity to settle, then update everything at once func (wt *WorkspaceTracker) monitorLoop(ctx context.Context) { defer wt.wg.Done() - ticker := time.NewTicker(2 * time.Second) - defer ticker.Stop() + // Debounce duration - wait this long after last file change before updating + const debounceDuration = 300 * time.Millisecond + + var debounceTimer *time.Timer + var pendingUpdate bool // Initial update wt.updateGitStatus(ctx) @@ -114,12 +110,49 @@ func (wt *WorkspaceTracker) monitorLoop(ctx context.Context) { for { select { case <-ctx.Done(): + if debounceTimer != nil { + debounceTimer.Stop() + } return case <-wt.stopCh: + if debounceTimer != nil { + debounceTimer.Stop() + } return - case <-ticker.C: - wt.updateGitStatus(ctx) - wt.updateFiles(ctx) + case <-wt.fsChangeTrigger: + // File change detected - start or reset debounce timer + if debounceTimer == nil { + debounceTimer = time.NewTimer(debounceDuration) + } else { + // Reset the timer if already running + if !debounceTimer.Stop() { + select { + case <-debounceTimer.C: + default: + } + } + debounceTimer.Reset(debounceDuration) + } + pendingUpdate = true + case <-func() <-chan time.Time { + if debounceTimer != nil { + return debounceTimer.C + } + return nil + }(): + // Debounce timer fired - update everything + if pendingUpdate { + wt.updateGitStatus(ctx) + wt.updateFiles(ctx) + // Notify file change subscribers that workspace changed + wt.notifyFileChangeSubscribers(FileChangeNotification{ + Timestamp: time.Now(), + Path: "", + Operation: "refresh", + }) + pendingUpdate = false + } + debounceTimer = nil } } } @@ -539,6 +572,207 @@ func (wt *WorkspaceTracker) GetCurrentFiles() FileListUpdate { return wt.currentFiles } +// SubscribeFileChanges creates a new file change subscriber +func (wt *WorkspaceTracker) SubscribeFileChanges() FileChangeSubscriber { + sub := make(FileChangeSubscriber, 100) + + wt.subMu.Lock() + wt.fileChangeSubscribers[sub] = struct{}{} + wt.subMu.Unlock() + + return sub +} + +// UnsubscribeFileChanges removes a file change subscriber +func (wt *WorkspaceTracker) UnsubscribeFileChanges(sub FileChangeSubscriber) { + wt.subMu.Lock() + delete(wt.fileChangeSubscribers, sub) + wt.subMu.Unlock() + close(sub) +} + +// notifyFileChangeSubscribers notifies all file change subscribers +func (wt *WorkspaceTracker) notifyFileChangeSubscribers(notification FileChangeNotification) { + wt.subMu.RLock() + defer wt.subMu.RUnlock() + + for sub := range wt.fileChangeSubscribers { + select { + case sub <- notification: + default: + // Subscriber is slow, skip + } + } +} + +// watchFilesystem watches for filesystem changes and triggers debounced updates +func (wt *WorkspaceTracker) watchFilesystem(ctx context.Context) { + defer wt.wg.Done() + + for { + select { + case <-ctx.Done(): + return + case <-wt.stopCh: + return + case event, ok := <-wt.watcher.Events: + if !ok { + return + } + + // If a directory was created, watch it + if event.Op&fsnotify.Create == fsnotify.Create { + if info, err := os.Stat(event.Name); err == nil && info.IsDir() { + wt.addDirectoryRecursive(event.Name) + } + } + + // Trigger debounced update (non-blocking) + select { + case wt.fsChangeTrigger <- struct{}{}: + default: + // Channel full, update already pending + } + + case err, ok := <-wt.watcher.Errors: + if !ok { + return + } + wt.logger.Debug("filesystem watcher error", zap.Error(err)) + } + } +} + +// addDirectoryRecursive adds a directory and all its subdirectories to the watcher +func (wt *WorkspaceTracker) addDirectoryRecursive(dir string) error { + return filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + + // Skip hidden directories and common ignore patterns + if d.IsDir() { + name := d.Name() + if name == ".git" || name == "node_modules" || name == ".next" || name == "dist" || name == "build" { + return filepath.SkipDir + } + + // Add directory to watcher + if err := wt.watcher.Add(path); err != nil { + wt.logger.Debug("failed to watch directory", zap.String("path", path), zap.Error(err)) + } + } + + return nil + }) +} + +// GetFileTree returns the file tree for a given path and depth +func (wt *WorkspaceTracker) GetFileTree(reqPath string, depth int) (*FileTreeNode, error) { + // Resolve the full path + fullPath := filepath.Join(wt.workDir, reqPath) + + // Check if path exists + info, err := os.Stat(fullPath) + if err != nil { + return nil, fmt.Errorf("path not found: %w", err) + } + + // Build the tree + node, err := wt.buildFileTreeNode(fullPath, reqPath, info, depth, 0) + if err != nil { + return nil, err + } + + return node, nil +} + +// buildFileTreeNode recursively builds a file tree node +func (wt *WorkspaceTracker) buildFileTreeNode(fullPath, relPath string, info os.FileInfo, maxDepth, currentDepth int) (*FileTreeNode, error) { + node := &FileTreeNode{ + Name: info.Name(), + Path: relPath, + IsDir: info.IsDir(), + Size: info.Size(), + } + + // If it's a file or we've reached max depth, return + if !info.IsDir() || (maxDepth > 0 && currentDepth >= maxDepth) { + return node, nil + } + + // Read directory contents + entries, err := os.ReadDir(fullPath) + if err != nil { + return node, nil // Return node without children on error + } + + // Build children + node.Children = make([]*FileTreeNode, 0, len(entries)) + for _, entry := range entries { + // Skip hidden files and common ignore patterns + name := entry.Name() + if strings.HasPrefix(name, ".") && name != "." && name != ".." { + continue + } + if name == "node_modules" || name == ".next" || name == "dist" || name == "build" { + continue + } + + childFullPath := filepath.Join(fullPath, name) + childRelPath := filepath.Join(relPath, name) + + childInfo, err := entry.Info() + if err != nil { + continue + } + + childNode, err := wt.buildFileTreeNode(childFullPath, childRelPath, childInfo, maxDepth, currentDepth+1) + if err != nil { + continue + } + + node.Children = append(node.Children, childNode) + } + + return node, nil +} + +// GetFileContent returns the content of a file +func (wt *WorkspaceTracker) GetFileContent(reqPath string) (string, int64, error) { + // Resolve the full path + fullPath := filepath.Join(wt.workDir, reqPath) + + // Check if file exists and is a regular file + info, err := os.Stat(fullPath) + if err != nil { + return "", 0, fmt.Errorf("file not found: %w", err) + } + + if info.IsDir() { + return "", 0, fmt.Errorf("path is a directory, not a file") + } + + // Check file size (limit to 10MB) + const maxFileSize = 10 * 1024 * 1024 + if info.Size() > maxFileSize { + return "", info.Size(), fmt.Errorf("file too large (max 10MB)") + } + + // Read file content + file, err := os.Open(fullPath) + if err != nil { + return "", 0, fmt.Errorf("failed to open file: %w", err) + } + defer file.Close() + + content, err := io.ReadAll(file) + if err != nil { + return "", 0, fmt.Errorf("failed to read file: %w", err) + } + + return string(content), info.Size(), nil +} diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 005981b3ab..a485ffb1c1 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -92,7 +92,8 @@ const ( // Event types for workspace/git status const ( - GitStatusUpdated = "git.status.updated" // Git status changed in workspace + GitStatusUpdated = "git.status.updated" // Git status changed in workspace + FileChangeNotified = "file.change.notified" // File changed in workspace ) // BuildACPSubject creates an ACP subject for a specific task @@ -114,3 +115,13 @@ func BuildGitStatusSubject(taskID string) string { func BuildGitStatusWildcardSubject() string { return GitStatusUpdated + ".*" } + +// BuildFileChangeSubject creates a file change subject for a specific task +func BuildFileChangeSubject(taskID string) string { + return FileChangeNotified + "." + taskID +} + +// BuildFileChangeWildcardSubject creates a wildcard subscription for all file change notifications +func BuildFileChangeWildcardSubject() string { + return FileChangeNotified + ".*" +} diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index dfd96c02d9..0d49547e51 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -401,6 +401,19 @@ func (s *SimulatedAgentManagerClient) GetRecoveredInstances() []executor.Recover return nil } +// IsAgentRunningForTask checks if a simulated agent is running for a task +func (s *SimulatedAgentManagerClient) IsAgentRunningForTask(ctx context.Context, taskID string) bool { + s.mu.Lock() + defer s.mu.Unlock() + + for _, inst := range s.instances { + if inst.taskID == taskID && inst.status == v1.AgentStatusRunning { + return true + } + } + return false +} + // ============================================ // ORCHESTRATOR TEST SERVER // ============================================ diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index cbf8ca34c0..1f52902bfa 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -53,6 +53,10 @@ type AgentManagerClient interface { // GetRecoveredInstances returns instances recovered from Docker during startup GetRecoveredInstances() []RecoveredInstanceInfo + + // IsAgentRunningForTask checks if an agent is actually running for a task + // This probes the actual agent (Docker container or standalone process) rather than relying on cached state + IsAgentRunningForTask(ctx context.Context, taskID string) bool } // LaunchAgentRequest contains parameters for launching an agent @@ -474,23 +478,49 @@ func (e *Executor) RespondToPermission(ctx context.Context, taskID, pendingID, o // GetExecution returns the current execution state for a task func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) { + ctx := context.Background() + e.mu.RLock() execution, exists := e.executions[taskID] e.mu.RUnlock() if exists { + // Verify the agent is actually running by probing agentctl + if !e.agentManager.IsAgentRunningForTask(ctx, taskID) { + // Agent stopped - clean up in-memory state + e.mu.Lock() + delete(e.executions, taskID) + e.mu.Unlock() + + // Also update DB if we have a session + if execution.SessionID != "" { + _ = e.repo.UpdateAgentSessionStatus(ctx, execution.SessionID, models.AgentSessionStatusStopped, "agent process stopped") + } + return nil, false + } + // Return a copy to avoid data races execCopy := *execution return &execCopy, true } // Try to load from database - ctx := context.Background() session, err := e.repo.GetActiveAgentSessionByTaskID(ctx, taskID) if err != nil { return nil, false } + // Verify the agent is actually running by probing the lifecycle manager + // This handles the case where backend was restarted and DB has stale "running" sessions + if !e.agentManager.IsAgentRunningForTask(ctx, taskID) { + // Agent is not running - mark the session as stopped in DB and return false + e.logger.Info("stale agent session detected - agent not running", + zap.String("task_id", taskID), + zap.String("session_id", session.ID)) + _ = e.repo.UpdateAgentSessionStatus(ctx, session.ID, models.AgentSessionStatusStopped, "agent not running after backend restart") + return nil, false + } + // Cache it in memory execution = FromAgentSession(session) e.mu.Lock() @@ -508,6 +538,20 @@ func (e *Executor) GetExecutionWithContext(ctx context.Context, taskID string) ( e.mu.RUnlock() if exists { + // Verify the agent is actually running by probing agentctl + if !e.agentManager.IsAgentRunningForTask(ctx, taskID) { + // Agent stopped - clean up in-memory state + e.mu.Lock() + delete(e.executions, taskID) + e.mu.Unlock() + + // Also update DB if we have a session + if execution.SessionID != "" { + _ = e.repo.UpdateAgentSessionStatus(ctx, execution.SessionID, models.AgentSessionStatusStopped, "agent process stopped") + } + return nil, false + } + execCopy := *execution return &execCopy, true } @@ -518,6 +562,15 @@ func (e *Executor) GetExecutionWithContext(ctx context.Context, taskID string) ( return nil, false } + // Verify the agent is actually running by probing the lifecycle manager + if !e.agentManager.IsAgentRunningForTask(ctx, taskID) { + e.logger.Info("stale agent session detected - agent not running", + zap.String("task_id", taskID), + zap.String("session_id", session.ID)) + _ = e.repo.UpdateAgentSessionStatus(ctx, session.ID, models.AgentSessionStatusStopped, "agent not running after backend restart") + return nil, false + } + // Cache it in memory execution = FromAgentSession(session) e.mu.Lock() @@ -743,3 +796,8 @@ func (m *MockAgentManagerClient) RespondToPermissionByTaskID(ctx context.Context func (m *MockAgentManagerClient) GetRecoveredInstances() []RecoveredInstanceInfo { return nil } + +// IsAgentRunningForTask mocks checking if an agent is running (always returns false for mock) +func (m *MockAgentManagerClient) IsAgentRunningForTask(ctx context.Context, taskID string) bool { + return false +} diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index 2d0380042e..877bc8708d 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -68,6 +68,10 @@ func (m *mockAgentManager) GetRecoveredInstances() []executor.RecoveredInstanceI return nil } +func (m *mockAgentManager) IsAgentRunningForTask(ctx context.Context, taskID string) bool { + return false +} + // testTaskRepository is an in-memory task repository for testing type testTaskRepository struct { tasks map[string]*v1.Task diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 7d7361f2e0..e341704f60 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -141,6 +141,11 @@ const ( ActionPermissionRequested = "permission.requested" // Agent requesting permission ActionPermissionRespond = "permission.respond" // User responding to permission request + // Workspace file operations + ActionWorkspaceFileTreeGet = "workspace.tree.get" + ActionWorkspaceFileContentGet = "workspace.file.get" + ActionWorkspaceFileChanges = "workspace.file.changes" // Notification + // User actions ActionUserGet = "user.get" ActionUserSettingsUpdate = "user.settings.update" diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index 6b8de16c64..43cbd0f22f 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -6,13 +6,17 @@ import { ResizableHandle, ResizablePanel, ResizablePanelGroup } from '@kandev/ui import { Tabs, TabsContent, TabsList, TabsTrigger } from '@kandev/ui/tabs'; import { TooltipProvider } from '@kandev/ui/tooltip'; import { Textarea } from '@kandev/ui/textarea'; +import { IconX } from '@tabler/icons-react'; import { getLocalStorage, setLocalStorage } from '@/lib/local-storage'; import { TaskChatPanel } from '@/components/task/task-chat-panel'; import { TaskTopBar } from '@/components/task/task-top-bar'; import type { Task, Comment } from '@/lib/types/http'; +import type { OpenFileTab } from '@/lib/types/backend'; +import { FILE_EXTENSION_COLORS } from '@/lib/types/backend'; import { TaskFilesPanel } from '@/components/task/task-files-panel'; import { TaskChangesPanel } from '@/components/task/task-changes-panel'; import { TaskRightPanel } from '@/components/task/task-right-panel'; +import { FileViewerContent } from '@/components/task/file-viewer-content'; import { getBackendConfig } from '@/lib/config'; import { listRepositories, listRepositoryBranches } from '@/lib/http/client'; import { useRequest } from '@/lib/http/use-request'; @@ -42,6 +46,8 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { const [isLoadingComments, setIsLoadingComments] = useState(false); const [isAgentRunning, setIsAgentRunning] = useState(false); const [isAgentLoading, setIsAgentLoading] = useState(false); + // File viewer tabs + const [openFileTabs, setOpenFileTabs] = useState([]); // Track worktree info separately since it's populated after agent starts const [worktreePath, setWorktreePath] = useState(initialTask?.worktree_path ?? null); const [worktreeBranch, setWorktreeBranch] = useState(initialTask?.worktree_branch ?? null); @@ -158,6 +164,28 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { setLeftTab('changes'); }, []); + const handleOpenFile = useCallback((file: OpenFileTab) => { + setOpenFileTabs((prev) => { + // If file is already open, just switch to it + if (prev.some((t) => t.path === file.path)) { + return prev; + } + // Add new tab (LRU eviction at max 4) + const maxTabs = 4; + const newTabs = prev.length >= maxTabs ? [...prev.slice(1), file] : [...prev, file]; + return newTabs; + }); + setLeftTab(`file:${file.path}`); + }, []); + + const handleCloseFileTab = useCallback((path: string) => { + setOpenFileTabs((prev) => prev.filter((t) => t.path !== path)); + // If closing the active tab, switch to chat + if (leftTab === `file:${path}`) { + setLeftTab('chat'); + } + }, [leftTab]); + const handleStartAgent = useCallback(async () => { if (!task?.id) return; @@ -211,7 +239,7 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { } }, [task?.id]); - const topFilesPanel = ; + const topFilesPanel = ; useEffect(() => { if (!task?.workspace_id || !task.repository_url) return; @@ -252,19 +280,52 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { setLeftTab(value)} - className="flex-1 min-h-0" + className="flex-1 min-h-0 flex flex-col" > - - - Notes - - - All changes - - - Chat - - +
+ + + Notes + + + All changes + + + Chat + + + {openFileTabs.length > 0 && ( + <> +
+ + {openFileTabs.map((tab) => { + const ext = tab.name.split('.').pop()?.toLowerCase() || ''; + const dotColor = FILE_EXTENSION_COLORS[ext] || 'bg-muted-foreground'; + return ( + + + {tab.name} + + + ); + })} + + + )} +