diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 5aa5c06b7b..15ab3f63c1 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -312,9 +312,28 @@ func main() { } result = append(result, notification) } + + // Also send current git status if available + session, err := taskRepo.GetActiveAgentSessionByTaskID(ctx, taskID) + if err == nil && session != nil && session.Metadata != nil { + if gitStatus, ok := session.Metadata["git_status"]; ok { + // Add task_id to the git status data + gitStatusData, ok := gitStatus.(map[string]interface{}) + if !ok { + gitStatusData = map[string]interface{}{"data": gitStatus} + } + gitStatusData["task_id"] = taskID + + gitStatusNotification, err := ws.NewNotification("git.status", gitStatusData) + if err == nil { + result = append(result, gitStatusNotification) + } + } + } + return result, nil }) - log.Info("Historical logs provider configured for task subscriptions (using comments)") + log.Info("Historical logs provider configured for task subscriptions (using comments and git status)") // NOTE: We no longer create comments for each ACP message chunk. // Instead, the lifecycle manager accumulates message chunks and: @@ -451,6 +470,25 @@ func main() { log.Info("Subscribed to comment.added events for WebSocket broadcasting") } + // Subscribe to git status events and broadcast to WebSocket subscribers + _, err = eventBus.Subscribe(events.BuildGitStatusWildcardSubject(), func(ctx context.Context, event *bus.Event) error { + data := event.Data + taskID, _ := data["task_id"].(string) + if taskID == "" { + return nil + } + + // Broadcast git status update to task subscribers + notification, _ := ws.NewNotification("git.status", data) + gateway.Hub.BroadcastToTask(taskID, notification) + return nil + }) + if err != nil { + log.Error("Failed to subscribe to git status events", zap.Error(err)) + } else { + log.Info("Subscribed to git status events for WebSocket broadcasting") + } + // ============================================ // HTTP SERVER (WebSocket + HTTP endpoints) // ============================================ diff --git a/apps/backend/internal/agent/agentctl/acp.go b/apps/backend/internal/agent/agentctl/acp.go index c8652f67cb..0666009d7c 100644 --- a/apps/backend/internal/agent/agentctl/acp.go +++ b/apps/backend/internal/agent/agentctl/acp.go @@ -240,73 +240,12 @@ func (c *Client) CloseACPStream() { } } -// StreamOutput opens a WebSocket connection for streaming output -func (c *Client) StreamOutput(ctx context.Context, includeHistory bool, handler func(*OutputLine)) error { - wsURL := "ws" + c.baseURL[4:] + "/api/v1/output/stream" - if includeHistory { - wsURL += "?history=true" - } - - conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) - if err != nil { - return fmt.Errorf("failed to connect to output stream: %w", err) - } - - c.mu.Lock() - c.outputConn = conn - c.mu.Unlock() - - c.logger.Info("connected to output stream", zap.String("url", wsURL)) - - // Read messages in a goroutine - go func() { - defer func() { - c.mu.Lock() - c.outputConn = 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("output stream closed normally") - } else { - c.logger.Debug("output stream error", zap.Error(err)) - } - return - } - - var line OutputLine - if err := json.Unmarshal(message, &line); err != nil { - c.logger.Warn("failed to parse output line", zap.Error(err)) - continue - } - - handler(&line) - } - }() - - return nil -} - -// CloseOutputStream closes the output stream connection -func (c *Client) CloseOutputStream() { - c.mu.Lock() - defer c.mu.Unlock() - - if c.outputConn != nil { - c.outputConn.Close() - c.outputConn = nil - } -} - // Close closes all connections func (c *Client) Close() { c.CloseACPStream() - c.CloseOutputStream() c.ClosePermissionStream() + c.CloseGitStatusStream() + c.CloseFilesStream() } // StreamPermissions opens a WebSocket connection for streaming permission requests diff --git a/apps/backend/internal/agent/agentctl/client.go b/apps/backend/internal/agent/agentctl/client.go index e672d2f2a8..a548e73a99 100644 --- a/apps/backend/internal/agent/agentctl/client.go +++ b/apps/backend/internal/agent/agentctl/client.go @@ -21,9 +21,10 @@ type Client struct { logger *logger.Logger // WebSocket connections for streaming - outputConn *websocket.Conn acpConn *websocket.Conn permissionConn *websocket.Conn + gitStatusConn *websocket.Conn + filesConn *websocket.Conn mu sync.RWMutex } @@ -33,13 +34,6 @@ type StatusResponse struct { ProcessInfo map[string]interface{} `json:"process_info"` } -// OutputLine from agentctl -type OutputLine struct { - Timestamp time.Time `json:"timestamp"` - Stream string `json:"stream"` - Content string `json:"content"` -} - // NewClient creates a new agentctl client func NewClient(host string, port int, log *logger.Logger) *Client { return &Client{ @@ -169,3 +163,159 @@ func (c *Client) WaitForReady(ctx context.Context, timeout time.Duration) error 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"` +} + +// StreamGitStatus opens a WebSocket connection for streaming git status updates +func (c *Client) StreamGitStatus(ctx context.Context, handler func(*GitStatusUpdate)) error { + wsURL := "ws" + c.baseURL[4:] + "/api/v1/workspace/git-status/stream" + + conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) + if err != nil { + return fmt.Errorf("failed to connect to git status stream: %w", err) + } + + c.mu.Lock() + c.gitStatusConn = conn + c.mu.Unlock() + + c.logger.Info("connected to git status stream", zap.String("url", wsURL)) + + // Read messages in a goroutine + go func() { + defer func() { + c.mu.Lock() + c.gitStatusConn = 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("git status stream closed normally") + } else { + c.logger.Debug("git status stream error", zap.Error(err)) + } + return + } + + var update GitStatusUpdate + if err := json.Unmarshal(message, &update); err != nil { + c.logger.Warn("failed to parse git status update", zap.Error(err)) + continue + } + + handler(&update) + } + }() + + 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() + defer c.mu.Unlock() + + if c.gitStatusConn != nil { + c.gitStatusConn.Close() + c.gitStatusConn = nil + } +} + +// CloseFilesStream closes the files stream connection +func (c *Client) CloseFilesStream() { + c.mu.Lock() + defer c.mu.Unlock() + + if c.filesConn != nil { + c.filesConn.Close() + c.filesConn = nil + } +} diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 6af424e7de..1281c052c8 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -553,6 +553,9 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta // 4. Set up permission stream to receive permission requests go m.handlePermissionStream(instance) + // 5. Set up git status stream to track workspace changes + go m.handleGitStatusStream(instance) + // 5. Send the task prompt if provided if taskDescription != "" { m.logger.Info("sending ACP prompt", @@ -644,6 +647,35 @@ func (m *Manager) handlePermissionStream(instance *AgentInstance) { } } +// handleGitStatusStream handles streaming git status updates from the agent workspace +func (m *Manager) handleGitStatusStream(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.StreamGitStatus(ctx, func(update *agentctl.GitStatusUpdate) { + m.handleGitStatusUpdate(instance, update) + }) + + if err == nil { + // Connection closed normally + return + } + + if attempt < maxRetries { + time.Sleep(backoff) + backoff *= 2 // Exponential backoff + } + } + + m.logger.Error("failed to connect to git status stream after retries", + zap.String("instance_id", instance.ID), + zap.Int("max_retries", maxRetries)) +} + // reconnectStreams reconnects to agent WebSocket streams after backend restart // This is called for recovered instances that were running before the restart func (m *Manager) reconnectStreams(instance *AgentInstance) { @@ -668,6 +700,7 @@ func (m *Manager) reconnectStreams(instance *AgentInstance) { // Reconnect to WebSocket streams go m.handleUpdatesStream(instance) go m.handlePermissionStream(instance) + go m.handleGitStatusStream(instance) // Mark the instance as READY so it can accept prompts m.mu.Lock() @@ -907,6 +940,66 @@ func (m *Manager) publishSessionInfo(instance *AgentInstance, sessionID string) } } +// handleGitStatusUpdate processes git status updates from the workspace tracker +func (m *Manager) handleGitStatusUpdate(instance *AgentInstance, update *agentctl.GitStatusUpdate) { + // Store git status in instance metadata + m.mu.Lock() + if inst, exists := m.instances[instance.ID]; exists { + if inst.Metadata == nil { + inst.Metadata = make(map[string]interface{}) + } + inst.Metadata["git_status"] = map[string]interface{}{ + "branch": update.Branch, + "remote_branch": update.RemoteBranch, + "modified": update.Modified, + "added": update.Added, + "deleted": update.Deleted, + "untracked": update.Untracked, + "renamed": update.Renamed, + "ahead": update.Ahead, + "behind": update.Behind, + "timestamp": update.Timestamp, + } + } + m.mu.Unlock() + + // Publish git status update to event bus for WebSocket streaming + m.publishGitStatus(instance, update) +} + +// publishGitStatus publishes a git status update to the event bus +func (m *Manager) publishGitStatus(instance *AgentInstance, update *agentctl.GitStatusUpdate) { + if m.eventBus == nil { + return + } + + data := map[string]interface{}{ + "task_id": instance.TaskID, + "agent_id": instance.ID, + "branch": update.Branch, + "remote_branch": update.RemoteBranch, + "modified": update.Modified, + "added": update.Added, + "deleted": update.Deleted, + "untracked": update.Untracked, + "renamed": update.Renamed, + "ahead": update.Ahead, + "behind": update.Behind, + "files": update.Files, + "timestamp": update.Timestamp.Format(time.RFC3339Nano), + } + + event := bus.NewEvent(events.GitStatusUpdated, "agent-manager", data) + subject := events.BuildGitStatusSubject(instance.TaskID) + + if err := m.eventBus.Publish(context.Background(), subject, event); err != nil { + m.logger.Error("failed to publish git status 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) { diff --git a/apps/backend/internal/agentctl/api/output.go b/apps/backend/internal/agentctl/api/output.go deleted file mode 100644 index cee6321da2..0000000000 --- a/apps/backend/internal/agentctl/api/output.go +++ /dev/null @@ -1,102 +0,0 @@ -package api - -import ( - "encoding/json" - "net/http" - "strconv" - - "github.com/gin-gonic/gin" - "github.com/gorilla/websocket" - "github.com/kandev/kandev/internal/agentctl/process" - "go.uber.org/zap" -) - -// OutputResponse is the response for getting output -type OutputResponse struct { - Lines []process.OutputLine `json:"lines"` - Total int `json:"total"` -} - -func (s *Server) handleGetOutput(c *gin.Context) { - // Parse optional 'last' query param - lastStr := c.DefaultQuery("last", "100") - last, err := strconv.Atoi(lastStr) - if err != nil || last < 1 { - last = 100 - } - - lines := s.procMgr.GetOutputBuffer().GetLast(last) - c.JSON(http.StatusOK, OutputResponse{ - Lines: lines, - Total: s.procMgr.GetOutputBuffer().Count(), - }) -} - -// handleOutputStreamWS streams output via WebSocket -func (s *Server) handleOutputStreamWS(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() - - s.logger.Info("Output stream WebSocket connected") - - // Check if client wants history first - sendHistory := c.DefaultQuery("history", "false") == "true" - historyCount, _ := strconv.Atoi(c.DefaultQuery("history_count", "100")) - - if sendHistory { - lines := s.procMgr.GetOutputBuffer().GetLast(historyCount) - for _, line := range lines { - data, err := json.Marshal(line) - if err != nil { - continue - } - if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { - s.logger.Debug("WebSocket write error (history)", zap.Error(err)) - return - } - } - } - - // Subscribe to real-time output - sub := s.procMgr.GetOutputBuffer().Subscribe() - defer s.procMgr.GetOutputBuffer().Unsubscribe(sub) - - // Handle WebSocket close - closeCh := make(chan struct{}) - go func() { - for { - _, _, err := conn.ReadMessage() - if err != nil { - close(closeCh) - return - } - } - }() - - // Stream output to client - for { - select { - case line, ok := <-sub: - if !ok { - return - } - data, err := json.Marshal(line) - if err != nil { - s.logger.Error("failed to marshal output line", zap.Error(err)) - continue - } - if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { - s.logger.Debug("WebSocket write error", zap.Error(err)) - return - } - case <-closeCh: - s.logger.Info("Output stream WebSocket closed by client") - return - } - } -} - diff --git a/apps/backend/internal/agentctl/api/server.go b/apps/backend/internal/agentctl/api/server.go index c1bed90543..a7ba1bcaff 100644 --- a/apps/backend/internal/agentctl/api/server.go +++ b/apps/backend/internal/agentctl/api/server.go @@ -74,9 +74,9 @@ func (s *Server) setupRoutes() { api.GET("/acp/permissions/stream", s.handlePermissionStreamWS) api.POST("/acp/permissions/respond", s.handlePermissionRespond) - // Output streaming - api.GET("/output", s.handleGetOutput) - api.GET("/output/stream", s.handleOutputStreamWS) + // Workspace monitoring (git status, files) + api.GET("/workspace/git-status/stream", s.handleGitStatusStreamWS) + api.GET("/workspace/files/stream", s.handleFilesStreamWS) } } diff --git a/apps/backend/internal/agentctl/api/workspace.go b/apps/backend/internal/agentctl/api/workspace.go new file mode 100644 index 0000000000..71165e6a22 --- /dev/null +++ b/apps/backend/internal/agentctl/api/workspace.go @@ -0,0 +1,126 @@ +package api + +import ( + "encoding/json" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "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) + if err != nil { + s.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + // Subscribe to git status updates + sub := s.procMgr.GetWorkspaceTracker().SubscribeGitStatus() + defer s.procMgr.GetWorkspaceTracker().UnsubscribeGitStatus(sub) + + // Handle WebSocket close + closeCh := make(chan struct{}) + go func() { + for { + _, _, err := conn.ReadMessage() + if err != nil { + close(closeCh) + return + } + } + }() + + // Stream git status updates to client + for { + select { + case update, ok := <-sub: + if !ok { + return + } + data, err := json.Marshal(update) + if err != nil { + s.logger.Error("failed to marshal git status update", zap.Error(err)) + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + case <-closeCh: + return + } + } +} + +// handleFilesStreamWS streams file listing updates via WebSocket +func (s *Server) handleFilesStreamWS(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 updates + sub := s.procMgr.GetWorkspaceTracker().SubscribeFiles() + defer s.procMgr.GetWorkspaceTracker().UnsubscribeFiles(sub) + + // Handle WebSocket close + closeCh := make(chan struct{}) + go func() { + for { + _, _, err := conn.ReadMessage() + if err != nil { + close(closeCh) + return + } + } + }() + + // Stream file updates to client + for { + select { + case update, ok := <-sub: + if !ok { + return + } + data, err := json.Marshal(update) + if err != nil { + s.logger.Error("failed to marshal file update", zap.Error(err)) + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + case <-closeCh: + return + } + } +} diff --git a/apps/backend/internal/agentctl/config/config.go b/apps/backend/internal/agentctl/config/config.go index 98fa58976d..33f4e8a14b 100644 --- a/apps/backend/internal/agentctl/config/config.go +++ b/apps/backend/internal/agentctl/config/config.go @@ -34,9 +34,6 @@ type Config struct { LogLevel string LogFormat string - // Buffer size for agent output (in lines) - OutputBufferSize int - // Health check interval HealthCheckInterval int } @@ -54,7 +51,6 @@ func Load() *Config { AutoApprovePermissions: getEnvBool("AGENTCTL_AUTO_APPROVE_PERMISSIONS", false), LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"), LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"), - OutputBufferSize: getEnvInt("AGENTCTL_OUTPUT_BUFFER_SIZE", 1000), HealthCheckInterval: getEnvInt("AGENTCTL_HEALTH_CHECK_INTERVAL", 5), } diff --git a/apps/backend/internal/agentctl/process/buffer.go b/apps/backend/internal/agentctl/process/buffer.go deleted file mode 100644 index 72af60eeca..0000000000 --- a/apps/backend/internal/agentctl/process/buffer.go +++ /dev/null @@ -1,131 +0,0 @@ -package process - -import ( - "sync" - "time" -) - -// OutputLine represents a line of output from the agent -type OutputLine struct { - Timestamp time.Time `json:"timestamp"` - Stream string `json:"stream"` // "stdout" or "stderr" - Content string `json:"content"` -} - -// Subscriber is a channel that receives output lines -type Subscriber chan OutputLine - -// OutputBuffer is a ring buffer for agent output that supports streaming -type OutputBuffer struct { - lines []OutputLine - size int - head int - count int - mu sync.RWMutex - - // Subscribers for real-time streaming - subscribers map[Subscriber]struct{} - subMu sync.RWMutex -} - -// NewOutputBuffer creates a new output buffer with the given capacity -func NewOutputBuffer(size int) *OutputBuffer { - return &OutputBuffer{ - lines: make([]OutputLine, size), - size: size, - subscribers: make(map[Subscriber]struct{}), - } -} - -// Add adds a new line to the buffer and notifies subscribers -func (b *OutputBuffer) Add(line OutputLine) { - b.mu.Lock() - - // Add to ring buffer - idx := (b.head + b.count) % b.size - if b.count < b.size { - b.count++ - } else { - b.head = (b.head + 1) % b.size - } - b.lines[idx] = line - - b.mu.Unlock() - - // Notify subscribers (non-blocking) - b.subMu.RLock() - for sub := range b.subscribers { - select { - case sub <- line: - default: - // Subscriber is slow, skip - } - } - b.subMu.RUnlock() -} - -// GetAll returns all lines in the buffer (oldest first) -func (b *OutputBuffer) GetAll() []OutputLine { - b.mu.RLock() - defer b.mu.RUnlock() - - result := make([]OutputLine, b.count) - for i := 0; i < b.count; i++ { - idx := (b.head + i) % b.size - result[i] = b.lines[idx] - } - return result -} - -// GetLast returns the last n lines from the buffer -func (b *OutputBuffer) GetLast(n int) []OutputLine { - b.mu.RLock() - defer b.mu.RUnlock() - - if n > b.count { - n = b.count - } - - result := make([]OutputLine, n) - start := b.count - n - for i := 0; i < n; i++ { - idx := (b.head + start + i) % b.size - result[i] = b.lines[idx] - } - return result -} - -// Subscribe creates a new subscriber that receives output lines in real-time -func (b *OutputBuffer) Subscribe() Subscriber { - sub := make(Subscriber, 100) - - b.subMu.Lock() - b.subscribers[sub] = struct{}{} - b.subMu.Unlock() - - return sub -} - -// Unsubscribe removes a subscriber -func (b *OutputBuffer) Unsubscribe(sub Subscriber) { - b.subMu.Lock() - delete(b.subscribers, sub) - b.subMu.Unlock() - close(sub) -} - -// Count returns the number of lines in the buffer -func (b *OutputBuffer) Count() int { - b.mu.RLock() - defer b.mu.RUnlock() - return b.count -} - -// Clear clears the buffer -func (b *OutputBuffer) Clear() { - b.mu.Lock() - defer b.mu.Unlock() - b.head = 0 - b.count = 0 -} - diff --git a/apps/backend/internal/agentctl/process/manager.go b/apps/backend/internal/agentctl/process/manager.go index 6fbf9b1e02..8ac43661ac 100644 --- a/apps/backend/internal/agentctl/process/manager.go +++ b/apps/backend/internal/agentctl/process/manager.go @@ -71,8 +71,8 @@ type Manager struct { exitCode atomic.Int32 exitErr atomic.Value // error - // Output buffer for streaming - outputBuffer *OutputBuffer + // Workspace tracker for git status and file changes + workspaceTracker *WorkspaceTracker // ACP SDK connection acpClient *acpclient.Client @@ -102,7 +102,7 @@ func NewManager(cfg *config.Config, log *logger.Logger) *Manager { m := &Manager{ cfg: cfg, logger: log.WithFields(zap.String("component", "process-manager")), - outputBuffer: NewOutputBuffer(cfg.OutputBufferSize), + workspaceTracker: NewWorkspaceTracker(cfg.WorkDir, log), updatesCh: make(chan acp.SessionNotification, 100), permissionCh: make(chan *PermissionNotification, 10), pendingPermissions: make(map[string]*PendingPermission), @@ -132,9 +132,9 @@ func (m *Manager) ExitError() error { return nil } -// GetOutputBuffer returns the output buffer for streaming -func (m *Manager) GetOutputBuffer() *OutputBuffer { - return m.outputBuffer +// GetWorkspaceTracker returns the workspace tracker for git status and file monitoring +func (m *Manager) GetWorkspaceTracker() *WorkspaceTracker { + return m.workspaceTracker } // Start starts the agent process @@ -218,6 +218,9 @@ func (m *Manager) Start(ctx context.Context) error { go m.readStderr() go m.waitForExit() + // Start workspace tracker with background context (not tied to HTTP request) + m.workspaceTracker.Start(context.Background()) + m.status.Store(StatusRunning) m.logger.Info("agent process started", zap.Int("pid", m.cmd.Process.Pid)) @@ -263,6 +266,11 @@ func (m *Manager) Stop(ctx context.Context) error { m.logger.Info("stopping agent process") m.status.Store(StatusStopping) + // Stop workspace tracker + if m.workspaceTracker != nil { + m.workspaceTracker.Stop() + } + // Close stop channel to signal readers if m.stopCh != nil { close(m.stopCh) @@ -297,18 +305,14 @@ func (m *Manager) Stop(ctx context.Context) error { -// readStderr reads stderr from the agent +// readStderr reads and logs stderr from the agent func (m *Manager) readStderr() { defer m.wg.Done() scanner := bufio.NewScanner(m.stderr) for scanner.Scan() { line := scanner.Text() - m.outputBuffer.Add(OutputLine{ - Timestamp: time.Now(), - Stream: "stderr", - Content: line, - }) + m.logger.Debug("agent stderr", zap.String("line", line)) } if err := scanner.Err(); err != nil { diff --git a/apps/backend/internal/agentctl/process/workspace_tracker.go b/apps/backend/internal/agentctl/process/workspace_tracker.go new file mode 100644 index 0000000000..fcc49574ac --- /dev/null +++ b/apps/backend/internal/agentctl/process/workspace_tracker.go @@ -0,0 +1,544 @@ +package process + +import ( + "context" + "os/exec" + "strconv" + "strings" + "sync" + "time" + + "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 + logger *logger.Logger + + // Current state + currentStatus GitStatusUpdate + currentFiles FileListUpdate + mu sync.RWMutex + + // Subscribers + gitStatusSubscribers map[GitStatusSubscriber]struct{} + filesSubscribers map[FilesSubscriber]struct{} + subMu sync.RWMutex + + // Control + stopCh chan struct{} + wg sync.WaitGroup +} + +// NewWorkspaceTracker creates a new workspace tracker +func NewWorkspaceTracker(workDir string, log *logger.Logger) *WorkspaceTracker { + 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{}), + } +} + +// Start begins monitoring the workspace +func (wt *WorkspaceTracker) Start(ctx context.Context) { + wt.wg.Add(1) + go wt.monitorLoop(ctx) +} + +// Stop stops the workspace tracker +func (wt *WorkspaceTracker) Stop() { + close(wt.stopCh) + wt.wg.Wait() + wt.logger.Info("workspace tracker stopped") +} + +// monitorLoop periodically checks for workspace changes +func (wt *WorkspaceTracker) monitorLoop(ctx context.Context) { + defer wt.wg.Done() + + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + + // Initial update + wt.updateGitStatus(ctx) + wt.updateFiles(ctx) + + for { + select { + case <-ctx.Done(): + return + case <-wt.stopCh: + return + case <-ticker.C: + wt.updateGitStatus(ctx) + wt.updateFiles(ctx) + } + } +} + +// updateGitStatus updates the git status +func (wt *WorkspaceTracker) updateGitStatus(ctx context.Context) { + status, err := wt.getGitStatus(ctx) + if err != nil { + return + } + + wt.mu.Lock() + wt.currentStatus = status + wt.mu.Unlock() + + // Notify subscribers + wt.notifyGitStatusSubscribers(status) +} + +// getGitStatus retrieves the current git status +func (wt *WorkspaceTracker) getGitStatus(ctx context.Context) (GitStatusUpdate, error) { + update := GitStatusUpdate{ + Timestamp: time.Now(), + Modified: []string{}, + Added: []string{}, + Deleted: []string{}, + Untracked: []string{}, + Renamed: []string{}, + Files: make(map[string]FileInfo), + } + + // Get current branch + branchCmd := exec.CommandContext(ctx, "git", "rev-parse", "--abbrev-ref", "HEAD") + branchCmd.Dir = wt.workDir + branchOut, err := branchCmd.Output() + if err != nil { + return update, err + } + update.Branch = strings.TrimSpace(string(branchOut)) + + // Get remote branch + remoteCmd := exec.CommandContext(ctx, "git", "rev-parse", "--abbrev-ref", "@{upstream}") + remoteCmd.Dir = wt.workDir + if remoteOut, err := remoteCmd.Output(); err == nil { + update.RemoteBranch = strings.TrimSpace(string(remoteOut)) + } + + // Get ahead/behind counts + if update.RemoteBranch != "" { + countCmd := exec.CommandContext(ctx, "git", "rev-list", "--left-right", "--count", update.Branch+"..."+update.RemoteBranch) + countCmd.Dir = wt.workDir + if countOut, err := countCmd.Output(); err == nil { + parts := strings.Fields(string(countOut)) + if len(parts) == 2 { + update.Ahead, _ = strconv.Atoi(parts[0]) + update.Behind, _ = strconv.Atoi(parts[1]) + } + } + } + + // Get file status using git status --porcelain + statusCmd := exec.CommandContext(ctx, "git", "status", "--porcelain") + statusCmd.Dir = wt.workDir + statusOut, err := statusCmd.Output() + if err != nil { + return update, err + } + + // Parse porcelain output + lines := strings.Split(string(statusOut), "\n") + for _, line := range lines { + if len(line) < 4 { + continue + } + + statusCode := line[:2] + filePath := strings.TrimSpace(line[3:]) + + fileInfo := FileInfo{ + Path: filePath, + } + + // Parse status code + switch { + case strings.HasPrefix(statusCode, "M ") || strings.HasPrefix(statusCode, " M"): + fileInfo.Status = "modified" + update.Modified = append(update.Modified, filePath) + case strings.HasPrefix(statusCode, "A "): + fileInfo.Status = "added" + update.Added = append(update.Added, filePath) + case strings.HasPrefix(statusCode, "D ") || strings.HasPrefix(statusCode, " D"): + fileInfo.Status = "deleted" + update.Deleted = append(update.Deleted, filePath) + case strings.HasPrefix(statusCode, "??"): + fileInfo.Status = "untracked" + update.Untracked = append(update.Untracked, filePath) + case strings.HasPrefix(statusCode, "R "): + fileInfo.Status = "renamed" + // Renamed files have format "old -> new" + if idx := strings.Index(filePath, " -> "); idx != -1 { + fileInfo.OldPath = filePath[:idx] + fileInfo.Path = filePath[idx+4:] + } + update.Renamed = append(update.Renamed, filePath) + } + + update.Files[filePath] = fileInfo + } + + // Enrich file info with diff data (additions, deletions, and actual diff content) + wt.enrichWithDiffData(ctx, &update) + + return update, nil +} + +// enrichWithDiffData adds diff information (additions, deletions, diff content) to file info +func (wt *WorkspaceTracker) enrichWithDiffData(ctx context.Context, update *GitStatusUpdate) { + // Determine the base ref to compare against + // Use remote branch if available, otherwise use HEAD + baseRef := "HEAD" + if update.RemoteBranch != "" { + baseRef = update.RemoteBranch + } + + // Get numstat for additions/deletions count (compare against base branch) + numstatCmd := exec.CommandContext(ctx, "git", "diff", "--numstat", baseRef) + numstatCmd.Dir = wt.workDir + numstatOut, err := numstatCmd.Output() + if err != nil { + wt.logger.Debug("failed to get numstat", zap.Error(err)) + return + } + + // Parse numstat output and update file info + lines := strings.Split(string(numstatOut), "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + + parts := strings.Fields(line) + if len(parts) < 3 { + continue + } + + additions, _ := strconv.Atoi(parts[0]) + deletions, _ := strconv.Atoi(parts[1]) + filePath := parts[2] + + // Update existing file info if it exists + if fileInfo, exists := update.Files[filePath]; exists { + fileInfo.Additions = additions + fileInfo.Deletions = deletions + + // Get the actual diff content for this file (compare against base branch) + diffCmd := exec.CommandContext(ctx, "git", "diff", baseRef, "--", filePath) + diffCmd.Dir = wt.workDir + if diffOut, err := diffCmd.Output(); err == nil { + fileInfo.Diff = string(diffOut) + } + + update.Files[filePath] = fileInfo + } else { + // File might be committed but not in status (because it's committed) + // Add it to the Files map + fileInfo := FileInfo{ + Path: filePath, + Status: "modified", // Assume modified if it has a diff + Additions: additions, + Deletions: deletions, + } + + // Get the actual diff content for this file + diffCmd := exec.CommandContext(ctx, "git", "diff", baseRef, "--", filePath) + diffCmd.Dir = wt.workDir + if diffOut, err := diffCmd.Output(); err == nil { + fileInfo.Diff = string(diffOut) + } + + update.Files[filePath] = fileInfo + update.Modified = append(update.Modified, filePath) + } + } + + // Also check for staged changes + stagedCmd := exec.CommandContext(ctx, "git", "diff", "--cached", "--numstat") + stagedCmd.Dir = wt.workDir + stagedOut, err := stagedCmd.Output() + if err != nil { + return + } + + lines = strings.Split(string(stagedOut), "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + + parts := strings.Fields(line) + if len(parts) < 3 { + continue + } + + additions, _ := strconv.Atoi(parts[0]) + deletions, _ := strconv.Atoi(parts[1]) + filePath := parts[2] + + // Update existing file info if it exists + if fileInfo, exists := update.Files[filePath]; exists { + // Add staged changes to the counts + fileInfo.Additions += additions + fileInfo.Deletions += deletions + + // Get the staged diff content + diffCmd := exec.CommandContext(ctx, "git", "diff", "--cached", "--", filePath) + diffCmd.Dir = wt.workDir + if diffOut, err := diffCmd.Output(); err == nil { + // Append staged diff to existing diff + if fileInfo.Diff != "" { + fileInfo.Diff += "\n\n--- Staged changes ---\n" + string(diffOut) + } else { + fileInfo.Diff = string(diffOut) + } + } + + update.Files[filePath] = fileInfo + } + } + + // For untracked files, show the entire file content as "added" + for filePath, fileInfo := range update.Files { + if fileInfo.Status == "untracked" { + // Show file content as diff (all additions) + catCmd := exec.CommandContext(ctx, "cat", filePath) + catCmd.Dir = wt.workDir + if catOut, err := catCmd.Output(); err == nil { + content := string(catOut) + lines := strings.Split(content, "\n") + fileInfo.Additions = len(lines) + fileInfo.Deletions = 0 + + // Format as a diff + var diffBuilder strings.Builder + diffBuilder.WriteString("--- /dev/null\n") + diffBuilder.WriteString("+++ b/" + filePath + "\n") + diffBuilder.WriteString("@@ -0,0 +1," + strconv.Itoa(len(lines)) + " @@\n") + for _, line := range lines { + diffBuilder.WriteString("+" + line + "\n") + } + fileInfo.Diff = diffBuilder.String() + + update.Files[filePath] = fileInfo + } + } + } +} + +// updateFiles updates the file listing +func (wt *WorkspaceTracker) updateFiles(ctx context.Context) { + files, err := wt.getFileList(ctx) + if err != nil { + wt.logger.Debug("failed to get file list", zap.Error(err)) + return + } + + wt.mu.Lock() + wt.currentFiles = files + wt.mu.Unlock() + + // Notify subscribers + wt.notifyFilesSubscribers(files) +} + +// getFileList retrieves the list of files in the workspace +func (wt *WorkspaceTracker) getFileList(ctx context.Context) (FileListUpdate, error) { + update := FileListUpdate{ + Timestamp: time.Now(), + Files: []FileEntry{}, + } + + // Use git ls-files to get tracked files + cmd := exec.CommandContext(ctx, "git", "ls-files") + cmd.Dir = wt.workDir + out, err := cmd.Output() + if err != nil { + return update, err + } + + lines := strings.Split(string(out), "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + update.Files = append(update.Files, FileEntry{ + Path: line, + IsDir: false, + }) + } + + return update, nil +} + + + +// Subscriber management methods + +// SubscribeGitStatus creates a new git status subscriber +func (wt *WorkspaceTracker) SubscribeGitStatus() GitStatusSubscriber { + sub := make(GitStatusSubscriber, 10) + + wt.subMu.Lock() + wt.gitStatusSubscribers[sub] = struct{}{} + wt.subMu.Unlock() + + // Send current status immediately + wt.mu.RLock() + current := wt.currentStatus + wt.mu.RUnlock() + + if current.Timestamp.IsZero() { + current.Timestamp = time.Now() + } + + select { + case sub <- current: + default: + } + + return sub +} + +// UnsubscribeGitStatus removes a git status subscriber +func (wt *WorkspaceTracker) UnsubscribeGitStatus(sub GitStatusSubscriber) { + wt.subMu.Lock() + delete(wt.gitStatusSubscribers, sub) + wt.subMu.Unlock() + close(sub) +} + +// SubscribeFiles creates a new files subscriber +func (wt *WorkspaceTracker) SubscribeFiles() FilesSubscriber { + sub := make(FilesSubscriber, 10) + + wt.subMu.Lock() + wt.filesSubscribers[sub] = struct{}{} + wt.subMu.Unlock() + + // Send current files immediately + wt.mu.RLock() + current := wt.currentFiles + wt.mu.RUnlock() + + if current.Timestamp.IsZero() { + current.Timestamp = time.Now() + } + + select { + case sub <- current: + default: + } + + return sub +} + +// UnsubscribeFiles removes a files subscriber +func (wt *WorkspaceTracker) UnsubscribeFiles(sub FilesSubscriber) { + wt.subMu.Lock() + delete(wt.filesSubscribers, sub) + wt.subMu.Unlock() + close(sub) +} + +// Notification methods + +// notifyGitStatusSubscribers notifies all git status subscribers +func (wt *WorkspaceTracker) notifyGitStatusSubscribers(update GitStatusUpdate) { + wt.subMu.RLock() + defer wt.subMu.RUnlock() + + for sub := range wt.gitStatusSubscribers { + select { + case sub <- update: + default: + // Subscriber is slow, skip + } + } +} + +// notifyFilesSubscribers notifies all files subscribers +func (wt *WorkspaceTracker) notifyFilesSubscribers(update FileListUpdate) { + wt.subMu.RLock() + defer wt.subMu.RUnlock() + + for sub := range wt.filesSubscribers { + select { + case sub <- update: + default: + // Subscriber is slow, skip + } + } +} + +// GetCurrentGitStatus returns the current git status +func (wt *WorkspaceTracker) GetCurrentGitStatus() GitStatusUpdate { + wt.mu.RLock() + defer wt.mu.RUnlock() + return wt.currentStatus +} + +// GetCurrentFiles returns the current file list +func (wt *WorkspaceTracker) GetCurrentFiles() FileListUpdate { + wt.mu.RLock() + defer wt.mu.RUnlock() + return wt.currentFiles +} + + + + diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 4c2653c5ea..66494d7fbe 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -71,6 +71,11 @@ const ( ToolCallComplete = "tool_call.complete" // Agent finished a tool call ) +// Event types for workspace/git status +const ( + GitStatusUpdated = "git.status.updated" // Git status changed in workspace +) + // BuildACPSubject creates an ACP subject for a specific task func BuildACPSubject(taskID string) string { return ACPMessage + "." + taskID @@ -80,3 +85,13 @@ func BuildACPSubject(taskID string) string { func BuildACPWildcardSubject() string { return ACPMessage + ".*" } + +// BuildGitStatusSubject creates a git status subject for a specific task +func BuildGitStatusSubject(taskID string) string { + return GitStatusUpdated + "." + taskID +} + +// BuildGitStatusWildcardSubject creates a wildcard subscription for all git status updates +func BuildGitStatusWildcardSubject() string { + return GitStatusUpdated + ".*" +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 1241162dd1..8caae60b83 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -60,6 +60,7 @@ type Service struct { eventBus bus.EventBus db *database.DB taskRepo scheduler.TaskRepository + repo repository.Repository // Full repository for agent sessions agentManager executor.AgentManagerClient // Components @@ -127,6 +128,7 @@ func NewService( eventBus: eventBus, db: db, taskRepo: taskRepo, + repo: repo, agentManager: agentManager, queue: taskQueue, executor: exec, @@ -143,6 +145,7 @@ func NewService( OnACPMessage: s.handleACPMessage, OnPromptComplete: s.handlePromptComplete, OnToolCallStarted: s.handleToolCallStarted, + OnGitStatusUpdated: s.handleGitStatusUpdated, } s.watcher = watcher.NewWatcher(eventBus, handlers, log) @@ -439,14 +442,42 @@ func (s *Service) handleTaskStateChanged(ctx context.Context, data watcher.TaskE } // handleAgentReady handles agent ready events (prompt completed, ready for follow-up) +// This is the definitive signal that the agent has finished processing a prompt. +// AgentReady fires ONCE after each prompt completes (unlike PromptComplete which +// fires for intermediate messages before tool calls). func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventData) { s.logger.Info("handling agent ready", zap.String("task_id", data.TaskID), zap.String("agent_instance_id", data.AgentInstanceID)) - // Agent completed initial prompt but is still running - // Task stays in IN_PROGRESS state - user can send follow-up prompts - // or explicitly complete/stop the task + // Get current task state to determine if we should transition to REVIEW + task, err := s.taskRepo.GetTask(ctx, data.TaskID) + if err != nil { + s.logger.Error("failed to get task for state check", + zap.String("task_id", data.TaskID), + zap.Error(err)) + return + } + + // Only transition to REVIEW if the task is currently IN_PROGRESS + // This implements the forward movement: IN_PROGRESS → (agent finishes) → REVIEW + // This works for both: + // 1. Initial task execution: Task starts IN_PROGRESS, agent finishes → REVIEW + // 2. Follow-up prompts: User message sets IN_PROGRESS, agent finishes → REVIEW + if task.State == v1.TaskStateInProgress { + if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil { + s.logger.Error("failed to update task state to REVIEW", + zap.String("task_id", data.TaskID), + zap.Error(err)) + } else { + s.logger.Info("task moved to REVIEW state", + zap.String("task_id", data.TaskID)) + } + } else { + s.logger.Debug("task not in IN_PROGRESS state, skipping REVIEW transition", + zap.String("task_id", data.TaskID), + zap.String("current_state", string(task.State))) + } } // handleAgentCompleted handles agent completion events @@ -459,13 +490,14 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv s.scheduler.HandleTaskCompleted(data.TaskID, true) s.scheduler.RemoveTask(data.TaskID) - // Update task state to COMPLETED - if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateCompleted); err != nil { - s.logger.Error("failed to update task state to COMPLETED", + // Move task to REVIEW state for user review + // The user can then send a follow-up (moves back to IN_PROGRESS) or mark as COMPLETED + if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil { + s.logger.Error("failed to update task state to REVIEW", zap.String("task_id", data.TaskID), zap.Error(err)) } else { - s.logger.Info("task marked as COMPLETED", + s.logger.Info("task moved to REVIEW state after agent completion", zap.String("task_id", data.TaskID)) } } @@ -483,11 +515,15 @@ func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEvent s.logger.Error("task failed and retry limit exceeded", zap.String("task_id", data.TaskID)) - // Update task state to FAILED - if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateFailed); err != nil { - s.logger.Error("failed to update task state to FAILED", + // Move task to REVIEW state even on failure - user can decide to retry or close + // This maintains the review cycle: user reviews the failure and decides next steps + if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil { + s.logger.Error("failed to update task state to REVIEW after failure", zap.String("task_id", data.TaskID), zap.Error(err)) + } else { + s.logger.Info("task moved to REVIEW state after failure (for user review)", + zap.String("task_id", data.TaskID)) } } } @@ -538,7 +574,58 @@ func (s *Service) handleInputRequired(ctx context.Context, taskID string, msg *p } } +// handleGitStatusUpdated handles git status updates and persists them to agent session metadata +func (s *Service) handleGitStatusUpdated(ctx context.Context, data watcher.GitStatusData) { + s.logger.Debug("handling git status update", + zap.String("task_id", data.TaskID), + zap.String("branch", data.Branch)) + + // Get the active agent session for this task + session, err := s.repo.GetActiveAgentSessionByTaskID(ctx, data.TaskID) + if err != nil { + s.logger.Debug("no active agent session for git status update", + zap.String("task_id", data.TaskID), + zap.Error(err)) + return + } + + // Update session metadata with git status + if session.Metadata == nil { + session.Metadata = make(map[string]interface{}) + } + session.Metadata["git_status"] = map[string]interface{}{ + "branch": data.Branch, + "remote_branch": data.RemoteBranch, + "modified": data.Modified, + "added": data.Added, + "deleted": data.Deleted, + "untracked": data.Untracked, + "renamed": data.Renamed, + "ahead": data.Ahead, + "behind": data.Behind, + "files": data.Files, + "timestamp": data.Timestamp, + } + + // Persist to database asynchronously + go func() { + if err := s.repo.UpdateAgentSession(context.Background(), session); err != nil { + s.logger.Error("failed to update agent session with git status", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID), + zap.Error(err)) + } else { + s.logger.Debug("persisted git status to agent session", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID)) + } + }() +} + // handlePromptComplete handles prompt complete events and saves agent response as comment +// Note: This is called for EVERY agent message (including intermediate ones before tool calls), +// so we should NOT transition task state here. State transitions happen in the comment handler +// after the synchronous PromptTask call returns. func (s *Service) handlePromptComplete(ctx context.Context, data watcher.PromptCompleteData) { s.logger.Info("handling prompt complete", zap.String("task_id", data.TaskID), diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index 41462b9504..bb200de106 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -50,6 +50,23 @@ type ToolCallData struct { Args map[string]interface{} `json:"args,omitempty"` // Tool call arguments (kind, path, locations, raw_input) } +// GitStatusData contains data from git status events +type GitStatusData struct { + TaskID string `json:"task_id"` + AgentID string `json:"agent_id"` + Branch string `json:"branch"` + RemoteBranch string `json:"remote_branch"` + 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"` + Files map[string]interface{} `json:"files"` + Timestamp string `json:"timestamp"` +} + // EventHandlers contains callbacks for different event types type EventHandlers struct { // Task events @@ -73,6 +90,9 @@ type EventHandlers struct { // Tool call events OnToolCallStarted func(ctx context.Context, data ToolCallData) + + // Git status events + OnGitStatusUpdated func(ctx context.Context, data GitStatusData) } // Watcher subscribes to events and dispatches to handlers @@ -139,6 +159,12 @@ func (w *Watcher) Start(ctx context.Context) error { return err } + // Subscribe to git status events + if err := w.subscribeToGitStatusEvents(); err != nil { + w.unsubscribeAll() + return err + } + w.running = true w.logger.Info("Event watcher started", zap.Int("subscriptions", len(w.subscriptions))) return nil @@ -422,6 +448,49 @@ func (w *Watcher) createToolCallHandler() bus.EventHandler { } } +// subscribeToGitStatusEvents subscribes to git status events +func (w *Watcher) subscribeToGitStatusEvents() error { + if w.handlers.OnGitStatusUpdated == nil { + return nil + } + + // Use wildcard to subscribe to all git status events (git.status.updated.*) + subject := events.BuildGitStatusWildcardSubject() + + // Use regular subscription (each instance needs all messages) + sub, err := w.eventBus.Subscribe(subject, w.createGitStatusHandler()) + if err != nil { + w.logger.Error("Failed to subscribe to git status events", + zap.String("subject", subject), + zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) + return nil +} + +// createGitStatusHandler creates a bus.EventHandler for git status events +func (w *Watcher) createGitStatusHandler() bus.EventHandler { + return func(ctx context.Context, event *bus.Event) error { + var data GitStatusData + if err := w.parseEventData(event.Data, &data); err != nil { + w.logger.Error("Failed to parse git status event data", + zap.String("event_type", event.Type), + zap.String("event_id", event.ID), + zap.Error(err)) + return nil // Don't return error to continue processing other events + } + + w.logger.Debug("Handling git status event", + zap.String("task_id", data.TaskID), + zap.String("branch", data.Branch), + zap.Int("modified", len(data.Modified))) + + w.handlers.OnGitStatusUpdated(ctx, data) + return nil + } +} + // parseEventData converts event data map to a typed struct func (w *Watcher) parseEventData(data map[string]interface{}, target interface{}) error { // Marshal to JSON and unmarshal to target type diff --git a/apps/backend/internal/task/handlers/comment_handlers.go b/apps/backend/internal/task/handlers/comment_handlers.go index a2b59e3e15..90098b6ba1 100644 --- a/apps/backend/internal/task/handlers/comment_handlers.go +++ b/apps/backend/internal/task/handlers/comment_handlers.go @@ -8,6 +8,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/task/service" + v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" ) @@ -69,6 +70,29 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "content is required", nil) } + // Get the current task state to determine if we need to transition + task, err := h.svc.GetTask(ctx, req.TaskID) + if err != nil { + h.logger.Error("failed to get task", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get task", nil) + } + + // If task is in REVIEW state, transition back to IN_PROGRESS + // This implements the backward movement in the review cycle: + // REVIEW → (user sends follow-up) → IN_PROGRESS + // The transition back to REVIEW happens in handleAgentReady when the agent finishes + if task.State == v1.TaskStateReview { + if _, err := h.svc.UpdateTaskState(ctx, req.TaskID, v1.TaskStateInProgress); err != nil { + h.logger.Error("failed to transition task from REVIEW to IN_PROGRESS", + zap.String("task_id", req.TaskID), + zap.Error(err)) + // Continue anyway - the comment should still be created + } else { + h.logger.Info("task transitioned from REVIEW to IN_PROGRESS", + zap.String("task_id", req.TaskID)) + } + } + comment, err := h.svc.CreateComment(ctx, &service.CreateCommentRequest{ TaskID: req.TaskID, Content: req.Content, @@ -82,6 +106,7 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w // Auto-forward comment as prompt to running agent if orchestrator is available // The agent's response will be saved as a comment via the prompt_complete event + // Note: State transition to REVIEW happens via handleAgentReady when the agent finishes if h.orchestrator != nil { _, err := h.orchestrator.PromptTask(ctx, req.TaskID, req.Content) if err != nil { diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index c905cedfef..672aa0b096 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -37,6 +37,7 @@ type Repository interface { // Column operations CreateColumn(ctx context.Context, column *models.Column) error GetColumn(ctx context.Context, id string) (*models.Column, error) + GetColumnByState(ctx context.Context, boardID string, state v1.TaskState) (*models.Column, error) UpdateColumn(ctx context.Context, column *models.Column) error DeleteColumn(ctx context.Context, id string) error ListColumns(ctx context.Context, boardID string) ([]*models.Column, error) diff --git a/apps/backend/internal/task/repository/memory.go b/apps/backend/internal/task/repository/memory.go index d5a067ae66..9ab2aa9c3d 100644 --- a/apps/backend/internal/task/repository/memory.go +++ b/apps/backend/internal/task/repository/memory.go @@ -374,6 +374,19 @@ func (r *MemoryRepository) GetColumn(ctx context.Context, id string) (*models.Co return column, nil } +// GetColumnByState retrieves a column by board ID and state +func (r *MemoryRepository) GetColumnByState(ctx context.Context, boardID string, state v1.TaskState) (*models.Column, error) { + r.mu.RLock() + defer r.mu.RUnlock() + + for _, column := range r.columns { + if column.BoardID == boardID && column.State == state { + return column, nil + } + } + return nil, fmt.Errorf("column not found for board %s with state %s", boardID, state) +} + // UpdateColumn updates an existing column func (r *MemoryRepository) UpdateColumn(ctx context.Context, column *models.Column) error { r.mu.Lock() diff --git a/apps/backend/internal/task/repository/sqlite.go b/apps/backend/internal/task/repository/sqlite.go index e21e16e1a7..218b0644bd 100644 --- a/apps/backend/internal/task/repository/sqlite.go +++ b/apps/backend/internal/task/repository/sqlite.go @@ -757,6 +757,13 @@ func (r *SQLiteRepository) AddTaskToBoard(ctx context.Context, taskID, boardID, VALUES (?, ?, ?, ?) ON CONFLICT(task_id, board_id) DO UPDATE SET column_id = excluded.column_id, position = excluded.position `, taskID, boardID, columnID, position) + if err != nil { + return err + } + // Also update the column_id in the tasks table so GetTask returns the correct column + _, err = r.db.ExecContext(ctx, ` + UPDATE tasks SET column_id = ?, position = ?, updated_at = ? WHERE id = ? + `, columnID, position, time.Now().UTC(), taskID) return err } @@ -776,11 +783,18 @@ func (r *SQLiteRepository) addTaskToBoardTx(ctx context.Context, tx *sql.Tx, tas return err } - _, err := tx.ExecContext(ctx, ` + if _, err := tx.ExecContext(ctx, ` INSERT INTO task_columns (task_id, board_id, column_id, position) VALUES (?, ?, ?, ?) ON CONFLICT(task_id, board_id) DO UPDATE SET column_id = excluded.column_id, position = excluded.position - `, taskID, boardID, columnID, position) + `, taskID, boardID, columnID, position); err != nil { + return err + } + + // Also update the column_id in the tasks table so GetTask returns the correct column + _, err := tx.ExecContext(ctx, ` + UPDATE tasks SET column_id = ?, position = ?, updated_at = ? WHERE id = ? + `, columnID, position, time.Now().UTC(), taskID) return err } @@ -914,6 +928,21 @@ func (r *SQLiteRepository) GetColumn(ctx context.Context, id string) (*models.Co return column, err } +// GetColumnByState retrieves a column by board ID and state +func (r *SQLiteRepository) GetColumnByState(ctx context.Context, boardID string, state v1.TaskState) (*models.Column, error) { + column := &models.Column{} + + err := r.db.QueryRowContext(ctx, ` + SELECT id, board_id, name, position, state, color, created_at, updated_at + FROM columns WHERE board_id = ? AND state = ? + `, boardID, state).Scan(&column.ID, &column.BoardID, &column.Name, &column.Position, &column.State, &column.Color, &column.CreatedAt, &column.UpdatedAt) + + if err == sql.ErrNoRows { + return nil, fmt.Errorf("column not found for board %s with state %s", boardID, state) + } + return column, err +} + // UpdateColumn updates an existing column func (r *SQLiteRepository) UpdateColumn(ctx context.Context, column *models.Column) error { column.UpdatedAt = time.Now().UTC() diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 0fc23b0b56..0182a65faa 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -289,7 +289,8 @@ func (s *Service) ListTasks(ctx context.Context, boardID string) ([]*models.Task return s.repo.ListTasks(ctx, boardID) } -// UpdateTaskState updates the state of a task and publishes a task.state_changed event +// UpdateTaskState updates the state of a task, moves it to the matching column, +// and publishes a task.state_changed event func (s *Service) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) (*models.Task, error) { task, err := s.repo.GetTask(ctx, id) if err != nil { @@ -303,12 +304,43 @@ func (s *Service) UpdateTaskState(ctx context.Context, id string, state v1.TaskS return nil, err } - // Reload task to get updated state + // Find the column that maps to the new state and move the task there + // This ensures the Kanban board stays in sync with task state + if task.BoardID != "" { + column, err := s.repo.GetColumnByState(ctx, task.BoardID, state) + if err != nil { + s.logger.Warn("no column found for state, task will stay in current column", + zap.String("task_id", id), + zap.String("state", string(state)), + zap.Error(err)) + } else if column.ID != task.ColumnID { + // Move task to the new column (keep it at position 0 - top of the column) + if err := s.repo.AddTaskToBoard(ctx, id, task.BoardID, column.ID, 0); err != nil { + s.logger.Error("failed to move task to new column", + zap.String("task_id", id), + zap.String("column_id", column.ID), + zap.Error(err)) + } else { + s.logger.Info("task moved to column matching new state", + zap.String("task_id", id), + zap.String("column_id", column.ID), + zap.String("column_name", column.Name), + zap.String("state", string(state))) + } + } + } + + // Reload task to get updated state and column task, err = s.repo.GetTask(ctx, id) if err != nil { return nil, err } + s.logger.Info("task reloaded after column move", + zap.String("task_id", id), + zap.String("column_id", task.ColumnID), + zap.String("state", string(task.State))) + s.publishTaskEvent(ctx, events.TaskStateChanged, task, &oldState) s.logger.Info("task state changed", zap.String("task_id", id), @@ -819,6 +851,14 @@ func (s *Service) publishTaskEvent(ctx context.Context, eventType string, task * event := bus.NewEvent(eventType, "task-service", data) + // Debug logging for state changes + s.logger.Info("publishing task event", + zap.String("event_type", eventType), + zap.String("task_id", task.ID), + zap.String("board_id", task.BoardID), + zap.String("column_id", task.ColumnID), + zap.String("state", string(task.State))) + if err := s.eventBus.Publish(ctx, eventType, event); err != nil { s.logger.Error("failed to publish task event", zap.String("event_type", eventType), diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index d9e83b9ac9..bf217482f4 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -97,6 +97,9 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) { // before the API call completes (fixes race condition on first agent start) store.getState().setCommentsTaskId(task.id); + // Clear git status when switching tasks to avoid showing stale data + store.getState().clearGitStatus(); + const fetchComments = async () => { const client = getWebSocketClient(); if (!client) return; diff --git a/apps/web/components/task/task-changes-panel.tsx b/apps/web/components/task/task-changes-panel.tsx index 34ea6822c4..bde779e97b 100644 --- a/apps/web/components/task/task-changes-panel.tsx +++ b/apps/web/components/task/task-changes-panel.tsx @@ -9,7 +9,7 @@ import { Button } from '@kandev/ui/button'; import { Tooltip, TooltipContent, TooltipTrigger } from '@kandev/ui/tooltip'; import { cn } from '@/lib/utils'; import { getLocalStorage, setLocalStorage } from '@/lib/local-storage'; -import { buildDiffData, CHANGED_FILES, DIFF_SAMPLES } from '@/components/task/task-data'; +import { useAppStore } from '@/components/state-provider'; type TaskChangesPanelProps = { selectedDiffPath: string | null; @@ -26,17 +26,25 @@ const TaskChangesPanel = memo(function TaskChangesPanel({ getLocalStorage('task-diff-view-mode', DEFAULT_DIFF_MODE) ); const { resolvedTheme } = useTheme(); + const gitStatus = useAppStore((state) => state.gitStatus); + + // Convert git status files to array for display + const changedFiles = useMemo(() => { + if (!gitStatus.files || Object.keys(gitStatus.files).length === 0) { + return []; + } + return Object.values(gitStatus.files); + }, [gitStatus.files]); const diffTargets = useMemo( - () => (selectedDiffPath ? [selectedDiffPath] : CHANGED_FILES.map((file) => file.path)), - [selectedDiffPath] + () => (selectedDiffPath ? [selectedDiffPath] : changedFiles.map((file) => file.path)), + [selectedDiffPath, changedFiles] ); const diffModeEnum = diffViewMode === 'split' ? DiffModeEnum.Split : DiffModeEnum.Unified; const diffTheme = resolvedTheme === 'dark' ? 'dark' : 'light'; - const selectedDiffContent = selectedDiffPath - ? DIFF_SAMPLES[selectedDiffPath]?.newContent ?? '' - : ''; - const isSingleDiffSelected = Boolean(selectedDiffPath && DIFF_SAMPLES[selectedDiffPath]); + const selectedFile = selectedDiffPath ? gitStatus.files[selectedDiffPath] : null; + const selectedDiffContent = selectedFile?.diff ?? ''; + const isSingleDiffSelected = Boolean(selectedDiffPath && selectedFile); return (
@@ -146,20 +154,63 @@ const TaskChangesPanel = memo(function TaskChangesPanel({
-
- {diffTargets.map((path) => ( -
- {!selectedDiffPath && ( -
- - {path} - + {changedFiles.length === 0 ? ( +
+ No changes detected +
+ ) : ( +
+ {diffTargets.map((path) => { + const file = gitStatus.files[path]; + if (!file || !file.diff) { + return null; + } + + // Detect file language from extension + const ext = path.split('.').pop() || ''; + const langMap: Record = { + ts: 'typescript', + tsx: 'tsx', + js: 'javascript', + jsx: 'jsx', + py: 'python', + go: 'go', + rs: 'rust', + java: 'java', + cpp: 'cpp', + c: 'c', + css: 'css', + html: 'html', + json: 'json', + md: 'markdown', + yaml: 'yaml', + yml: 'yaml', + }; + const fileLang = langMap[ext] || 'plaintext'; + + return ( +
+ {!selectedDiffPath && ( +
+ + {path} + +
+ )} +
- )} - -
- ))} -
+ ); + })} +
+ )}
); diff --git a/apps/web/components/task/task-files-panel.tsx b/apps/web/components/task/task-files-panel.tsx index e80c3363e2..dd7c6b0136 100644 --- a/apps/web/components/task/task-files-panel.tsx +++ b/apps/web/components/task/task-files-panel.tsx @@ -1,7 +1,7 @@ 'use client'; import type { CSSProperties } from 'react'; -import { memo, useState } from 'react'; +import { memo, useMemo, useState } from 'react'; import { IconArrowBackUp, IconChevronRight, @@ -31,7 +31,9 @@ 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 { CHANGED_FILES, FILE_TREE } from '@/components/task/task-data'; +import { FILE_TREE } from '@/components/task/task-data'; +import { useAppStore } from '@/components/state-provider'; +import type { FileInfo } from '@/lib/state/store'; type TaskFilesPanelProps = { onSelectDiffPath: (path: string) => void; @@ -42,11 +44,24 @@ type TreeItem = string | TreeItem[]; const badgeClass = (status: string) => cn( 'text-[10px] font-semibold', - status === 'M' && 'bg-yellow-500/15 text-yellow-700', - status === 'A' && 'bg-emerald-500/15 text-emerald-700', - status === 'D' && 'bg-rose-500/15 text-rose-700' + (status === 'M' || status === 'modified') && 'bg-yellow-500/15 text-yellow-700', + (status === 'A' || status === 'added') && 'bg-emerald-500/15 text-emerald-700', + (status === 'D' || status === 'deleted') && 'bg-rose-500/15 text-rose-700', + status === 'untracked' && 'bg-blue-500/15 text-blue-700', + status === 'renamed' && 'bg-purple-500/15 text-purple-700' ); +const statusLabel = (status: FileInfo['status']): string => { + switch (status) { + case 'modified': return 'M'; + case 'added': return 'A'; + case 'deleted': return 'D'; + case 'untracked': return 'U'; + case 'renamed': return 'R'; + default: return '?'; + } +}; + const splitPath = (path: string) => { const lastSlash = path.lastIndexOf('/'); if (lastSlash === -1) return { folder: '', file: path }; @@ -95,13 +110,28 @@ function Tree({ item }: { item: TreeItem }) { const TaskFilesPanel = memo(function TaskFilesPanel({ onSelectDiffPath }: TaskFilesPanelProps) { const [topTab, setTopTab] = useState<'diff' | 'files'>('diff'); + const gitStatus = useAppStore((state) => state.gitStatus); + + // Convert git status files to array for display + const changedFiles = useMemo(() => { + if (!gitStatus.files || Object.keys(gitStatus.files).length === 0) { + return []; + } + return Object.values(gitStatus.files).map((file) => ({ + path: file.path, + status: file.status, + plus: file.additions, + minus: file.deletions, + oldPath: file.old_path, + })); + }, [gitStatus.files]); return (
setTopTab(value as 'diff' | 'files')} className="flex-1 min-h-0"> - Diff files + Diff files {changedFiles.length > 0 && `(${changedFiles.length})`} All files @@ -109,59 +139,70 @@ const TaskFilesPanel = memo(function TaskFilesPanel({ onSelectDiffPath }: TaskFi
-
    - {CHANGED_FILES.map((file) => { - const { folder, file: name } = splitPath(file.path); - return ( -
  • onSelectDiffPath(file.path)} - > - -
    -
    - - - - - Discard changes - - - - - - Open in editor - + {changedFiles.length === 0 ? ( +
    + No changes detected +
    + ) : ( +
      + {changedFiles.map((file) => { + const { folder, file: name } = splitPath(file.path); + return ( +
    • onSelectDiffPath(file.path)} + > + +
      +
      + + + + + Discard changes + + + + + + Open in editor + +
      + + {statusLabel(file.status)}
      - - {file.status} -
    -
  • - ); - })} -
+ + ); + })} + + )}
diff --git a/apps/web/components/task/task-top-bar.tsx b/apps/web/components/task/task-top-bar.tsx index 7c6557df24..dba80363e7 100644 --- a/apps/web/components/task/task-top-bar.tsx +++ b/apps/web/components/task/task-top-bar.tsx @@ -1,6 +1,6 @@ 'use client'; -import { memo, useState } from 'react'; +import { memo, useMemo, useState } from 'react'; import Link from 'next/link'; import { IconArrowDown, @@ -25,6 +25,7 @@ import { CommitStatBadge, LineStat } from '@/components/diff-stat'; import { DropdownMenu, DropdownMenuContent, DropdownMenuItem, DropdownMenuTrigger } from '@kandev/ui/dropdown-menu'; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from '@kandev/ui/select'; import { Tooltip, TooltipContent, TooltipTrigger } from '@kandev/ui/tooltip'; +import { useAppStore } from '@/components/state-provider'; type TaskTopBarProps = { taskTitle?: string; @@ -55,6 +56,24 @@ const TaskTopBar = memo(function TaskTopBar({ const [isEditingBranch, setIsEditingBranch] = useState(false); const [selectedBaseBranch, setSelectedBaseBranch] = useState(baseBranch ?? 'origin/main'); + // Get git status from store + const gitStatus = useAppStore((state) => state.gitStatus); + + // Calculate total additions and deletions from all files + const { totalAdditions, totalDeletions } = useMemo(() => { + if (!gitStatus.files || Object.keys(gitStatus.files).length === 0) { + return { totalAdditions: 0, totalDeletions: 0 }; + } + + return Object.values(gitStatus.files).reduce( + (acc, file) => ({ + totalAdditions: acc.totalAdditions + (file.additions || 0), + totalDeletions: acc.totalDeletions + (file.deletions || 0), + }), + { totalAdditions: 0, totalDeletions: 0 } + ); + }, [gitStatus.files]); + return (
@@ -144,32 +163,46 @@ const TaskTopBar = memo(function TaskTopBar({
-
+ {/* Git Status: Ahead/Behind */} + {(gitStatus.ahead > 0 || gitStatus.behind > 0) && ( +
+ {gitStatus.ahead > 0 && ( + + + + + + + + {gitStatus.ahead} commit{gitStatus.ahead !== 1 ? 's' : ''} ahead of {gitStatus.remote_branch || 'remote'} + + + )} + {gitStatus.behind > 0 && ( + + + + + + + + {gitStatus.behind} commit{gitStatus.behind !== 1 ? 's' : ''} behind {gitStatus.remote_branch || 'remote'} + + + )} +
+ )} + {/* Git Status: Lines Changed */} + {(totalAdditions > 0 || totalDeletions > 0) && ( - + - Commits ahead of base + Lines changed - - - - - - - Commits behind base - -
- - - - - - - Lines changed - + )} {/* Worktree Info */} {worktreePath && (
diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index decc6aa0c4..b36e7b6726 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -38,6 +38,30 @@ export type TerminalState = { terminals: Array<{ id: string; output: string[] }>; }; +export type FileInfo = { + path: string; + status: 'modified' | 'added' | 'deleted' | 'untracked' | 'renamed'; + additions?: number; + deletions?: number; + old_path?: string; + diff?: string; +}; + +export type GitStatusState = { + taskId: string | null; + branch: string | null; + remote_branch: string | null; + modified: string[]; + added: string[]; + deleted: string[]; + untracked: string[]; + renamed: string[]; + ahead: number; + behind: number; + files: Record; + timestamp: string | null; +}; + export type DiffState = { files: Array<{ path: string; status: 'A' | 'M' | 'D'; plus: number; minus: number }>; }; @@ -61,6 +85,7 @@ export type AppState = { agents: AgentState; terminal: TerminalState; diffs: DiffState; + gitStatus: GitStatusState; connection: ConnectionState; comments: CommentsState; hydrate: (state: Partial) => void; @@ -74,6 +99,8 @@ export type AppState = { setCommentsTaskId: (taskId: string | null) => void; addComment: (comment: Comment) => void; setCommentsLoading: (loading: boolean) => void; + setGitStatus: (taskId: string, gitStatus: Omit) => void; + clearGitStatus: () => void; }; export type AppStore = ReturnType; @@ -86,6 +113,20 @@ const defaultState: AppState = { agents: { agents: [] }, terminal: { terminals: [] }, diffs: { files: [] }, + gitStatus: { + taskId: null, + branch: null, + remote_branch: null, + modified: [], + added: [], + deleted: [], + untracked: [], + renamed: [], + ahead: 0, + behind: 0, + files: {}, + timestamp: null, + }, connection: { status: 'disconnected', error: null }, comments: { taskId: null, items: [], isLoading: false }, hydrate: () => undefined, @@ -99,9 +140,11 @@ const defaultState: AppState = { setCommentsTaskId: () => undefined, addComment: () => undefined, setCommentsLoading: () => undefined, + setGitStatus: () => undefined, + clearGitStatus: () => undefined, }; -function mergeInitialState(initialState?: Partial): Omit { +function mergeInitialState(initialState?: Partial): Omit { if (!initialState) return defaultState; return { workspaces: { ...defaultState.workspaces, ...initialState.workspaces }, @@ -111,6 +154,7 @@ function mergeInitialState(initialState?: Partial): Omit) { set((draft) => { draft.comments.isLoading = loading; }), + setGitStatus: (taskId, gitStatus) => + set((draft) => { + draft.gitStatus = { + taskId, + ...gitStatus, + }; + }), + clearGitStatus: () => + set((draft) => { + draft.gitStatus = { + taskId: null, + branch: null, + remote_branch: null, + modified: [], + added: [], + deleted: [], + untracked: [], + renamed: [], + ahead: 0, + behind: 0, + files: {}, + timestamp: null, + }; + }), })) ); } diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index 35ef47a02d..d68b550a59 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -7,6 +7,7 @@ export type BackendMessageType = | 'agent.updated' | 'terminal.output' | 'diff.update' + | 'git.status' | 'system.error' | 'workspace.created' | 'workspace.updated' @@ -121,6 +122,30 @@ export type CommentAddedPayload = { created_at: string; }; +export type FileInfo = { + path: string; + status: 'modified' | 'added' | 'deleted' | 'untracked' | 'renamed'; + additions?: number; + deletions?: number; + old_path?: string; + diff?: string; +}; + +export type GitStatusPayload = { + task_id: string; + branch: string; + remote_branch?: string; + modified: string[]; + added: string[]; + deleted: string[]; + untracked: string[]; + renamed: string[]; + ahead: number; + behind: number; + files: Record; + timestamp: string; +}; + export type BackendMessageMap = { 'kanban.update': BackendMessage<'kanban.update', KanbanUpdatePayload>; 'task.created': BackendMessage<'task.created', TaskEventPayload>; @@ -130,6 +155,7 @@ export type BackendMessageMap = { 'agent.updated': BackendMessage<'agent.updated', AgentUpdatePayload>; 'terminal.output': BackendMessage<'terminal.output', TerminalOutputPayload>; 'diff.update': BackendMessage<'diff.update', DiffUpdatePayload>; + 'git.status': BackendMessage<'git.status', GitStatusPayload>; 'system.error': BackendMessage<'system.error', SystemErrorPayload>; 'workspace.created': BackendMessage<'workspace.created', WorkspacePayload>; 'workspace.updated': BackendMessage<'workspace.updated', WorkspacePayload>; diff --git a/apps/web/lib/ws/router.ts b/apps/web/lib/ws/router.ts index bbc7c2803a..d974e879ee 100644 --- a/apps/web/lib/ws/router.ts +++ b/apps/web/lib/ws/router.ts @@ -85,10 +85,20 @@ export function registerWsHandlers(store: StoreApi) { })); }, 'task.state_changed': (message: BackendMessageMap['task.state_changed']) => { + console.log('[WS Router] task.state_changed received:', { + task_id: message.payload.task_id, + board_id: message.payload.board_id, + column_id: message.payload.column_id, + state: message.payload.state, + }); store.setState((state) => { + console.log('[WS Router] Current board_id:', state.kanban.boardId, 'Event board_id:', message.payload.board_id); if (state.kanban.boardId !== message.payload.board_id) { + console.log('[WS Router] Skipping - board_id mismatch'); return state; } + const existingTask = state.kanban.tasks.find((t) => t.id === message.payload.task_id); + console.log('[WS Router] Existing task:', existingTask); const nextTask = { id: message.payload.task_id, columnId: message.payload.column_id, @@ -97,6 +107,7 @@ export function registerWsHandlers(store: StoreApi) { position: message.payload.position ?? 0, state: message.payload.state, }; + console.log('[WS Router] Next task:', nextTask); return { ...state, kanban: { @@ -331,5 +342,41 @@ export function registerWsHandlers(store: StoreApi) { }); console.log('[WS] addComment called'); }, + 'git.status': (message: BackendMessageMap['git.status']) => { + const payload = message.payload; + console.log('[WS] git.status received:', { + task_id: payload.task_id, + branch: payload.branch, + modified: payload.modified.length, + added: payload.added.length, + deleted: payload.deleted.length, + untracked: payload.untracked.length, + }); + const state = store.getState(); + + // Only update git status if it's for the current task + // This prevents stale data from showing when switching tasks + if (state.comments.taskId !== payload.task_id) { + console.log('[WS] Ignoring git.status for different task:', { + current_task: state.comments.taskId, + received_task: payload.task_id, + }); + return; + } + + state.setGitStatus(payload.task_id, { + branch: payload.branch, + remote_branch: payload.remote_branch ?? null, + modified: payload.modified, + added: payload.added, + deleted: payload.deleted, + untracked: payload.untracked, + renamed: payload.renamed, + ahead: payload.ahead, + behind: payload.behind, + files: payload.files, + timestamp: payload.timestamp, + }); + }, }; }