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 <kandev@example.com>
This commit is contained in:
João Salavisa
@@ -81,3 +81,4 @@ pnpm-lock.yaml
|
||||
# TypeScript
|
||||
*.tsbuildinfo
|
||||
apps/landing/next-env.d.ts
|
||||
package-lock.json
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -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 + ".*"
|
||||
}
|
||||
|
||||
@@ -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
|
||||
// ============================================
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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<OpenFileTab[]>([]);
|
||||
// Track worktree info separately since it's populated after agent starts
|
||||
const [worktreePath, setWorktreePath] = useState<string | null>(initialTask?.worktree_path ?? null);
|
||||
const [worktreeBranch, setWorktreeBranch] = useState<string | null>(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 = <TaskFilesPanel onSelectDiffPath={handleSelectDiffPath} />;
|
||||
const topFilesPanel = <TaskFilesPanel taskId={task?.id ?? null} onSelectDiffPath={handleSelectDiffPath} onOpenFile={handleOpenFile} />;
|
||||
|
||||
useEffect(() => {
|
||||
if (!task?.workspace_id || !task.repository_url) return;
|
||||
@@ -252,19 +280,52 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
|
||||
<Tabs
|
||||
value={leftTab}
|
||||
onValueChange={(value) => setLeftTab(value)}
|
||||
className="flex-1 min-h-0"
|
||||
className="flex-1 min-h-0 flex flex-col"
|
||||
>
|
||||
<TabsList>
|
||||
<TabsTrigger value="notes" className="cursor-pointer">
|
||||
Notes
|
||||
</TabsTrigger>
|
||||
<TabsTrigger value="changes" className="cursor-pointer">
|
||||
All changes
|
||||
</TabsTrigger>
|
||||
<TabsTrigger value="chat" className="cursor-pointer">
|
||||
Chat
|
||||
</TabsTrigger>
|
||||
</TabsList>
|
||||
<div className="flex items-center gap-1">
|
||||
<TabsList>
|
||||
<TabsTrigger value="notes" className="cursor-pointer">
|
||||
Notes
|
||||
</TabsTrigger>
|
||||
<TabsTrigger value="changes" className="cursor-pointer">
|
||||
All changes
|
||||
</TabsTrigger>
|
||||
<TabsTrigger value="chat" className="cursor-pointer">
|
||||
Chat
|
||||
</TabsTrigger>
|
||||
</TabsList>
|
||||
{openFileTabs.length > 0 && (
|
||||
<>
|
||||
<div className="h-4 w-px bg-border mx-1" />
|
||||
<TabsList className="bg-transparent">
|
||||
{openFileTabs.map((tab) => {
|
||||
const ext = tab.name.split('.').pop()?.toLowerCase() || '';
|
||||
const dotColor = FILE_EXTENSION_COLORS[ext] || 'bg-muted-foreground';
|
||||
return (
|
||||
<TabsTrigger
|
||||
key={tab.path}
|
||||
value={`file:${tab.path}`}
|
||||
className="cursor-pointer relative group gap-1.5 data-[state=active]:bg-muted"
|
||||
>
|
||||
<span className={`h-2 w-2 rounded-full ${dotColor}`} />
|
||||
<span className="truncate max-w-[100px]">{tab.name}</span>
|
||||
<button
|
||||
type="button"
|
||||
onClick={(e) => {
|
||||
e.stopPropagation();
|
||||
handleCloseFileTab(tab.path);
|
||||
}}
|
||||
className="ml-0.5 opacity-0 group-hover:opacity-100 transition-opacity hover:text-foreground"
|
||||
>
|
||||
<IconX className="h-3 w-3" />
|
||||
</button>
|
||||
</TabsTrigger>
|
||||
);
|
||||
})}
|
||||
</TabsList>
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
|
||||
<TabsContent value="notes" className="mt-3 flex-1 min-h-0">
|
||||
<Textarea
|
||||
@@ -296,6 +357,12 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
|
||||
</div>
|
||||
)}
|
||||
</TabsContent>
|
||||
|
||||
{openFileTabs.map((tab) => (
|
||||
<TabsContent key={tab.path} value={`file:${tab.path}`} className="mt-3 flex-1 min-h-0">
|
||||
<FileViewerContent path={tab.path} content={tab.content} />
|
||||
</TabsContent>
|
||||
))}
|
||||
</Tabs>
|
||||
</div>
|
||||
</ResizablePanel>
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
'use client';
|
||||
|
||||
import React, { useEffect, useState, useCallback } from 'react';
|
||||
import { IconChevronRight, IconChevronDown, IconFile, IconFolder, IconFolderOpen } from '@tabler/icons-react';
|
||||
import { ScrollArea } from '@kandev/ui/scroll-area';
|
||||
import { getWebSocketClient } from '@/lib/ws/connection';
|
||||
import { requestFileTree, requestFileContent } from '@/lib/ws/workspace-files';
|
||||
import type { FileTreeNode, FileContentResponse, OpenFileTab } from '@/lib/types/backend';
|
||||
|
||||
type FileBrowserProps = {
|
||||
taskId: string;
|
||||
onOpenFile: (file: OpenFileTab) => void;
|
||||
};
|
||||
|
||||
export function FileBrowser({ taskId, onOpenFile }: FileBrowserProps) {
|
||||
const [tree, setTree] = useState<FileTreeNode | null>(null);
|
||||
const [expandedPaths, setExpandedPaths] = useState<Set<string>>(new Set());
|
||||
const [loadingPaths, setLoadingPaths] = useState<Set<string>>(new Set());
|
||||
const [isLoadingTree, setIsLoadingTree] = useState(true);
|
||||
|
||||
// Load initial tree
|
||||
useEffect(() => {
|
||||
const loadTree = async () => {
|
||||
try {
|
||||
const client = getWebSocketClient();
|
||||
if (!client) return;
|
||||
|
||||
const response = await requestFileTree(client, taskId, '', 1);
|
||||
setTree(response.root);
|
||||
} catch (error) {
|
||||
console.error('Failed to load file tree:', error);
|
||||
} finally {
|
||||
setIsLoadingTree(false);
|
||||
}
|
||||
};
|
||||
|
||||
loadTree();
|
||||
}, [taskId]);
|
||||
|
||||
// Subscribe to file changes and refresh tree
|
||||
useEffect(() => {
|
||||
const client = getWebSocketClient();
|
||||
if (!client) return;
|
||||
|
||||
const unsubscribe = client.on('workspace.file.changes', async () => {
|
||||
// Refresh the root tree when any file changes
|
||||
try {
|
||||
const response = await requestFileTree(client, taskId, '', 1);
|
||||
setTree(response.root);
|
||||
} catch (error) {
|
||||
console.error('Failed to refresh file tree:', error);
|
||||
}
|
||||
});
|
||||
|
||||
return unsubscribe;
|
||||
}, [taskId]);
|
||||
|
||||
const toggleExpand = useCallback(async (node: FileTreeNode) => {
|
||||
if (!node.is_dir) return;
|
||||
|
||||
const newExpanded = new Set(expandedPaths);
|
||||
|
||||
if (newExpanded.has(node.path)) {
|
||||
newExpanded.delete(node.path);
|
||||
setExpandedPaths(newExpanded);
|
||||
} else {
|
||||
// If children not loaded yet, fetch them
|
||||
if (!node.children || node.children.length === 0) {
|
||||
setLoadingPaths((prev) => new Set(prev).add(node.path));
|
||||
try {
|
||||
const client = getWebSocketClient();
|
||||
if (!client) return;
|
||||
|
||||
const response = await requestFileTree(client, taskId, node.path, 1);
|
||||
|
||||
// Update tree with new children
|
||||
const updateNode = (n: FileTreeNode): FileTreeNode => {
|
||||
if (n.path === node.path) {
|
||||
return { ...n, children: response.root.children };
|
||||
}
|
||||
if (n.children) {
|
||||
return { ...n, children: n.children.map(updateNode) };
|
||||
}
|
||||
return n;
|
||||
};
|
||||
|
||||
if (tree) {
|
||||
setTree(updateNode(tree));
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Failed to load children:', error);
|
||||
} finally {
|
||||
setLoadingPaths((prev) => {
|
||||
const next = new Set(prev);
|
||||
next.delete(node.path);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
newExpanded.add(node.path);
|
||||
setExpandedPaths(newExpanded);
|
||||
}
|
||||
}, [expandedPaths, taskId, tree]);
|
||||
|
||||
const openFile = useCallback(async (node: FileTreeNode) => {
|
||||
if (node.is_dir) return;
|
||||
|
||||
try {
|
||||
const client = getWebSocketClient();
|
||||
if (!client) return;
|
||||
|
||||
const response: FileContentResponse = await requestFileContent(client, taskId, node.path);
|
||||
|
||||
onOpenFile({
|
||||
path: node.path,
|
||||
name: node.name,
|
||||
content: response.content,
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('Failed to load file content:', error);
|
||||
}
|
||||
}, [taskId, onOpenFile]);
|
||||
|
||||
const renderTreeNode = (node: FileTreeNode, depth: number = 0): React.ReactNode => {
|
||||
const isExpanded = expandedPaths.has(node.path);
|
||||
const isLoading = loadingPaths.has(node.path);
|
||||
|
||||
return (
|
||||
<div key={node.path}>
|
||||
<div
|
||||
className="flex items-center gap-1 py-1 px-2 hover:bg-accent cursor-pointer rounded text-sm"
|
||||
style={{ paddingLeft: `${depth * 12 + 8}px` }}
|
||||
onClick={() => (node.is_dir ? toggleExpand(node) : openFile(node))}
|
||||
>
|
||||
{node.is_dir && (
|
||||
<span className="flex-shrink-0">
|
||||
{isLoading ? (
|
||||
<span className="animate-spin">⏳</span>
|
||||
) : isExpanded ? (
|
||||
<IconChevronDown className="h-4 w-4" />
|
||||
) : (
|
||||
<IconChevronRight className="h-4 w-4" />
|
||||
)}
|
||||
</span>
|
||||
)}
|
||||
{node.is_dir ? (
|
||||
isExpanded ? (
|
||||
<IconFolderOpen className="h-4 w-4 flex-shrink-0 text-blue-500" />
|
||||
) : (
|
||||
<IconFolder className="h-4 w-4 flex-shrink-0 text-blue-500" />
|
||||
)
|
||||
) : (
|
||||
<IconFile className="h-4 w-4 flex-shrink-0 text-muted-foreground" />
|
||||
)}
|
||||
<span className="truncate">{node.name}</span>
|
||||
{!node.is_dir && node.size && (
|
||||
<span className="text-xs text-muted-foreground ml-auto">
|
||||
{(node.size / 1024).toFixed(1)}KB
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
{node.is_dir && isExpanded && node.children && (
|
||||
<div>
|
||||
{[...node.children]
|
||||
.sort((a, b) => {
|
||||
// Folders first, then alphabetically
|
||||
if (a.is_dir && !b.is_dir) return -1;
|
||||
if (!a.is_dir && b.is_dir) return 1;
|
||||
return a.name.localeCompare(b.name);
|
||||
})
|
||||
.map((child) => renderTreeNode(child, depth + 1))}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
};
|
||||
|
||||
return (
|
||||
<div className="flex flex-col h-full">
|
||||
<div className="text-sm font-medium mb-2 px-2">Files</div>
|
||||
<ScrollArea className="flex-1">
|
||||
{isLoadingTree ? (
|
||||
<div className="p-4 text-sm text-muted-foreground">Loading files...</div>
|
||||
) : tree ? (
|
||||
<div className="pb-2">{renderTreeNode(tree)}</div>
|
||||
) : (
|
||||
<div className="p-4 text-sm text-muted-foreground">No files found</div>
|
||||
)}
|
||||
</ScrollArea>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
'use client';
|
||||
|
||||
import CodeMirror from '@uiw/react-codemirror';
|
||||
import { javascript } from '@codemirror/lang-javascript';
|
||||
import { python } from '@codemirror/lang-python';
|
||||
import { go } from '@codemirror/lang-go';
|
||||
import { rust } from '@codemirror/lang-rust';
|
||||
import { java } from '@codemirror/lang-java';
|
||||
import { cpp } from '@codemirror/lang-cpp';
|
||||
import { css } from '@codemirror/lang-css';
|
||||
import { html } from '@codemirror/lang-html';
|
||||
import { json } from '@codemirror/lang-json';
|
||||
import { markdown } from '@codemirror/lang-markdown';
|
||||
import { yaml } from '@codemirror/lang-yaml';
|
||||
import { vscodeDark } from '@uiw/codemirror-theme-vscode';
|
||||
import { EditorView } from '@codemirror/view';
|
||||
import type { Extension } from '@codemirror/state';
|
||||
|
||||
type FileViewerContentProps = {
|
||||
path: string;
|
||||
content: string;
|
||||
};
|
||||
|
||||
const getLanguageExtension = (path: string): Extension | undefined => {
|
||||
const ext = path.split('.').pop()?.toLowerCase();
|
||||
switch (ext) {
|
||||
case 'js':
|
||||
case 'mjs':
|
||||
case 'cjs':
|
||||
return javascript();
|
||||
case 'jsx':
|
||||
return javascript({ jsx: true });
|
||||
case 'ts':
|
||||
case 'mts':
|
||||
case 'cts':
|
||||
return javascript({ typescript: true });
|
||||
case 'tsx':
|
||||
return javascript({ jsx: true, typescript: true });
|
||||
case 'py':
|
||||
return python();
|
||||
case 'go':
|
||||
return go();
|
||||
case 'rs':
|
||||
return rust();
|
||||
case 'java':
|
||||
return java();
|
||||
case 'cpp':
|
||||
case 'cc':
|
||||
case 'cxx':
|
||||
case 'c':
|
||||
case 'h':
|
||||
case 'hpp':
|
||||
return cpp();
|
||||
case 'css':
|
||||
case 'scss':
|
||||
case 'less':
|
||||
return css();
|
||||
case 'html':
|
||||
case 'htm':
|
||||
return html();
|
||||
case 'json':
|
||||
return json();
|
||||
case 'md':
|
||||
case 'mdx':
|
||||
return markdown();
|
||||
case 'yaml':
|
||||
case 'yml':
|
||||
return yaml();
|
||||
default:
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
export function FileViewerContent({ path, content }: FileViewerContentProps) {
|
||||
const langExt = getLanguageExtension(path);
|
||||
const extensions: Extension[] = [
|
||||
EditorView.lineWrapping,
|
||||
EditorView.editable.of(false),
|
||||
];
|
||||
if (langExt) {
|
||||
extensions.push(langExt);
|
||||
}
|
||||
|
||||
return (
|
||||
<CodeMirror
|
||||
value={content}
|
||||
height="100%"
|
||||
theme={vscodeDark}
|
||||
extensions={extensions}
|
||||
readOnly
|
||||
basicSetup={{
|
||||
lineNumbers: true,
|
||||
foldGutter: true,
|
||||
highlightActiveLine: false,
|
||||
highlightSelectionMatches: true,
|
||||
}}
|
||||
className="h-full overflow-auto text-sm"
|
||||
/>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,45 +1,30 @@
|
||||
'use client';
|
||||
|
||||
import type { CSSProperties } from 'react';
|
||||
import { memo, useMemo, useState } from 'react';
|
||||
import {
|
||||
IconArrowBackUp,
|
||||
IconChevronRight,
|
||||
IconExternalLink,
|
||||
IconFile,
|
||||
IconFolder,
|
||||
} from '@tabler/icons-react';
|
||||
import { Badge } from '@kandev/ui/badge';
|
||||
import {
|
||||
Collapsible,
|
||||
CollapsibleContent,
|
||||
CollapsibleTrigger,
|
||||
} from '@kandev/ui/collapsible';
|
||||
import {
|
||||
Sidebar,
|
||||
SidebarContent,
|
||||
SidebarGroup,
|
||||
SidebarGroupContent,
|
||||
SidebarMenu,
|
||||
SidebarMenuButton,
|
||||
SidebarMenuItem,
|
||||
SidebarMenuSub,
|
||||
SidebarProvider,
|
||||
SidebarRail,
|
||||
} from '@kandev/ui/sidebar';
|
||||
import { Tabs, TabsContent, TabsList, TabsTrigger } from '@kandev/ui/tabs';
|
||||
import { Tooltip, TooltipContent, TooltipTrigger } from '@kandev/ui/tooltip';
|
||||
import { LineStat } from '@/components/diff-stat';
|
||||
import { cn } from '@/lib/utils';
|
||||
import { FILE_TREE } from '@/components/task/task-data';
|
||||
import { useAppStore } from '@/components/state-provider';
|
||||
import type { FileInfo } from '@/lib/state/store';
|
||||
import { FileBrowser } from '@/components/task/file-browser';
|
||||
|
||||
type TaskFilesPanelProps = {
|
||||
onSelectDiffPath: (path: string) => void;
|
||||
type OpenFileTab = {
|
||||
path: string;
|
||||
name: string;
|
||||
content: string;
|
||||
};
|
||||
|
||||
type TreeItem = string | TreeItem[];
|
||||
type TaskFilesPanelProps = {
|
||||
taskId: string | null;
|
||||
onSelectDiffPath: (path: string) => void;
|
||||
onOpenFile: (file: OpenFileTab) => void;
|
||||
};
|
||||
|
||||
const badgeClass = (status: string) =>
|
||||
cn(
|
||||
@@ -71,44 +56,7 @@ const splitPath = (path: string) => {
|
||||
};
|
||||
};
|
||||
|
||||
function Tree({ item }: { item: TreeItem }) {
|
||||
const [name, ...items] = Array.isArray(item) ? item : [item];
|
||||
|
||||
if (!items.length) {
|
||||
return (
|
||||
<SidebarMenuButton className="data-[active=true]:bg-transparent">
|
||||
<IconFile className="h-4 w-4" />
|
||||
{name}
|
||||
</SidebarMenuButton>
|
||||
);
|
||||
}
|
||||
|
||||
return (
|
||||
<SidebarMenuItem>
|
||||
<Collapsible
|
||||
className="group/collapsible [&[data-state=open]>button>svg:first-child]:rotate-90"
|
||||
defaultOpen={name === 'components' || name === 'ui'}
|
||||
>
|
||||
<CollapsibleTrigger asChild>
|
||||
<SidebarMenuButton>
|
||||
<IconChevronRight className="h-4 w-4 transition-transform" />
|
||||
<IconFolder className="h-4 w-4" />
|
||||
{name}
|
||||
</SidebarMenuButton>
|
||||
</CollapsibleTrigger>
|
||||
<CollapsibleContent>
|
||||
<SidebarMenuSub>
|
||||
{items.map((subItem, index) => (
|
||||
<Tree key={index} item={subItem} />
|
||||
))}
|
||||
</SidebarMenuSub>
|
||||
</CollapsibleContent>
|
||||
</Collapsible>
|
||||
</SidebarMenuItem>
|
||||
);
|
||||
}
|
||||
|
||||
const TaskFilesPanel = memo(function TaskFilesPanel({ onSelectDiffPath }: TaskFilesPanelProps) {
|
||||
const TaskFilesPanel = memo(function TaskFilesPanel({ taskId, onSelectDiffPath, onOpenFile }: TaskFilesPanelProps) {
|
||||
const [topTab, setTopTab] = useState<'diff' | 'files'>('diff');
|
||||
const gitStatus = useAppStore((state) => state.gitStatus);
|
||||
|
||||
@@ -207,25 +155,13 @@ const TaskFilesPanel = memo(function TaskFilesPanel({ onSelectDiffPath }: TaskFi
|
||||
</TabsContent>
|
||||
<TabsContent value="files" className="mt-3 flex-1 min-h-0">
|
||||
<div className="flex-1 min-h-0 overflow-y-auto rounded-lg bg-background h-full">
|
||||
<SidebarProvider
|
||||
className="h-full w-full"
|
||||
style={{ "--sidebar-width": "100%" } as CSSProperties}
|
||||
>
|
||||
<Sidebar collapsible="none" className="h-full w-full">
|
||||
<SidebarContent>
|
||||
<SidebarGroup>
|
||||
<SidebarGroupContent>
|
||||
<SidebarMenu>
|
||||
{FILE_TREE.map((item, index) => (
|
||||
<Tree key={index} item={item} />
|
||||
))}
|
||||
</SidebarMenu>
|
||||
</SidebarGroupContent>
|
||||
</SidebarGroup>
|
||||
</SidebarContent>
|
||||
<SidebarRail />
|
||||
</Sidebar>
|
||||
</SidebarProvider>
|
||||
{taskId ? (
|
||||
<FileBrowser taskId={taskId} onOpenFile={onOpenFile} />
|
||||
) : (
|
||||
<div className="flex items-center justify-center h-full text-muted-foreground text-sm">
|
||||
No task selected
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
</TabsContent>
|
||||
</Tabs>
|
||||
|
||||
@@ -28,7 +28,8 @@ export type BackendMessageType =
|
||||
| 'agent.profile.deleted'
|
||||
| 'agent.profile.created'
|
||||
| 'agent.profile.updated'
|
||||
| 'user.settings.updated';
|
||||
| 'user.settings.updated'
|
||||
| 'workspace.file.changes';
|
||||
|
||||
export type BackendMessage<T extends BackendMessageType, P> = {
|
||||
id?: string;
|
||||
@@ -231,4 +232,56 @@ export type BackendMessageMap = {
|
||||
'agent.profile.created': BackendMessage<'agent.profile.created', AgentProfileChangedPayload>;
|
||||
'agent.profile.updated': BackendMessage<'agent.profile.updated', AgentProfileChangedPayload>;
|
||||
'user.settings.updated': BackendMessage<'user.settings.updated', UserSettingsUpdatedPayload>;
|
||||
'workspace.file.changes': BackendMessage<'workspace.file.changes', FileChangeNotificationPayload>;
|
||||
};
|
||||
|
||||
// Workspace file types
|
||||
export type FileTreeNode = {
|
||||
name: string;
|
||||
path: string;
|
||||
is_dir: boolean;
|
||||
size?: number;
|
||||
children?: FileTreeNode[];
|
||||
};
|
||||
|
||||
export type FileTreeResponse = {
|
||||
request_id?: string;
|
||||
root: FileTreeNode;
|
||||
error?: string;
|
||||
};
|
||||
|
||||
export type FileContentResponse = {
|
||||
request_id?: string;
|
||||
path: string;
|
||||
content: string;
|
||||
size: number;
|
||||
error?: string;
|
||||
};
|
||||
|
||||
export type FileChangeNotificationPayload = {
|
||||
timestamp: string;
|
||||
path: string;
|
||||
operation: 'create' | 'write' | 'remove' | 'rename' | 'chmod' | 'refresh';
|
||||
};
|
||||
|
||||
// Open file tab for file viewer
|
||||
export type OpenFileTab = {
|
||||
path: string;
|
||||
name: string;
|
||||
content: string;
|
||||
};
|
||||
|
||||
// File extension to color mapping for file type indicators
|
||||
export const FILE_EXTENSION_COLORS: Record<string, string> = {
|
||||
ts: 'bg-blue-500',
|
||||
tsx: 'bg-blue-400',
|
||||
js: 'bg-yellow-500',
|
||||
jsx: 'bg-yellow-400',
|
||||
go: 'bg-cyan-500',
|
||||
py: 'bg-green-500',
|
||||
rs: 'bg-orange-500',
|
||||
json: 'bg-amber-400',
|
||||
css: 'bg-purple-500',
|
||||
html: 'bg-red-500',
|
||||
md: 'bg-gray-400',
|
||||
};
|
||||
|
||||
@@ -427,5 +427,10 @@ export function registerWsHandlers(store: StoreApi<AppState>) {
|
||||
},
|
||||
}));
|
||||
},
|
||||
'workspace.file.changes': (message: BackendMessageMap['workspace.file.changes']) => {
|
||||
// File change notifications will be handled by the file browser component
|
||||
// This handler is here for completeness but the component will subscribe directly
|
||||
console.log('File change notification:', message.payload);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
import type { WebSocketClient } from './client';
|
||||
import type { FileTreeResponse, FileContentResponse } from '@/lib/types/backend';
|
||||
|
||||
/**
|
||||
* Request file tree from the backend
|
||||
*/
|
||||
export async function requestFileTree(
|
||||
client: WebSocketClient,
|
||||
taskId: string,
|
||||
path: string = '',
|
||||
depth: number = 1
|
||||
): Promise<FileTreeResponse> {
|
||||
return client.request<FileTreeResponse>('workspace.tree.get', {
|
||||
task_id: taskId,
|
||||
path,
|
||||
depth,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Request file content from the backend
|
||||
*/
|
||||
export async function requestFileContent(
|
||||
client: WebSocketClient,
|
||||
taskId: string,
|
||||
path: string
|
||||
): Promise<FileContentResponse> {
|
||||
return client.request<FileContentResponse>('workspace.file.get', {
|
||||
task_id: taskId,
|
||||
path,
|
||||
});
|
||||
}
|
||||
@@ -11,6 +11,19 @@
|
||||
},
|
||||
"dependencies": {
|
||||
"@base-ui/react": "^1.0.0",
|
||||
"@codemirror/lang-cpp": "^6.0.3",
|
||||
"@codemirror/lang-css": "^6.3.1",
|
||||
"@codemirror/lang-go": "^6.0.1",
|
||||
"@codemirror/lang-html": "^6.4.11",
|
||||
"@codemirror/lang-java": "^6.0.2",
|
||||
"@codemirror/lang-javascript": "^6.2.4",
|
||||
"@codemirror/lang-json": "^6.0.2",
|
||||
"@codemirror/lang-markdown": "^6.5.0",
|
||||
"@codemirror/lang-python": "^6.2.1",
|
||||
"@codemirror/lang-rust": "^6.0.2",
|
||||
"@codemirror/lang-yaml": "^6.1.2",
|
||||
"@codemirror/state": "^6.5.3",
|
||||
"@codemirror/view": "^6.39.10",
|
||||
"@dnd-kit/core": "^6.3.1",
|
||||
"@dnd-kit/sortable": "^10.0.0",
|
||||
"@dnd-kit/utilities": "^3.2.2",
|
||||
@@ -19,6 +32,8 @@
|
||||
"@kandev/types": "workspace:*",
|
||||
"@kandev/ui": "workspace:*",
|
||||
"@tabler/icons-react": "^3.36.1",
|
||||
"@uiw/codemirror-theme-vscode": "^4.25.4",
|
||||
"@uiw/react-codemirror": "^4.25.4",
|
||||
"class-variance-authority": "^0.7.1",
|
||||
"clsx": "^2.1.1",
|
||||
"cmdk": "^1.1.1",
|
||||
|
||||
Reference in New Issue
Block a user