diff --git a/apps/backend/internal/agent/handlers/shell_handlers.go b/apps/backend/internal/agent/handlers/shell_handlers.go index cc5df32c0c..fea5684600 100644 --- a/apps/backend/internal/agent/handlers/shell_handlers.go +++ b/apps/backend/internal/agent/handlers/shell_handlers.go @@ -4,6 +4,7 @@ package handlers import ( "context" "fmt" + "time" "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/common/logger" @@ -150,10 +151,23 @@ func (h *ShellHandlers) wsShellInput(ctx context.Context, msg *ws.Message) (*ws. return nil, fmt.Errorf("no agent running for session %s", req.SessionID) } - // Use the workspace stream from AgentExecution (managed by lifecycle) - workspaceStream := execution.GetWorkspaceStream() + // Wait for the workspace stream to be ready with a timeout. + // This handles the race condition where client sends shell.input before + // the workspace stream is fully connected (e.g., joining a task too fast). + var workspaceStream = execution.GetWorkspaceStream() if workspaceStream == nil { - return nil, fmt.Errorf("no workspace stream for session %s", req.SessionID) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + time.Sleep(100 * time.Millisecond) + workspaceStream = execution.GetWorkspaceStream() + if workspaceStream != nil { + break + } + } + } + + if workspaceStream == nil { + return nil, fmt.Errorf("workspace stream not ready for session %s", req.SessionID) } if err := workspaceStream.WriteShellInput(req.Data); err != nil { diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 4aae487dc3..99d8624967 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -288,7 +288,7 @@ func (m *Manager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID if m.streamManager != nil { m.logger.Info("connecting workspace stream for workspace-only execution", zap.String("execution_id", execution.ID)) - go m.streamManager.connectWorkspaceStream(execution) + go m.streamManager.connectWorkspaceStream(execution, nil) } }() @@ -1732,7 +1732,12 @@ func (m *Manager) StopAgent(ctx context.Context, executionID string, force bool) StandalonePort: execution.standalonePort, } if err := rt.StopInstance(ctx, runtimeInstance, force); err != nil { - return err + // Log the error but don't return - we still need to clean up the execution store. + // This ensures that even if the runtime is unavailable (e.g., process crashed), + // the execution is removed from tracking so new executions can be launched. + m.logger.Warn("failed to stop runtime instance, continuing with cleanup", + zap.String("execution_id", executionID), + zap.Error(err)) } } } @@ -2372,7 +2377,7 @@ func (m *Manager) startPassthroughSession(ctx context.Context, execution *AgentE // The passthrough terminal handles agent interaction (center panel), while the // workspace stream provides shell I/O (right panel) - they work independently. if m.streamManager != nil && execution.agentctl != nil { - go m.streamManager.connectWorkspaceStream(execution) + go m.streamManager.connectWorkspaceStream(execution, nil) } return nil @@ -2558,7 +2563,7 @@ func (m *Manager) ResumePassthroughSession(ctx context.Context, sessionID string // Connect to workspace stream for shell/git/file features. // This re-establishes the connection that was lost on backend restart. if m.streamManager != nil && execution.agentctl != nil { - go m.streamManager.connectWorkspaceStream(execution) + go m.streamManager.connectWorkspaceStream(execution, nil) } return nil diff --git a/apps/backend/internal/agent/lifecycle/streams.go b/apps/backend/internal/agent/lifecycle/streams.go index 2ab831c73d..d4231c7312 100644 --- a/apps/backend/internal/agent/lifecycle/streams.go +++ b/apps/backend/internal/agent/lifecycle/streams.go @@ -38,10 +38,23 @@ func NewStreamManager(log *logger.Logger, callbacks StreamCallbacks) *StreamMana } // ConnectAll connects to all streams for an execution. -// If ready is non-nil, it will be closed when the updates stream is connected. +// If ready is non-nil, it will be closed when both streams are connected. func (sm *StreamManager) ConnectAll(execution *AgentExecution, ready chan<- struct{}) { - go sm.connectUpdatesStream(execution, ready) - go sm.connectWorkspaceStream(execution) + // Create channels to track when each stream is connected + updatesReady := make(chan struct{}) + workspaceReady := make(chan struct{}) + + go sm.connectUpdatesStream(execution, updatesReady) + go sm.connectWorkspaceStream(execution, workspaceReady) + + // If caller wants to know when streams are ready, wait for both + if ready != nil { + go func() { + <-updatesReady + <-workspaceReady + close(ready) + }() + } } // ReconnectAll reconnects to all streams (used after backend restart). @@ -98,12 +111,24 @@ func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready c } // connectWorkspaceStream handles the unified workspace stream with retry logic -func (sm *StreamManager) connectWorkspaceStream(execution *AgentExecution) { +func (sm *StreamManager) connectWorkspaceStream(execution *AgentExecution, ready chan<- struct{}) { ctx := context.Background() // Retry connection with exponential backoff maxRetries := 5 backoff := 1 * time.Second + signaled := false + + // Helper to signal ready (only once) + signalReady := func() { + if !signaled && ready != nil { + close(ready) + signaled = true + } + } + + // Ensure we signal ready even on failure (so callers don't hang) + defer signalReady() for attempt := 1; attempt <= maxRetries; attempt++ { callbacks := agentctl.WorkspaceStreamCallbacks{ @@ -178,6 +203,9 @@ func (sm *StreamManager) connectWorkspaceStream(execution *AgentExecution) { sm.logger.Debug("connected to unified workspace stream", zap.String("instance_id", execution.ID)) + // Signal that workspace stream is ready + signalReady() + // Wait for the stream to close (it stays open until disconnected) <-ws.Done() return diff --git a/apps/backend/internal/agentctl/server/adapter/acp_adapter.go b/apps/backend/internal/agentctl/server/adapter/acp_adapter.go index bfc986a9f1..8542b5121d 100644 --- a/apps/backend/internal/agentctl/server/adapter/acp_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/acp_adapter.go @@ -319,6 +319,11 @@ func (a *ACPAdapter) Close() error { return nil } +// RequiresProcessKill returns false because ACP agents exit when stdin is closed. +func (a *ACPAdapter) RequiresProcessKill() bool { + return false +} + // GetACPConnection returns the underlying ACP connection for advanced usage. func (a *ACPAdapter) GetACPConnection() *acp.ClientSideConnection { a.mu.RLock() diff --git a/apps/backend/internal/agentctl/server/adapter/adapter.go b/apps/backend/internal/agentctl/server/adapter/adapter.go index 4cb305c641..c2b8612817 100644 --- a/apps/backend/internal/agentctl/server/adapter/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/adapter.go @@ -118,6 +118,13 @@ type AgentAdapter interface { // Close releases resources held by the adapter. Close() error + + // RequiresProcessKill returns true if the adapter's subprocess needs to be + // explicitly killed during shutdown. Adapters that communicate via stdin/stdout + // (ACP, Codex, Claude Code) return false because closing stdin causes the + // subprocess to exit. HTTP-server-based adapters (OpenCode) return true because + // they don't exit on stdin close. + RequiresProcessKill() bool } // McpServerConfig holds configuration for an MCP server. diff --git a/apps/backend/internal/agentctl/server/adapter/claude_code_adapter.go b/apps/backend/internal/agentctl/server/adapter/claude_code_adapter.go index 85c4111347..537ee98de6 100644 --- a/apps/backend/internal/agentctl/server/adapter/claude_code_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/claude_code_adapter.go @@ -304,6 +304,11 @@ func (a *ClaudeCodeAdapter) Close() error { return nil } +// RequiresProcessKill returns false because Claude Code agents exit when stdin is closed. +func (a *ClaudeCodeAdapter) RequiresProcessKill() bool { + return false +} + // sendUpdate safely sends an event to the updates channel. func (a *ClaudeCodeAdapter) sendUpdate(update AgentEvent) { select { diff --git a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go index 365a35c815..b2d24cdfd3 100644 --- a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go @@ -579,6 +579,11 @@ func (a *CodexAdapter) Close() error { return nil } +// RequiresProcessKill returns false because Codex agents exit when stdin is closed. +func (a *CodexAdapter) RequiresProcessKill() bool { + return false +} + // sendUpdate safely sends an event to the updates channel. func (a *CodexAdapter) sendUpdate(update AgentEvent) { select { diff --git a/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go b/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go index cb02a3a507..32946c31c9 100644 --- a/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go @@ -421,26 +421,38 @@ func (a *OpenCodeAdapter) SetPermissionHandler(handler PermissionHandler) { // Close releases resources held by the adapter. func (a *OpenCodeAdapter) Close() error { + a.logger.Info("OpenCode adapter Close() called") + a.mu.Lock() defer a.mu.Unlock() if a.closed { + a.logger.Debug("OpenCode adapter already closed") return nil } a.closed = true + a.logger.Debug("OpenCode adapter: cancelling context") a.cancel() + a.logger.Debug("OpenCode adapter: closing HTTP client") if a.client != nil { a.client.Close() } + a.logger.Debug("OpenCode adapter: closing updates channel") close(a.updatesCh) - a.logger.Info("OpenCode adapter closed") + a.logger.Info("OpenCode adapter closed successfully") return nil } +// RequiresProcessKill returns true because OpenCode runs as an HTTP server +// and does not exit when stdin is closed. +func (a *OpenCodeAdapter) RequiresProcessKill() bool { + return true +} + // clearSessionState clears session-specific tracking state. // This should be called when starting a new session to prevent memory leaks // and stale data from previous sessions. diff --git a/apps/backend/internal/agentctl/server/instance/manager.go b/apps/backend/internal/agentctl/server/instance/manager.go index babefbf868..38adf399cc 100644 --- a/apps/backend/internal/agentctl/server/instance/manager.go +++ b/apps/backend/internal/agentctl/server/instance/manager.go @@ -226,6 +226,8 @@ func (m *Manager) StopInstance(ctx context.Context, id string) error { return fmt.Errorf("instance %s not found", id) } + m.logger.Debug("stopping instance", zap.String("instance_id", id)) + // Stop the process manager if inst.manager != nil { if err := inst.manager.Stop(ctx); err != nil { @@ -235,6 +237,10 @@ func (m *Manager) StopInstance(ctx context.Context, id string) error { } } + m.logger.Debug("StopInstance: shutting down HTTP server", + zap.String("instance_id", id), + zap.Int("port", inst.Port)) + // Shutdown HTTP server if inst.server != nil { if err := inst.server.Shutdown(ctx); err != nil { @@ -244,13 +250,17 @@ func (m *Manager) StopInstance(ctx context.Context, id string) error { } } + m.logger.Debug("StopInstance: releasing port", + zap.String("instance_id", id), + zap.Int("port", inst.Port)) + // Release port m.portAlloc.Release(inst.Port) // Remove from instances map delete(m.instances, id) - m.logger.Info("stopped instance", + m.logger.Info("StopInstance completed", zap.String("instance_id", id), zap.Int("port", inst.Port)) diff --git a/apps/backend/internal/agentctl/server/process/manager.go b/apps/backend/internal/agentctl/server/process/manager.go index 38a9c75429..9a6a0204a7 100644 --- a/apps/backend/internal/agentctl/server/process/manager.go +++ b/apps/backend/internal/agentctl/server/process/manager.go @@ -225,6 +225,10 @@ func (m *Manager) Start(ctx context.Context) error { m.cmd = exec.Command(m.cfg.AgentArgs[0], m.cfg.AgentArgs[1:]...) m.cmd.Dir = m.cfg.WorkDir m.cmd.Env = m.cfg.AgentEnv + // Create a new process group so we can kill all child processes together. + // This is important for adapters like OpenCode that spawn child processes + // (npx -> sh -> node -> opencode binary). + setProcGroup(m.cmd) // Set up pipes var err error @@ -476,52 +480,85 @@ func (m *Manager) Stop(ctx context.Context) error { status := m.Status() if status == StatusStopped || status == StatusStopping { + m.logger.Info("Stop called but already stopped/stopping", zap.String("status", string(status))) return nil } - m.logger.Info("stopping agent process") + m.logger.Info("stopping agent process - START") m.status.Store(StatusStopping) // Stop shell session + m.logger.Debug("stopping shell session") if m.shell != nil { if err := m.shell.Stop(); err != nil { m.logger.Debug("failed to stop shell session", zap.Error(err)) } m.shell = nil } + m.logger.Debug("shell session stopped") // Stop all running workspace processes (dev server, setup, cleanup, custom). + m.logger.Debug("stopping workspace processes") if m.processRunner != nil { if err := m.processRunner.StopAll(ctx); err != nil { m.logger.Debug("failed to stop workspace processes", zap.Error(err)) } } + m.logger.Debug("workspace processes stopped") // Stop workspace tracker + m.logger.Debug("stopping workspace tracker") if m.workspaceTracker != nil { m.workspaceTracker.Stop() } + m.logger.Debug("workspace tracker stopped") // Close the adapter + m.logger.Debug("closing adapter") if m.adapter != nil { if err := m.adapter.Close(); err != nil { m.logger.Debug("failed to close adapter", zap.Error(err)) } } + m.logger.Debug("adapter closed") // Close stop channel to signal readers + m.logger.Debug("closing stop channel") if m.stopCh != nil { close(m.stopCh) } + m.logger.Debug("stop channel closed") // Close stdin to signal EOF to agent + m.logger.Debug("closing stdin") if m.stdin != nil { if err := m.stdin.Close(); err != nil { m.logger.Debug("failed to close stdin", zap.Error(err)) } } + m.logger.Debug("stdin closed") + + // Some adapters (like OpenCode) run as HTTP servers and don't exit on stdin close. + // For these, we need to kill the process group immediately. + // We kill the process group to ensure all child processes are killed too. + // This is important because OpenCode spawns: npx -> sh -> node -> opencode binary + if m.adapter != nil && m.adapter.RequiresProcessKill() { + if m.cmd != nil && m.cmd.Process != nil { + pid := m.cmd.Process.Pid + m.logger.Debug("killing process group", zap.Int("pgid", pid)) + // Kill the entire process group (platform-specific implementation) + if err := killProcessGroup(pid); err != nil { + m.logger.Debug("failed to kill process group, trying single process", zap.Error(err)) + // Fallback to killing just the process + if err := m.cmd.Process.Kill(); err != nil { + m.logger.Warn("failed to kill process", zap.Error(err)) + } + } + } + } // Wait for process to exit with timeout + m.logger.Debug("waiting for process to exit") done := make(chan struct{}) go func() { m.wg.Wait() @@ -542,6 +579,7 @@ func (m *Manager) Stop(ctx context.Context) error { } m.status.Store(StatusStopped) + m.logger.Info("stopping agent process - COMPLETE") return nil } diff --git a/apps/backend/internal/agentctl/server/process/procattr_unix.go b/apps/backend/internal/agentctl/server/process/procattr_unix.go new file mode 100644 index 0000000000..7fdc90d8d5 --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/procattr_unix.go @@ -0,0 +1,22 @@ +//go:build unix + +package process + +import ( + "os/exec" + "syscall" +) + +// setProcGroup configures the command to run in its own process group. +// This allows us to kill all child processes together. +func setProcGroup(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} +} + +// killProcessGroup kills the entire process group for the given PID. +// Returns nil if successful, or an error if the kill failed. +func killProcessGroup(pid int) error { + // Kill the entire process group by using negative PID + return syscall.Kill(-pid, syscall.SIGKILL) +} + diff --git a/apps/backend/internal/agentctl/server/process/procattr_windows.go b/apps/backend/internal/agentctl/server/process/procattr_windows.go new file mode 100644 index 0000000000..8fa5dfb567 --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/procattr_windows.go @@ -0,0 +1,29 @@ +//go:build windows + +package process + +import ( + "fmt" + "os/exec" + "syscall" +) + +// setProcGroup configures the command to run in its own process group. +// On Windows, we use CREATE_NEW_PROCESS_GROUP flag. +func setProcGroup(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{ + CreationFlags: syscall.CREATE_NEW_PROCESS_GROUP, + } +} + +// killProcessGroup kills the entire process tree for the given PID. +// On Windows, we use taskkill with /T flag to kill the process tree. +func killProcessGroup(pid int) error { + // taskkill /F /T /PID + // /F = force kill + // /T = kill child processes (tree) + // /PID = process ID + kill := exec.Command("taskkill", "/F", "/T", "/PID", fmt.Sprintf("%d", pid)) + return kill.Run() +} + diff --git a/apps/backend/internal/agentctl/server/process/workspace_tracker.go b/apps/backend/internal/agentctl/server/process/workspace_tracker.go index e177e2d046..cb49cb3c4c 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_tracker.go +++ b/apps/backend/internal/agentctl/server/process/workspace_tracker.go @@ -697,8 +697,12 @@ func (wt *WorkspaceTracker) enrichWithDiffData(ctx context.Context, update *type // uncommitted changes in the "files changed" count. } - // Also check for staged changes - stagedCmd := exec.CommandContext(ctx, "git", "diff", "--cached", "--numstat") + // For staged files that don't have unstaged changes, we need to get the diff from the index. + // The first diff (git diff baseRef) shows worktree vs baseRef, but if a file is staged + // and has no additional unstaged changes, its diff won't appear there. + // We use git diff --cached baseRef to get the staged diff content for such files, + // using the same baseRef for consistency. + stagedCmd := exec.CommandContext(ctx, "git", "diff", "--cached", "--numstat", baseRef) stagedCmd.Dir = wt.workDir stagedOut, err := stagedCmd.Output() if err != nil { @@ -723,18 +727,19 @@ func (wt *WorkspaceTracker) enrichWithDiffData(ctx context.Context, update *type // 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 + // Only set additions/deletions if they weren't already set by the first diff. + // This prevents double-counting when changes appear in both diffs. + if fileInfo.Additions == 0 && fileInfo.Deletions == 0 { + 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 { + // Get the staged diff content if we don't have diff content yet + // Use the same baseRef for consistency with unstaged files + if fileInfo.Diff == "" { + diffCmd := exec.CommandContext(ctx, "git", "diff", "--cached", baseRef, "--", filePath) + diffCmd.Dir = wt.workDir + if diffOut, err := diffCmd.Output(); err == nil { fileInfo.Diff = string(diffOut) } } @@ -1115,12 +1120,9 @@ func (wt *WorkspaceTracker) buildFileTreeNode(fullPath, relPath string, info os. // Build children node.Children = make([]*types.FileTreeNode, 0, len(entries)) for _, entry := range entries { - // Skip hidden files and common ignore patterns + // Skip specific directories that should be ignored name := entry.Name() - if strings.HasPrefix(name, ".") && name != "." && name != ".." { - continue - } - if name == "node_modules" || name == ".next" || name == "dist" || name == "build" { + if name == ".git" || name == "node_modules" || name == ".next" || name == "dist" || name == "build" { continue } diff --git a/apps/backend/pkg/opencode/client.go b/apps/backend/pkg/opencode/client.go index b028be15e0..03ae848c08 100644 --- a/apps/backend/pkg/opencode/client.go +++ b/apps/backend/pkg/opencode/client.go @@ -108,6 +108,34 @@ func (c *Client) doRequest(ctx context.Context, method, path string, body io.Rea return c.httpClient.Do(req) } +// doPromptRequest performs an HTTP request with a longer timeout suitable for prompts. +// Prompts can take minutes to complete, so we use a 60-minute timeout. +func (c *Client) doPromptRequest(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) { + url := c.baseURL + path + if strings.Contains(path, "?") { + url += "&directory=" + c.directory + } else { + url += "?directory=" + c.directory + } + + req, err := http.NewRequestWithContext(ctx, method, url, body) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + + req.Header.Set("Authorization", c.buildAuthHeader()) + req.Header.Set("X-OpenCode-Directory", c.directory) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + + // Use a client with extended timeout for prompts + promptClient := &http.Client{ + Timeout: 60 * time.Minute, + } + return promptClient.Do(req) +} + // WaitForHealth waits for the OpenCode server to be healthy func (c *Client) WaitForHealth(ctx context.Context) error { deadline := time.Now().Add(20 * time.Second) @@ -230,7 +258,8 @@ func (c *Client) SendPrompt(ctx context.Context, sessionID, prompt string, model } path := fmt.Sprintf("/session/%s/message", sessionID) - resp, err := c.doRequest(ctx, http.MethodPost, path, strings.NewReader(string(body))) + // Use a dedicated client with long timeout for prompts - they can take minutes + resp, err := c.doPromptRequest(ctx, http.MethodPost, path, strings.NewReader(string(body))) if err != nil { return fmt.Errorf("send prompt request: %w", err) } @@ -422,10 +451,15 @@ func (c *Client) processEventStream(ctx context.Context, sessionID string, body c.logger.Error("event stream error", zap.Error(err)) } - // Notify disconnection - select { - case c.controlCh <- ControlEvent{Type: "disconnected"}: - default: + // Notify disconnection (check if closed first to avoid panic) + c.mu.RLock() + closed := c.closed + c.mu.RUnlock() + if !closed { + select { + case c.controlCh <- ControlEvent{Type: "disconnected"}: + default: + } } } @@ -470,6 +504,14 @@ func (c *Client) eventMatchesSession(event *SDKEventEnvelope, sessionID string) // processControlEvent handles control flow events func (c *Client) processControlEvent(event *SDKEventEnvelope, sessionID string) { + // Check if closed first to avoid panic on closed channel + c.mu.RLock() + closed := c.closed + c.mu.RUnlock() + if closed { + return + } + switch event.Type { case SDKEventSessionIdle: select { diff --git a/apps/web/components/task/task-session-sidebar.tsx b/apps/web/components/task/task-session-sidebar.tsx index 0a19cd3f2b..f12091c8dd 100644 --- a/apps/web/components/task/task-session-sidebar.tsx +++ b/apps/web/components/task/task-session-sidebar.tsx @@ -137,7 +137,7 @@ export function TaskSessionSidebar({ workspaceId, boardId }: TaskSessionSidebarP workspaceId={workspaceId} boardId={boardId} defaultColumnId={columns[0]?.id ?? null} - columns={columns.map((step: KanbanState['steps'][number]) => ({ id: step.id, title: step.title, color: step.color }))} + columns={columns.map((step: KanbanState['steps'][number]) => ({ id: step.id, title: step.title, color: step.color, autoStartAgent: step.autoStartAgent }))} onSuccess={(task, _mode, meta) => { store.setState((state) => { if (state.kanban.boardId !== task.board_id) return state;