From 159f98efe6f68a5447229b43af80bee81c2acf08 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Wed, 14 Jan 2026 01:24:05 +0000 Subject: [PATCH] feat(web): File browser with syntax-highlighted viewer (#13) * feat(web): add file browser with syntax-highlighted viewer - Add file browser component in right panel to browse workspace files - Open files in left panel tabs alongside Notes/Changes/Chat - Color-coded dots for file types (TS, JS, Go, Python, etc.) - Visual separator between main tabs and file tabs - Folders sorted to top in file tree view - LRU eviction for max 4 open file tabs - Backend handlers for workspace file tree and content via WebSocket * fix(web): refresh file tree on filesystem changes via fsnotify - Subscribe to file change notifications from agentctl's fsnotify watcher - Backend lifecycle manager now connects to file changes stream - Publishes FileChangeNotified events to event bus - WebSocket broadcasts workspace.file.changes to task subscribers - Frontend file browser refreshes tree when changes detected * perf(agentctl): use fsnotify instead of polling for git status - Replace 2-second polling with fsnotify-triggered updates - Add debouncing (300ms) to batch rapid file changes - Git status updates now triggered by actual filesystem changes - More efficient and responsive than periodic polling * refactor(agentctl): consolidate fs events into single debounced trigger - Single fsChangeTrigger channel for all filesystem events - After 300ms debounce: update git status, files, and notify subscribers - Removed per-file notifications in favor of single 'refresh' event - Simpler and more efficient than separate triggers * refactor(agentctl): simplify file API - use HTTP for tree/content - Remove unused filesConn, fileTreeConn, fileContentConn from Client struct - Convert RequestFileTree and RequestFileContent from WebSocket to HTTP GET - Remove StreamFiles and CloseFilesStream (redundant with file changes stream) - Keep only fileChangesConn for the debounced refresh notification HTTP is simpler than WebSocket for request/response patterns. * feat(web): switch to CodeMirror for file viewer - Replace react-syntax-highlighter with @uiw/react-codemirror - Add language extensions for JS/TS, Python, Go, Rust, Java, C++, CSS, HTML, JSON, Markdown, YAML - Use VS Code dark theme for consistent look - Enable code folding and read-only mode - Better performance and features than syntax highlighter * refactor: consolidate duplicated types and extract shared constants - Re-export file types from process package in agentctl/client.go instead of duplicating - Move OpenFileTab type to lib/types/backend.ts (shared by file-browser and page-client) - Extract FILE_EXTENSION_COLORS constant for file type color indicators - Remove console.log statement and unused imports in file-browser.tsx - Add 'refresh' operation to FileChangeNotificationPayload type * refactor(agentctl): move Close() method from acp.go to client.go Close() is a general client cleanup method that closes all streams (ACP, permissions, git status, file changes). It belongs in client.go since acp.go should only contain ACP-specific communication logic. * fix(web): remove unused cn import in file-browser * refactor(agentctl): consolidate duplicated types into process/types.go - Create process/types.go with all workspace-related types - Remove duplicate GitStatusUpdate, FileInfo from workspace_tracker.go - Remove duplicate GitStatusUpdate, FileInfo from api/workspace.go - Re-export all types from agent/agentctl/client.go for convenience Types are now defined once and imported everywhere. * chore: remove package-lock.json, add to .gitignore Project uses pnpm (pnpm-lock.yaml), not npm. * fix(orchestrator): probe agent before returning stale execution state When checking task.execution, the backend now probes the actual agent: - For tracked instances: health check via agentctl - For Docker: verify container is running If the agent is not running but DB says 'running', the session is marked as stopped and false is returned. This fixes the UI showing agents as running after backend restart when they're actually dead. Changes: - Add IsAgentRunningForTask to AgentManagerClient interface - Add IsContainerRunning to docker client - Update GetExecution to verify agent is actually running - Clean up stale sessions in database when agent not found * fix(orchestrator): use agentctl status to verify agent is running Simplified approach - probe agentctl's /api/v1/status endpoint to check if agent process is running. Same check for both Docker and standalone. - Add IsAgentRunning() helper to StatusResponse - Simplify IsAgentRunningForTask to just use agentctl.GetStatus() * fix(orchestrator): also verify agent running for in-memory executions The previous fix only checked agent status when loading from DB. But if an execution is cached in memory and the agent process stops (while agentctl stays running), the stale state was still returned. Now both paths verify the agent is actually running by probing agentctl. If agent stopped, clean up in-memory state and update DB session. * chore(web): remove unused react-syntax-highlighter Switched to CodeMirror for code viewing, no longer needed. * fix(tests): add IsAgentRunningForTask to mock agent managers Add the new interface method to test mocks: - mockAgentManager in scheduler_test.go - SimulatedAgentManagerClient in orchestrator_test.go --------- Co-authored-by: Kandev Agent --- .gitignore | 1 + apps/backend/cmd/kandev/main.go | 30 ++ apps/backend/internal/agent/agentctl/acp.go | 8 - .../backend/internal/agent/agentctl/client.go | 234 +++++++----- apps/backend/internal/agent/docker/client.go | 9 + .../agent/handlers/workspace_file_handlers.go | 110 ++++++ .../internal/agent/lifecycle/manager.go | 90 ++++- apps/backend/internal/agentctl/api/server.go | 5 + .../internal/agentctl/api/workspace.go | 115 ++++-- .../internal/agentctl/process/types.go | 94 +++++ .../agentctl/process/workspace_tracker.go | 350 +++++++++++++++--- apps/backend/internal/events/types.go | 13 +- .../internal/integration/orchestrator_test.go | 13 + .../orchestrator/executor/executor.go | 60 ++- .../orchestrator/scheduler/scheduler_test.go | 4 + apps/backend/pkg/websocket/actions.go | 5 + apps/web/app/task/[id]/page-client.tsx | 93 ++++- apps/web/components/task/file-browser.tsx | 193 ++++++++++ .../components/task/file-viewer-content.tsx | 101 +++++ apps/web/components/task/task-files-panel.tsx | 100 +---- apps/web/lib/types/backend.ts | 55 ++- apps/web/lib/ws/router.ts | 5 + apps/web/lib/ws/workspace-files.ts | 32 ++ apps/web/package.json | 15 + 24 files changed, 1450 insertions(+), 285 deletions(-) create mode 100644 apps/backend/internal/agent/handlers/workspace_file_handlers.go create mode 100644 apps/backend/internal/agentctl/process/types.go create mode 100644 apps/web/components/task/file-browser.tsx create mode 100644 apps/web/components/task/file-viewer-content.tsx create mode 100644 apps/web/lib/ws/workspace-files.ts 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} + + + ); + })} + + + )} +