diff --git a/apps/backend/go.mod b/apps/backend/go.mod index cfd9a591c9..455a8b1b13 100644 --- a/apps/backend/go.mod +++ b/apps/backend/go.mod @@ -7,7 +7,6 @@ require ( github.com/coder/acp-go-sdk v0.6.3 github.com/creack/pty v1.1.21 github.com/docker/docker v28.5.2+incompatible - github.com/fsnotify/fsnotify v1.6.0 github.com/gin-gonic/gin v1.9.1 github.com/github/copilot-sdk/go v0.1.18 github.com/google/uuid v1.6.0 @@ -45,6 +44,7 @@ require ( github.com/docker/go-connections v0.6.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect + github.com/fsnotify/fsnotify v1.6.0 // indirect github.com/gabriel-vasile/mimetype v1.4.2 // indirect github.com/gin-contrib/sse v0.1.0 // indirect github.com/go-logr/logr v1.4.3 // indirect diff --git a/apps/backend/internal/agentctl/server/process/git.go b/apps/backend/internal/agentctl/server/process/git.go index effb3d3f42..7f32368891 100644 --- a/apps/backend/internal/agentctl/server/process/git.go +++ b/apps/backend/internal/agentctl/server/process/git.go @@ -168,19 +168,12 @@ func filterGitEnv(env []string) []string { return result } -// triggerFsNotify creates and removes a temporary file to trigger fsnotify and refresh git status. -// This is OS-agnostic and reliably triggers filesystem events that the workspace tracker watches. -// Called after git operations like commit, push, pull, etc. to refresh the UI. -func (g *GitOperator) triggerFsNotify() { - sentinelPath := filepath.Join(g.workDir, ".git-op-complete") - f, err := os.Create(sentinelPath) - if err != nil { - g.logger.Debug("failed to create sentinel file", zap.Error(err)) - return - } - _ = f.Close() - if err := os.Remove(sentinelPath); err != nil { - g.logger.Debug("failed to remove sentinel file", zap.Error(err)) +// triggerRefresh refreshes git status in the workspace tracker immediately. +// Called after git operations like commit, push, pull, etc. to refresh the UI +// without waiting for the next poll cycle. +func (g *GitOperator) triggerRefresh() { + if g.workspaceTracker != nil { + g.workspaceTracker.RefreshGitStatus(context.Background()) } } @@ -669,7 +662,7 @@ func (g *GitOperator) Discard(ctx context.Context, paths []string) (*GitOperatio result.Success = true } - g.triggerFsNotify() + g.triggerRefresh() g.logger.Info("discard completed", zap.Int("total_files", len(paths)), zap.Int("untracked_files", len(untrackedFiles)), @@ -1098,9 +1091,9 @@ func (g *GitOperator) unlock() { g.inProgress = false g.currentOp = "" - // Trigger fsnotify to refresh git status in the workspace tracker. + // Refresh git status in the workspace tracker immediately. // This is called after every git operation completes. - g.triggerFsNotify() + g.triggerRefresh() } // PRCreateResult represents the result of a PR creation operation. diff --git a/apps/backend/internal/agentctl/server/process/workspace_files.go b/apps/backend/internal/agentctl/server/process/workspace_files.go index e2cc43b208..2041b898db 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_files.go +++ b/apps/backend/internal/agentctl/server/process/workspace_files.go @@ -325,7 +325,7 @@ func (wt *WorkspaceTracker) ApplyFileDiff(reqPath string, unifiedDiff string, or // Notify with the original relative path (not the resolved symlink target) relPath := strings.TrimPrefix(safePath, cleanWorkDir+string(os.PathSeparator)) - wt.addPendingChange(relPath, types.FileOpWrite) + wt.notifyFileChange(relPath, types.FileOpWrite) wt.logger.Debug("applied file diff using git apply", zap.String("path", relPath), @@ -389,7 +389,7 @@ func (wt *WorkspaceTracker) CreateFile(reqPath string) error { // Notify with the relative path cleanWorkDir := filepath.Clean(wt.workDir) relPath := strings.TrimPrefix(safePath, cleanWorkDir+string(os.PathSeparator)) - wt.addPendingChange(relPath, types.FileOpCreate) + wt.notifyFileChange(relPath, types.FileOpCreate) return nil } @@ -429,7 +429,7 @@ func (wt *WorkspaceTracker) DeleteFile(reqPath string) error { } relPath := strings.TrimPrefix(safePath, cleanWorkDir+string(os.PathSeparator)) - wt.addPendingChange(relPath, types.FileOpRemove) + wt.notifyFileChange(relPath, types.FileOpRemove) return nil } @@ -475,9 +475,9 @@ func (wt *WorkspaceTracker) RenameFile(oldPath, newPath string) error { cleanWorkDir := wt.resolvedWorkDir() oldRelPath := strings.TrimPrefix(oldSafePath, cleanWorkDir+string(os.PathSeparator)) newRelPath := strings.TrimPrefix(newSafePath, cleanWorkDir+string(os.PathSeparator)) - wt.addPendingChange(oldRelPath, types.FileOpRename) + wt.notifyFileChange(oldRelPath, types.FileOpRename) if newRelPath != oldRelPath { - wt.addPendingChange(newRelPath, types.FileOpRename) + wt.notifyFileChange(newRelPath, types.FileOpRename) } return nil diff --git a/apps/backend/internal/agentctl/server/process/workspace_monitor.go b/apps/backend/internal/agentctl/server/process/workspace_monitor.go index 2fc0f7c2f0..15e45bb2c2 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_monitor.go +++ b/apps/backend/internal/agentctl/server/process/workspace_monitor.go @@ -2,185 +2,96 @@ package process import ( "context" - "io/fs" + "fmt" "os" - "path/filepath" + "os/exec" "time" - "github.com/fsnotify/fsnotify" "github.com/kandev/kandev/internal/agentctl/types" "go.uber.org/zap" ) -// monitorLoop handles debounced filesystem change events -// When files change, we wait for activity to settle, then update everything at once +// DefaultFilePollInterval is the interval for polling file modification times. +// This replaces fsnotify which on macOS (kqueue) opens a file descriptor per file, +// causing "too many open files" errors in large workspaces. +const DefaultFilePollInterval = 2 * time.Second + +// monitorLoop polls for file changes using file mtimes and quick git checks. +// This is more efficient than fsnotify on macOS and avoids file descriptor exhaustion. func (wt *WorkspaceTracker) monitorLoop(ctx context.Context) { defer wt.wg.Done() - // Debounce duration - wait this long after last file change before updating - const debounceDuration = 300 * time.Millisecond - - var debounceTimer *time.Timer - var pendingUpdate bool + ticker := time.NewTicker(DefaultFilePollInterval) + defer ticker.Stop() // Initial update wt.updateGitStatus(ctx) wt.updateFiles(ctx) + // Cache the last known state + lastState := wt.getWorkspaceState(ctx) + + wt.logger.Info("file polling started", zap.Duration("interval", DefaultFilePollInterval)) + for { select { case <-ctx.Done(): - if debounceTimer != nil { - debounceTimer.Stop() - } return case <-wt.stopCh: - if debounceTimer != nil { - debounceTimer.Stop() - } return - 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) + case <-ticker.C: + // Quick state check using mtime + diff-files + currentState := wt.getWorkspaceState(ctx) + if currentState.changed(lastState) { + lastState = currentState + wt.logger.Debug("workspace state changed, updating") + + // Update file list and notify subscribers wt.updateFiles(ctx) - - // Drain pending changes and emit specific notifications - wt.pendingChangesMu.Lock() - changes := wt.pendingChanges - wt.pendingChanges = nil - wt.pendingChangesMu.Unlock() - - wt.emitFileChanges(changes) - pendingUpdate = false + wt.notifyWorkspaceStreamFileChange(types.FileChangeNotification{ + Timestamp: time.Now(), + Operation: types.FileOpRefresh, + }) } - debounceTimer = nil } } } -// 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 - } - - // Ignore CHMOD events - permission changes don't affect file content - // and shouldn't trigger UI refreshes. This prevents loops from filesystem - // scanners, git operations, or other tools that touch file permissions. - if event.Op == fsnotify.Chmod { - continue - } - - // 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() { - if err := wt.addDirectoryRecursive(event.Name); err != nil { - wt.logger.Debug("failed to watch new directory", zap.Error(err)) - } - } - } - - // Record specific change with relative path - relPath, relErr := filepath.Rel(wt.workDir, event.Name) - if relErr != nil { - relPath = event.Name - } - wt.addPendingChange(relPath, fsOpToString(event.Op)) - - case err, ok := <-wt.watcher.Errors: - if !ok { - return - } - wt.logger.Debug("filesystem watcher error", zap.Error(err)) - } - } +// workspaceState holds quick-check state for detecting workspace changes. +type workspaceState struct { + indexMtime time.Time // index mtime - changes on stage/unstage/commit + diffFilesID string // hash of git diff-files output - changes on any file modification } -// fsOpToString converts an fsnotify operation to a FileOp string constant. -func fsOpToString(op fsnotify.Op) string { - switch { - case op&fsnotify.Create != 0: - return types.FileOpCreate - case op&fsnotify.Write != 0: - return types.FileOpWrite - case op&fsnotify.Remove != 0: - return types.FileOpRemove - case op&fsnotify.Rename != 0: - return types.FileOpRename - default: - return types.FileOpRefresh +// getWorkspaceState returns the current workspace state using fast checks. +// Uses index mtime + git diff-files output which includes file content hashes. +func (wt *WorkspaceTracker) getWorkspaceState(ctx context.Context) workspaceState { + var state workspaceState + + // Check index mtime (gitIndexPath is pre-validated at startup) + if wt.gitIndexPath != "" { + if info, err := os.Stat(wt.gitIndexPath); err == nil { + state.indexMtime = info.ModTime() + } } + + // git diff-files shows changed files with their blob hashes + // Output changes whenever file content changes (not just dirty/clean boolean) + cmd := exec.CommandContext(ctx, "git", "diff-files") + cmd.Dir = wt.workDir + out, _ := cmd.Output() + // Use length + first/last bytes as cheap identity check + // (full hash would work too but this is faster for our use case) + if len(out) > 0 { + state.diffFilesID = fmt.Sprintf("%d:%x:%x", len(out), out[0], out[len(out)-1]) + } + + return state } -// 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 - }) -} - -// addPendingChange records a specific file change and triggers a debounced update. -func (wt *WorkspaceTracker) addPendingChange(relPath, operation string) { - wt.pendingChangesMu.Lock() - wt.pendingChanges = append(wt.pendingChanges, types.FileChangeNotification{ - Timestamp: time.Now(), - Path: relPath, - Operation: operation, - }) - wt.pendingChangesMu.Unlock() - - // Trigger debounced update (non-blocking) - select { - case wt.fsChangeTrigger <- struct{}{}: - default: - } +// changed returns true if the workspace state has changed. +func (s workspaceState) changed(other workspaceState) bool { + return s.indexMtime != other.indexMtime || s.diffFilesID != other.diffFilesID } // emitFileChanges sends accumulated file changes to subscribers. @@ -198,3 +109,14 @@ func (wt *WorkspaceTracker) emitFileChanges(changes []types.FileChangeNotificati wt.notifyWorkspaceStreamFileChange(changes[i]) } } + +// notifyFileChange sends a file change notification immediately. +// Used by direct file operations (create, delete, rename, apply diff) to notify +// subscribers without waiting for the next poll cycle. +func (wt *WorkspaceTracker) notifyFileChange(relPath, operation string) { + wt.notifyWorkspaceStreamFileChange(types.FileChangeNotification{ + Timestamp: time.Now(), + Path: relPath, + Operation: operation, + }) +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_tracker.go b/apps/backend/internal/agentctl/server/process/workspace_tracker.go index 3d24a2ed3d..524d22f156 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_tracker.go +++ b/apps/backend/internal/agentctl/server/process/workspace_tracker.go @@ -2,10 +2,13 @@ package process import ( "context" + "os" + "os/exec" + "path/filepath" + "strings" "sync" "time" - "github.com/fsnotify/fsnotify" "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/common/logger" "go.uber.org/zap" @@ -20,10 +23,13 @@ const ( fileStatusModified = "modified" ) -// WorkspaceTracker monitors workspace changes and provides real-time updates +// WorkspaceTracker monitors workspace changes and provides real-time updates. +// It uses git status polling instead of fsnotify to avoid file descriptor exhaustion +// on macOS (where kqueue opens a file descriptor for every watched file). type WorkspaceTracker struct { - workDir string - logger *logger.Logger + workDir string + gitIndexPath string // Cached, validated path to git index file (works with worktrees) + logger *logger.Logger // Current state currentStatus types.GitStatusUpdate @@ -39,16 +45,6 @@ type WorkspaceTracker struct { workspaceStreamSubscribers map[types.WorkspaceStreamSubscriber]struct{} workspaceSubMu sync.RWMutex - // Filesystem watcher - watcher *fsnotify.Watcher - - // Debounce channel for filesystem change events - fsChangeTrigger chan struct{} - - // Pending file changes accumulated between debounce flushes - pendingChanges []types.FileChangeNotification - pendingChangesMu sync.Mutex - // Git polling interval gitPollInterval time.Duration @@ -62,24 +58,48 @@ type WorkspaceTracker struct { // NewWorkspaceTracker creates a new workspace tracker func NewWorkspaceTracker(workDir string, log *logger.Logger) *WorkspaceTracker { resolvedWorkDir := resolveExistingWorkDir(workDir, log.WithFields(zap.String("component", "workspace-tracker"))) - watcher, err := fsnotify.NewWatcher() - if err != nil { - log.Error("failed to create filesystem watcher", zap.Error(err)) - watcher = nil - } + + // Cache validated git index path (works with worktrees where .git is a file) + gitIndexPath := resolveGitIndexPath(resolvedWorkDir) return &WorkspaceTracker{ workDir: resolvedWorkDir, + gitIndexPath: gitIndexPath, logger: log.WithFields(zap.String("component", "workspace-tracker")), workspaceStreamSubscribers: make(map[types.WorkspaceStreamSubscriber]struct{}), - watcher: watcher, - fsChangeTrigger: make(chan struct{}, 1), // Buffered to avoid blocking gitPollInterval: DefaultGitPollInterval, stopCh: make(chan struct{}), } } -// Start begins monitoring the workspace +// resolveGitIndexPath returns the validated path to the git index file. +// Returns empty string if the path cannot be resolved or validated. +// This handles worktrees where .git is a file pointing elsewhere. +func resolveGitIndexPath(workDir string) string { + cmd := exec.Command("git", "rev-parse", "--git-dir") + cmd.Dir = workDir + out, err := cmd.Output() + if err != nil { + return "" + } + gitDir := strings.TrimSpace(string(out)) + if !filepath.IsAbs(gitDir) { + gitDir = filepath.Join(workDir, gitDir) + } + // Clean and construct the index path + gitDir = filepath.Clean(gitDir) + indexPath := filepath.Join(gitDir, "index") + // Validate the index file exists (this proves gitDir is a valid git directory) + info, err := os.Stat(indexPath) + if err != nil || info.IsDir() { + return "" + } + return indexPath +} + +// Start begins monitoring the workspace using polling (no fsnotify). +// File changes are detected via git status polling, which is efficient and +// doesn't consume file descriptors like fsnotify/kqueue does on macOS. func (wt *WorkspaceTracker) Start(ctx context.Context) { wt.mu.Lock() if wt.started { @@ -90,34 +110,19 @@ func (wt *WorkspaceTracker) Start(ctx context.Context) { wt.started = true wt.mu.Unlock() + // Start file change monitoring (uses git status polling) wt.wg.Add(1) go wt.monitorLoop(ctx) - // Start git polling for detecting manual git operations + // Start git polling for detecting manual git operations (commits, resets, etc.) wt.wg.Add(1) go wt.pollGitChanges(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() { wt.stopOnce.Do(func() { close(wt.stopCh) - if wt.watcher != nil { - if err := wt.watcher.Close(); err != nil { - wt.logger.Debug("failed to close watcher", zap.Error(err)) - } - } wt.wg.Wait() wt.logger.Info("workspace tracker stopped") }) diff --git a/apps/backend/internal/agentctl/server/process/workspace_tracker_test.go b/apps/backend/internal/agentctl/server/process/workspace_tracker_test.go index a99d515318..bf245bc2a0 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_tracker_test.go +++ b/apps/backend/internal/agentctl/server/process/workspace_tracker_test.go @@ -10,7 +10,6 @@ import ( "testing" "time" - "github.com/fsnotify/fsnotify" "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/agentctl/types/streams" ) @@ -577,28 +576,6 @@ func TestRenameFile(t *testing.T) { }) } -func TestFsOpToString(t *testing.T) { - tests := []struct { - op fsnotify.Op - want string - }{ - {fsnotify.Create, types.FileOpCreate}, - {fsnotify.Write, types.FileOpWrite}, - {fsnotify.Remove, types.FileOpRemove}, - {fsnotify.Rename, types.FileOpRename}, - {fsnotify.Chmod, types.FileOpRefresh}, - } - - for _, tt := range tests { - t.Run(tt.op.String(), func(t *testing.T) { - got := fsOpToString(tt.op) - if got != tt.want { - t.Errorf("fsOpToString(%v) = %q, want %q", tt.op, got, tt.want) - } - }) - } -} - func TestRewriteDiffPaths(t *testing.T) { tests := []struct { name string