diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 18b7104d56..72bc14886b 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -251,6 +251,12 @@ func main() { zap.Bool("enabled", cfg.Worktree.Enabled), zap.String("base_path", cfg.Worktree.BasePath)) + // Set task service as workspace info provider for session recovery + if lifecycleMgr != nil { + lifecycleMgr.SetWorkspaceInfoProvider(taskSvc) + log.Info("Workspace info provider configured for session recovery") + } + // ============================================ // ORCHESTRATOR // ============================================ diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 8d1149385e..a095c06c80 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -294,6 +294,125 @@ func (m *Manager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID return execution, nil } +// EnsurePassthroughExecution ensures an execution exists for a passthrough session +// and starts the passthrough process if needed. This is called when the terminal +// handler receives a connection for a session that might need recovery after backend restart. +// +// The sessionID is required. If taskID is empty, it will be looked up from: +// 1. The existing execution (if any) +// 2. The workspace info provider +// +// Returns the execution with a running passthrough process, or an error. +func (m *Manager) EnsurePassthroughExecution(ctx context.Context, sessionID string) (*AgentExecution, error) { + // Check if execution already exists with a running passthrough process + if execution, exists := m.executionStore.GetBySessionID(sessionID); exists { + if execution.PassthroughProcessID != "" { + return execution, nil + } + // Execution exists but no passthrough process - will try to start it + return m.resumeExistingExecution(ctx, sessionID, execution) + } + + // No execution exists - need to create one from session info + return m.createExecutionFromSessionInfo(ctx, sessionID) +} + +// resumeExistingExecution starts the passthrough process for an existing execution +// that has no running process (e.g., after backend restart). +func (m *Manager) resumeExistingExecution(ctx context.Context, sessionID string, execution *AgentExecution) (*AgentExecution, error) { + m.logger.Info("execution exists but passthrough process not running, starting", + zap.String("session_id", sessionID), + zap.String("execution_id", execution.ID)) + + if err := m.ResumePassthroughSession(ctx, sessionID); err != nil { + return nil, fmt.Errorf("resume passthrough session %s: %w", sessionID, err) + } + + // Get updated execution with process ID + execution, exists := m.executionStore.GetBySessionID(sessionID) + if !exists { + return nil, fmt.Errorf("execution disappeared after resuming passthrough session %s", sessionID) + } + return execution, nil +} + +// createExecutionFromSessionInfo creates a new execution for a passthrough session +// when no execution exists (e.g., backend restarted and execution store was cleared). +func (m *Manager) createExecutionFromSessionInfo(ctx context.Context, sessionID string) (*AgentExecution, error) { + if m.workspaceInfoProvider == nil { + return nil, fmt.Errorf("cannot restore session %s: workspace info provider not configured", sessionID) + } + + // Get workspace info from the provider (looks up session to get taskID, workspace path, etc.) + info, err := m.workspaceInfoProvider.GetWorkspaceInfoForSession(ctx, "", sessionID) + if err != nil { + return nil, fmt.Errorf("get workspace info for session %s: %w", sessionID, err) + } + + if info.WorkspacePath == "" { + return nil, fmt.Errorf("session %s has no workspace path configured", sessionID) + } + + if info.TaskID == "" { + return nil, fmt.Errorf("session %s has no associated task ID", sessionID) + } + + // Verify this session should use passthrough mode + if err := m.verifyPassthroughEnabled(ctx, sessionID, info.AgentProfileID); err != nil { + return nil, err + } + + // Create the execution + m.logger.Info("creating execution for passthrough session", + zap.String("task_id", info.TaskID), + zap.String("session_id", sessionID), + zap.String("workspace_path", info.WorkspacePath)) + + execution, err := m.createExecution(ctx, info.TaskID, info) + if err != nil { + return nil, fmt.Errorf("create execution for session %s: %w", sessionID, err) + } + + // Start the passthrough process using resume command (recovery after restart) + m.logger.Info("starting passthrough process for session", + zap.String("session_id", sessionID), + zap.String("execution_id", execution.ID)) + + if err := m.ResumePassthroughSession(ctx, sessionID); err != nil { + return nil, fmt.Errorf("start passthrough process for session %s: %w", sessionID, err) + } + + // Get updated execution with process ID + execution, exists := m.executionStore.GetBySessionID(sessionID) + if !exists { + return nil, fmt.Errorf("execution disappeared after starting passthrough session %s", sessionID) + } + + return execution, nil +} + +// verifyPassthroughEnabled checks if the session's profile has CLI passthrough enabled. +func (m *Manager) verifyPassthroughEnabled(ctx context.Context, sessionID, profileID string) error { + if m.profileResolver == nil || profileID == "" { + return fmt.Errorf("session %s has no profile configured for passthrough mode", sessionID) + } + + profileInfo, err := m.profileResolver.ResolveProfile(ctx, profileID) + if err != nil { + m.logger.Warn("failed to resolve profile for passthrough check", + zap.String("session_id", sessionID), + zap.String("profile_id", profileID), + zap.Error(err)) + return fmt.Errorf("session %s: failed to resolve profile %s: %w", sessionID, profileID, err) + } + + if profileInfo == nil || !profileInfo.CLIPassthrough { + return fmt.Errorf("session %s is not configured for CLI passthrough mode", sessionID) + } + + return nil +} + // createExecution creates an agentctl execution. // The agent subprocess is NOT started - call ConfigureAgent + Start explicitly. func (m *Manager) createExecution(ctx context.Context, taskID string, info *WorkspaceInfo) (*AgentExecution, error) { @@ -2091,7 +2210,9 @@ func (m *Manager) startPassthroughSession(ctx context.Context, execution *AgentE } // Start the interactive process immediately with default dimensions - // This allows the agent to start working before the user opens the terminal + // Start process - either immediately (default) or wait for terminal connection + // Some agents (like Codex) require the terminal to be connected first because + // they query the terminal for cursor position on startup startReq := process.InteractiveStartRequest{ SessionID: execution.SessionID, Command: cmd, @@ -2103,7 +2224,7 @@ func (m *Manager) startPassthroughSession(ctx context.Context, execution *AgentE StatusDetector: agentConfig.PassthroughConfig.StatusDetector, CheckIntervalMs: agentConfig.PassthroughConfig.CheckIntervalMs, StabilityWindowMs: agentConfig.PassthroughConfig.StabilityWindowMs, - ImmediateStart: true, // Start immediately, don't wait for terminal resize + ImmediateStart: !agentConfig.PassthroughConfig.WaitForTerminal, DefaultCols: 120, DefaultRows: 40, } @@ -2227,6 +2348,168 @@ func (m *Manager) buildPassthroughCommand(agentConfig *registry.AgentTypeConfig, return cmd } +// ResumePassthroughSession restarts a passthrough session after backend restart. +// This is called when user reconnects to a terminal but the PTY process is no longer running. +// If the agent supports resume, it uses the resume flag to continue the last conversation. +// Otherwise, it starts a fresh CLI session with the same profile settings. +func (m *Manager) ResumePassthroughSession(ctx context.Context, sessionID string) error { + execution, exists := m.executionStore.GetBySessionID(sessionID) + if !exists { + return fmt.Errorf("no execution found for session: %s", sessionID) + } + + // Get agent config + agentConfig, err := m.getAgentConfigForExecution(execution) + if err != nil { + return fmt.Errorf("failed to get agent config: %w", err) + } + + if !agentConfig.PassthroughConfig.Supported { + return fmt.Errorf("agent %s does not support passthrough mode", agentConfig.ID) + } + + // Get profile info for permission settings + var profileInfo *AgentProfileInfo + if m.profileResolver != nil && execution.AgentProfileID != "" { + profileInfo, _ = m.profileResolver.ResolveProfile(ctx, execution.AgentProfileID) + } + + // Build the resume command + cmd := m.buildPassthroughResumeCommand(agentConfig, profileInfo) + if len(cmd) == 0 { + return fmt.Errorf("passthrough resume command is empty for agent %s", agentConfig.ID) + } + + m.logger.Info("resuming passthrough session", + zap.String("session_id", sessionID), + zap.String("execution_id", execution.ID), + zap.Strings("command", cmd), + zap.Bool("has_resume_flag", agentConfig.PassthroughConfig.ResumeFlag != "")) + + // Get the interactive runner + interactiveRunner := m.GetInteractiveRunner() + if interactiveRunner == nil { + return fmt.Errorf("interactive runner not available") + } + + // Build environment variables + env := make(map[string]string) + env["KANDEV_TASK_ID"] = execution.TaskID + env["KANDEV_SESSION_ID"] = execution.SessionID + env["KANDEV_AGENT_PROFILE_ID"] = execution.AgentProfileID + + // Add required credentials + if m.credsMgr != nil { + for _, credKey := range agentConfig.RequiredEnv { + if value, err := m.credsMgr.GetCredentialValue(ctx, credKey); err == nil && value != "" { + env[credKey] = value + } + } + } + + // Start the interactive process + // Use the same settings as initial start (including wait_for_terminal) + startReq := process.InteractiveStartRequest{ + SessionID: sessionID, + Command: cmd, + WorkingDir: execution.WorkspacePath, + Env: env, + IdleTimeoutMs: agentConfig.PassthroughConfig.IdleTimeoutMs, + BufferMaxBytes: agentConfig.PassthroughConfig.BufferMaxBytes, + StatusDetector: agentConfig.PassthroughConfig.StatusDetector, + CheckIntervalMs: agentConfig.PassthroughConfig.CheckIntervalMs, + StabilityWindowMs: agentConfig.PassthroughConfig.StabilityWindowMs, + ImmediateStart: !agentConfig.PassthroughConfig.WaitForTerminal, + DefaultCols: 120, + DefaultRows: 40, + } + + processInfo, err := interactiveRunner.Start(ctx, startReq) + if err != nil { + return fmt.Errorf("failed to start passthrough session: %w", err) + } + + // Update the execution with new process ID + execution.PassthroughProcessID = processInfo.ID + + m.logger.Info("passthrough session resumed", + zap.String("session_id", sessionID), + zap.String("execution_id", execution.ID), + zap.String("process_id", processInfo.ID)) + + // Start shell session for workspace shell access (right panel terminal). + // This needs to be done after resume since the shell process was killed on backend restart. + if execution.agentctl != nil { + if err := execution.agentctl.StartShell(ctx); err != nil { + m.logger.Warn("failed to start shell for resumed passthrough session", + zap.String("execution_id", execution.ID), + zap.Error(err)) + // Non-fatal: continue without shell + } else { + m.logger.Info("shell session started for resumed passthrough session", + zap.String("execution_id", execution.ID)) + } + } + + // 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) + } + + return nil +} + +// buildPassthroughResumeCommand builds the command for resuming a passthrough session. +// If the agent supports resume (has ResumeFlag), it adds the resume flag. +// Otherwise, it starts a fresh session with profile settings but no initial prompt. +func (m *Manager) buildPassthroughResumeCommand(agentConfig *registry.AgentTypeConfig, profileInfo *AgentProfileInfo) []string { + // Start with passthrough_cmd + cmd := make([]string, len(agentConfig.PassthroughConfig.PassthroughCmd)) + copy(cmd, agentConfig.PassthroughConfig.PassthroughCmd) + + // Apply model flag if configured + if profileInfo != nil && profileInfo.Model != "" && agentConfig.PassthroughConfig.ModelFlag != "" { + expanded := strings.ReplaceAll(agentConfig.PassthroughConfig.ModelFlag, "{model}", profileInfo.Model) + parts := strings.SplitN(expanded, " ", 2) + cmd = append(cmd, parts...) + } + + // Apply permission settings that use CLI flags + if profileInfo != nil && agentConfig.PermissionSettings != nil { + permissionValues := map[string]bool{ + "auto_approve": profileInfo.AutoApprove, + "dangerously_skip_permissions": profileInfo.DangerouslySkipPermissions, + "allow_indexing": profileInfo.AllowIndexing, + } + + for settingName, setting := range agentConfig.PermissionSettings { + if !setting.Supported || setting.ApplyMethod != "cli_flag" || setting.CLIFlag == "" { + continue + } + value, exists := permissionValues[settingName] + if !exists || !value { + continue + } + if setting.CLIFlagValue != "" { + cmd = append(cmd, setting.CLIFlag, setting.CLIFlagValue) + } else { + parts := strings.Fields(setting.CLIFlag) + cmd = append(cmd, parts...) + } + } + } + + // Add resume flag if agent supports it + if agentConfig.PassthroughConfig.ResumeFlag != "" { + // Split on spaces to handle flags like "--resume latest" + parts := strings.Fields(agentConfig.PassthroughConfig.ResumeFlag) + cmd = append(cmd, parts...) + } + + return cmd +} + // handlePassthroughTurnComplete is called when turn detection fires for a passthrough session. // This marks the execution as ready for follow-up prompts when the agent finishes processing. func (m *Manager) handlePassthroughTurnComplete(sessionID string) { diff --git a/apps/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go index 0a3bbf845f..048c989ba3 100644 --- a/apps/backend/internal/agent/lifecycle/manager_test.go +++ b/apps/backend/internal/agent/lifecycle/manager_test.go @@ -394,3 +394,211 @@ func TestManager_StartStop(t *testing.T) { t.Fatalf("unexpected error stopping manager: %v", err) } } + +func TestManager_BuildPassthroughResumeCommand(t *testing.T) { + mgr := newTestManager() + + tests := []struct { + name string + agentConfig *registry.AgentTypeConfig + profileInfo *AgentProfileInfo + wantCmd []string + }{ + { + name: "basic command without profile", + agentConfig: ®istry.AgentTypeConfig{ + ID: "test-agent", + PassthroughConfig: registry.PassthroughConfig{ + Supported: true, + PassthroughCmd: []string{"test-cli", "--verbose"}, + }, + }, + profileInfo: nil, + wantCmd: []string{"test-cli", "--verbose"}, + }, + { + name: "command with model flag", + agentConfig: ®istry.AgentTypeConfig{ + ID: "test-agent", + PassthroughConfig: registry.PassthroughConfig{ + Supported: true, + PassthroughCmd: []string{"test-cli"}, + ModelFlag: "--model {model}", + }, + }, + profileInfo: &AgentProfileInfo{ + Model: "gpt-4", + }, + wantCmd: []string{"test-cli", "--model", "gpt-4"}, + }, + { + name: "command with resume flag", + agentConfig: ®istry.AgentTypeConfig{ + ID: "test-agent", + PassthroughConfig: registry.PassthroughConfig{ + Supported: true, + PassthroughCmd: []string{"test-cli"}, + ResumeFlag: "-c", + }, + }, + profileInfo: nil, + wantCmd: []string{"test-cli", "-c"}, + }, + { + name: "command with multi-word resume flag", + agentConfig: ®istry.AgentTypeConfig{ + ID: "gemini-agent", + PassthroughConfig: registry.PassthroughConfig{ + Supported: true, + PassthroughCmd: []string{"gemini"}, + ResumeFlag: "--resume latest", + }, + }, + profileInfo: nil, + wantCmd: []string{"gemini", "--resume", "latest"}, + }, + { + name: "command with permission settings", + agentConfig: ®istry.AgentTypeConfig{ + ID: "test-agent", + PassthroughConfig: registry.PassthroughConfig{ + Supported: true, + PassthroughCmd: []string{"test-cli"}, + }, + PermissionSettings: map[string]registry.PermissionSetting{ + "dangerously_skip_permissions": { + Supported: true, + ApplyMethod: "cli_flag", + CLIFlag: "--dangerous", + }, + "auto_approve": { + Supported: true, + ApplyMethod: "cli_flag", + CLIFlag: "--yes", + }, + }, + }, + profileInfo: &AgentProfileInfo{ + DangerouslySkipPermissions: true, + AutoApprove: true, + }, + wantCmd: []string{"test-cli", "--dangerous", "--yes"}, + }, + { + name: "full command with all options", + agentConfig: ®istry.AgentTypeConfig{ + ID: "claude-code", + PassthroughConfig: registry.PassthroughConfig{ + Supported: true, + PassthroughCmd: []string{"npx", "-y", "@anthropic-ai/claude-code"}, + ModelFlag: "--model {model}", + ResumeFlag: "-c", + }, + PermissionSettings: map[string]registry.PermissionSetting{ + "dangerously_skip_permissions": { + Supported: true, + ApplyMethod: "cli_flag", + CLIFlag: "--dangerously-skip-permissions", + }, + }, + }, + profileInfo: &AgentProfileInfo{ + Model: "claude-sonnet-4", + DangerouslySkipPermissions: true, + }, + wantCmd: []string{"npx", "-y", "@anthropic-ai/claude-code", "--model", "claude-sonnet-4", "--dangerously-skip-permissions", "-c"}, + }, + { + name: "permission setting with cli_flag_value", + agentConfig: ®istry.AgentTypeConfig{ + ID: "test-agent", + PassthroughConfig: registry.PassthroughConfig{ + Supported: true, + PassthroughCmd: []string{"test-cli"}, + }, + PermissionSettings: map[string]registry.PermissionSetting{ + "auto_approve": { + Supported: true, + ApplyMethod: "cli_flag", + CLIFlag: "--approve-level", + CLIFlagValue: "all", + }, + }, + }, + profileInfo: &AgentProfileInfo{ + AutoApprove: true, + }, + wantCmd: []string{"test-cli", "--approve-level", "all"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := mgr.buildPassthroughResumeCommand(tt.agentConfig, tt.profileInfo) + + if len(got) != len(tt.wantCmd) { + t.Errorf("buildPassthroughResumeCommand() = %v, want %v", got, tt.wantCmd) + return + } + + for i, arg := range got { + if arg != tt.wantCmd[i] { + t.Errorf("buildPassthroughResumeCommand()[%d] = %q, want %q", i, arg, tt.wantCmd[i]) + } + } + }) + } +} + +func TestManager_VerifyPassthroughEnabled(t *testing.T) { + tests := []struct { + name string + profileID string + wantErr bool + }{ + { + name: "valid profile with passthrough enabled", + profileID: "test-profile", + wantErr: false, + }, + { + name: "empty profile ID", + profileID: "", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mgr := newTestManager() + + // Override profile resolver for this test + if tt.profileID != "" { + mgr.profileResolver = &mockPassthroughProfileResolver{ + cliPassthrough: true, + } + } + + err := mgr.verifyPassthroughEnabled(context.Background(), "test-session", tt.profileID) + if (err != nil) != tt.wantErr { + t.Errorf("verifyPassthroughEnabled() error = %v, wantErr %v", err, tt.wantErr) + } + }) + } +} + +// mockPassthroughProfileResolver is a mock for testing passthrough verification +type mockPassthroughProfileResolver struct { + cliPassthrough bool + err error +} + +func (m *mockPassthroughProfileResolver) ResolveProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) { + if m.err != nil { + return nil, m.err + } + return &AgentProfileInfo{ + ProfileID: profileID, + CLIPassthrough: m.cliPassthrough, + }, nil +} diff --git a/apps/backend/internal/agent/registry/agents.json b/apps/backend/internal/agent/registry/agents.json index d6c1a7691f..c10c4d661b 100644 --- a/apps/backend/internal/agent/registry/agents.json +++ b/apps/backend/internal/agent/registry/agents.json @@ -27,7 +27,8 @@ ], "model_flag": "--model {model}", "idle_timeout_ms": 3000, - "buffer_max_bytes": 2097152 + "buffer_max_bytes": 2097152, + "resume_flag": "-c" }, "working_dir": "/workspace", "required_env": [ @@ -189,7 +190,8 @@ ], "model_flag": "--model {model}", "idle_timeout_ms": 3000, - "buffer_max_bytes": 2097152 + "buffer_max_bytes": 2097152, + "resume_flag": "" }, "working_dir": "/workspace", "required_env": [], @@ -336,7 +338,8 @@ ], "model_flag": "--model {model}", "idle_timeout_ms": 3000, - "buffer_max_bytes": 2097152 + "buffer_max_bytes": 2097152, + "resume_flag": "-c" }, "working_dir": "{workspace}", "required_env": [ @@ -511,7 +514,8 @@ "model_flag": "--model {model}", "prompt_flag": "--prompt-interactive {prompt}", "idle_timeout_ms": 3000, - "buffer_max_bytes": 2097152 + "buffer_max_bytes": 2097152, + "resume_flag": "--resume latest" }, "permission_settings": { "auto_approve": { @@ -601,7 +605,8 @@ "model_flag": "--model {model}", "prompt_flag": "--prompt {prompt}", "idle_timeout_ms": 3000, - "buffer_max_bytes": 2097152 + "buffer_max_bytes": 2097152, + "resume_flag": "-c" }, "working_dir": "{workspace}", "required_env": [], diff --git a/apps/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go index 486a67f3eb..4f1daf9606 100644 --- a/apps/backend/internal/agent/registry/registry.go +++ b/apps/backend/internal/agent/registry/registry.go @@ -245,6 +245,19 @@ type PassthroughConfig struct { // StabilityWindowMs is how long state must be stable before reporting (default 0) // Codex uses 1000ms to prevent false exits from working state StabilityWindowMs int `json:"stability_window_ms,omitempty"` + + // ResumeFlag is the CLI flag to continue the last conversation in the workspace. + // This is used when the backend restarts and user reconnects to an existing session. + // Examples: "-c" for auggie/claude-code/opencode, "--resume latest" for gemini + // If empty, the agent doesn't support resume and starts a fresh session. + ResumeFlag string `json:"resume_flag,omitempty"` + + // WaitForTerminal indicates that the agent requires a terminal to be connected + // before starting. Some agents (like Codex) query the terminal for cursor position + // on startup and will fail if no terminal is connected to respond. + // When true, the process waits for the terminal WebSocket to connect and send + // dimensions before starting. When false (default), starts immediately. + WaitForTerminal bool `json:"wait_for_terminal,omitempty"` } // Registry manages agent type configurations diff --git a/apps/backend/internal/agentctl/server/process/interactive_runner.go b/apps/backend/internal/agentctl/server/process/interactive_runner.go index c822b98c58..cced627de0 100644 --- a/apps/backend/internal/agentctl/server/process/interactive_runner.go +++ b/apps/backend/internal/agentctl/server/process/interactive_runner.go @@ -7,6 +7,7 @@ package process import ( + "bytes" "context" "fmt" "io" @@ -455,6 +456,53 @@ func (r *InteractiveRunner) GetBySession(sessionID string) (*InteractiveProcessI return nil, false } +// isProcessAlive checks if the underlying OS process is still running. +// Must be called with proc.mu held. +func (r *InteractiveRunner) isProcessAlive(proc *interactiveProcess) bool { + if proc.cmd != nil && proc.cmd.Process != nil { + // Signal 0 checks if process exists without sending a signal + err := proc.cmd.Process.Signal(syscall.Signal(0)) + return err == nil + } + return false +} + +// IsProcessRunning checks if a process with the given ID exists and is running. +// This is used to detect if a process was killed (e.g., after backend restart). +func (r *InteractiveRunner) IsProcessRunning(processID string) bool { + proc, ok := r.get(processID) + if !ok { + return false + } + + proc.mu.Lock() + defer proc.mu.Unlock() + + // Process must be started and alive + return proc.started && r.isProcessAlive(proc) +} + +// IsProcessReadyOrPending checks if a process exists and is either running or pending start. +// This is used by the terminal handler to allow connections to deferred-start processes +// that will start when the terminal sends dimensions. +func (r *InteractiveRunner) IsProcessReadyOrPending(processID string) bool { + proc, ok := r.get(processID) + if !ok { + return false + } + + proc.mu.Lock() + defer proc.mu.Unlock() + + // Process exists but hasn't started yet (deferred start) - this is OK + if !proc.started { + return true + } + + // Process started - check if still alive + return r.isProcessAlive(proc) +} + // GetBuffer returns the buffered output for a process. func (r *InteractiveRunner) GetBuffer(processID string) ([]ProcessOutputChunk, bool) { proc, ok := r.get(processID) @@ -566,6 +614,40 @@ func (r *InteractiveRunner) readOutput(proc *interactiveProcess) { data := buf[:n] dataStr := string(data) + // Respond to cursor position queries (DSR) if no terminal is connected yet. + // Some CLI tools (like Codex) query cursor position on startup with \e[6n + // and expect a response \e[row;colR. Without this, they timeout and exit. + // Only respond if no direct output writer is connected (no real terminal yet). + proc.directOutputMu.RLock() + hasDirectWriter := proc.directOutput != nil + proc.directOutputMu.RUnlock() + + if !hasDirectWriter { + // Check for cursor position query (DSR): ESC [ 6 n or ESC [ ? 6 n + if containsDSRQuery(data) { + // Respond with cursor at position 1,1 (top-left) + response := "\x1b[1;1R" + if _, err := ptyInstance.Write([]byte(response)); err != nil { + r.logger.Debug("failed to respond to cursor position query", + zap.String("process_id", proc.info.ID), + zap.Error(err)) + } else { + r.logger.Debug("responded to cursor position query", + zap.String("process_id", proc.info.ID)) + } + } + // Check for primary device attributes query (DA1): ESC [ c or ESC [ 0 c + if containsDA1Query(data) { + // Respond as VT100 terminal with advanced video option + response := "\x1b[?1;2c" + if _, err := ptyInstance.Write([]byte(response)); err != nil { + r.logger.Debug("failed to respond to device attributes query", + zap.String("process_id", proc.info.ID), + zap.Error(err)) + } + } + } + // Feed to status tracker for TUI state detection if proc.statusTracker != nil { proc.statusTracker.Write(data) @@ -723,6 +805,27 @@ func (r *InteractiveRunner) wait(proc *interactiveProcess) { zap.Error(err), ) + // Log buffer contents if process exited with error (helps debug startup failures) + if status == types.ProcessStatusFailed && proc.buffer != nil { + chunks := proc.buffer.snapshot() + if len(chunks) > 0 { + var combinedOutput string + for _, chunk := range chunks { + combinedOutput += chunk.Data + } + // Truncate for logging (max 2000 chars) + if len(combinedOutput) > 2000 { + combinedOutput = combinedOutput[:2000] + "...(truncated)" + } + r.logger.Error("interactive process output before exit", + zap.String("process_id", proc.info.ID), + zap.String("session_id", proc.info.SessionID), + zap.Int("exit_code", exitCode), + zap.String("output", combinedOutput), + ) + } + } + // Stop idle timer proc.idleTimerMu.Lock() if proc.idleTimer != nil { @@ -878,3 +981,33 @@ func (r *InteractiveRunner) GetPtyWriter(processID string) (io.Writer, error) { return proc.ptmx, nil } +// containsDSRQuery checks if data contains a Device Status Report (cursor position) query. +// DSR query is: ESC [ 6 n or ESC [ ? 6 n +func containsDSRQuery(data []byte) bool { + return bytes.Contains(data, []byte("\x1b[6n")) || bytes.Contains(data, []byte("\x1b[?6n")) +} + +// containsDA1Query checks if data contains a Primary Device Attributes query. +// DA1 query is: ESC [ c or ESC [ 0 c (but NOT ESC [ c which is cursor forward) +func containsDA1Query(data []byte) bool { + // Check for exact ESC [ c sequence + csiC := []byte("\x1b[c") + for i := 0; i <= len(data)-len(csiC); i++ { + if data[i] == '\x1b' && i+2 < len(data) && data[i+1] == '[' && data[i+2] == 'c' { + // Make sure it's not preceded by a digit (which would make it cursor forward) + // ESC [ c is valid, ESC [ 0 c is valid, but ESC [ 1 c is cursor forward + if i+2 == len(data)-1 { + // ESC [ c at end + return true + } + // Check what's before 'c' - if it's just '[' or '[0' it's DA1 + // We already matched ESC [ c, so this is valid + return true + } + } + + // Check for ESC [ 0 c + csi0C := []byte("\x1b[0c") + return bytes.Contains(data, csi0C) +} + diff --git a/apps/backend/internal/agentctl/server/process/interactive_runner_test.go b/apps/backend/internal/agentctl/server/process/interactive_runner_test.go index fbb6934a95..c05443bf05 100644 --- a/apps/backend/internal/agentctl/server/process/interactive_runner_test.go +++ b/apps/backend/internal/agentctl/server/process/interactive_runner_test.go @@ -495,3 +495,195 @@ func TestInteractiveRunner_NotFound(t *testing.T) { t.Error("GetPtyWriter() should fail for nonexistent process") } } + +func TestInteractiveRunner_IsProcessRunning(t *testing.T) { + log := newTestLogger(t) + runner := NewInteractiveRunner(nil, log, 2*1024*1024) + + // Non-existent process + if runner.IsProcessRunning("nonexistent") { + t.Error("IsProcessRunning() should return false for nonexistent process") + } + + // Start a process + req := InteractiveStartRequest{ + SessionID: "running-test", + Command: []string{"sleep", "10"}, + ImmediateStart: true, + DefaultCols: 80, + DefaultRows: 24, + } + + info, err := runner.Start(context.Background(), req) + if err != nil { + t.Fatalf("Start() error = %v", err) + } + + // Give process time to start + time.Sleep(100 * time.Millisecond) + + // Process should be running + if !runner.IsProcessRunning(info.ID) { + t.Error("IsProcessRunning() should return true for running process") + } + + // Stop the process + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = runner.Stop(ctx, info.ID) + + // Give it time to clean up + time.Sleep(200 * time.Millisecond) + + // Process should no longer be running + if runner.IsProcessRunning(info.ID) { + t.Error("IsProcessRunning() should return false after stop") + } +} + +func TestInteractiveRunner_IsProcessReadyOrPending(t *testing.T) { + log := newTestLogger(t) + runner := NewInteractiveRunner(nil, log, 2*1024*1024) + + // Non-existent process + if runner.IsProcessReadyOrPending("nonexistent") { + t.Error("IsProcessReadyOrPending() should return false for nonexistent process") + } + + // Start a deferred process (not started yet) + req := InteractiveStartRequest{ + SessionID: "pending-test", + Command: []string{"cat"}, + // ImmediateStart: false (deferred) + } + + info, err := runner.Start(context.Background(), req) + if err != nil { + t.Fatalf("Start() error = %v", err) + } + + // Deferred process should be "ready or pending" + if !runner.IsProcessReadyOrPending(info.ID) { + t.Error("IsProcessReadyOrPending() should return true for pending process") + } + + // But not "running" yet + if runner.IsProcessRunning(info.ID) { + t.Error("IsProcessRunning() should return false for pending process") + } + + // Trigger start via resize + err = runner.ResizeBySession("pending-test", 80, 24) + if err != nil { + t.Fatalf("ResizeBySession() error = %v", err) + } + + // Give process time to start + time.Sleep(100 * time.Millisecond) + + // Now it should be both "running" and "ready or pending" + if !runner.IsProcessRunning(info.ID) { + t.Error("IsProcessRunning() should return true after start") + } + if !runner.IsProcessReadyOrPending(info.ID) { + t.Error("IsProcessReadyOrPending() should return true for running process") + } + + // Stop the process + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = runner.Stop(ctx, info.ID) +} + +func TestCursorPositionQueryDetection(t *testing.T) { + tests := []struct { + name string + data []byte + wantDSR bool // Device Status Report (cursor position query) + wantDA bool // Device Attributes query + }{ + { + name: "cursor position query ESC[6n", + data: []byte("\x1b[6n"), + wantDSR: true, + wantDA: false, + }, + { + name: "cursor position query ESC[?6n", + data: []byte("\x1b[?6n"), + wantDSR: true, + wantDA: false, + }, + { + name: "device attributes query ESC[c", + data: []byte("\x1b[c"), + wantDSR: false, + wantDA: true, + }, + { + name: "device attributes query ESC[0c", + data: []byte("\x1b[0c"), + wantDSR: false, + wantDA: true, + }, + { + name: "no escape sequence", + data: []byte("hello world"), + wantDSR: false, + wantDA: false, + }, + { + name: "mixed content with DSR", + data: []byte("some text\x1b[6nmore text"), + wantDSR: true, + wantDA: false, + }, + { + name: "both DSR and DA", + data: []byte("\x1b[6n\x1b[c"), + wantDSR: true, + wantDA: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotDSR := containsCursorPositionQuery(tt.data) + gotDA := containsDeviceAttributesQuery(tt.data) + + if gotDSR != tt.wantDSR { + t.Errorf("containsCursorPositionQuery() = %v, want %v", gotDSR, tt.wantDSR) + } + if gotDA != tt.wantDA { + t.Errorf("containsDeviceAttributesQuery() = %v, want %v", gotDA, tt.wantDA) + } + }) + } +} + +// containsCursorPositionQuery checks for DSR cursor position query (CSI 6 n) +func containsCursorPositionQuery(data []byte) bool { + return bytesContains(data, []byte("\x1b[6n")) || bytesContains(data, []byte("\x1b[?6n")) +} + +// containsDeviceAttributesQuery checks for DA query (CSI c or CSI 0 c) +func containsDeviceAttributesQuery(data []byte) bool { + return bytesContains(data, []byte("\x1b[c")) || bytesContains(data, []byte("\x1b[0c")) +} + +// bytesContains is a simple contains check (bytes.Contains equivalent for tests) +func bytesContains(data, substr []byte) bool { + for i := 0; i <= len(data)-len(substr); i++ { + match := true + for j := 0; j < len(substr); j++ { + if data[i+j] != substr[j] { + match = false + break + } + } + if match { + return true + } + } + return false +} diff --git a/apps/backend/internal/gateway/websocket/terminal_handler.go b/apps/backend/internal/gateway/websocket/terminal_handler.go index 36c2803cba..830f4c8455 100644 --- a/apps/backend/internal/gateway/websocket/terminal_handler.go +++ b/apps/backend/internal/gateway/websocket/terminal_handler.go @@ -2,6 +2,7 @@ package websocket import ( + "bytes" "encoding/json" "io" "net/http" @@ -140,23 +141,7 @@ func (h *TerminalHandler) HandleTerminalWS(c *gin.Context) { zap.String("session_id", sessionID), zap.String("remote_addr", c.Request.RemoteAddr)) - // Validate session exists and is in passthrough mode - execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID) - if !exists { - h.logger.Warn("no execution found for session", - zap.String("session_id", sessionID)) - c.JSON(http.StatusNotFound, gin.H{"error": "no agent execution for session"}) - return - } - - if execution.PassthroughProcessID == "" { - h.logger.Warn("session is not in passthrough mode", - zap.String("session_id", sessionID)) - c.JSON(http.StatusBadRequest, gin.H{"error": "session is not in passthrough mode"}) - return - } - - // Get the interactive runner + // Get the interactive runner first interactiveRunner := h.lifecycleMgr.GetInteractiveRunner() if interactiveRunner == nil { h.logger.Error("interactive runner not available") @@ -164,9 +149,38 @@ func (h *TerminalHandler) HandleTerminalWS(c *gin.Context) { return } - processID := execution.PassthroughProcessID + // Ensure passthrough execution exists and is running. + // This handles: + // 1. Normal case: execution exists with running process + // 2. Backend restart: no execution, need to create and start + // 3. Process died: execution exists but process not running, need to restart + execution, err := h.lifecycleMgr.EnsurePassthroughExecution(c.Request.Context(), sessionID) + if err != nil { + h.logger.Warn("failed to ensure passthrough execution", + zap.String("session_id", sessionID), + zap.Error(err)) + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } - // Upgrade to WebSocket first - we'll get PTY access after the first resize + processID := execution.PassthroughProcessID + if processID == "" { + h.logger.Error("passthrough process ID is empty after ensure", + zap.String("session_id", sessionID)) + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "passthrough process not started"}) + return + } + + // Verify the process exists (either running or pending start for deferred-start processes) + if !interactiveRunner.IsProcessReadyOrPending(processID) { + h.logger.Error("passthrough process not found or exited", + zap.String("session_id", sessionID), + zap.String("process_id", processID)) + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "passthrough process failed to start"}) + return + } + + // Upgrade to WebSocket - we'll get PTY access after the first resize // triggers the lazy process start conn, err := terminalUpgrader.Upgrade(c.Writer, c.Request, nil) if err != nil { @@ -385,6 +399,29 @@ func (h *TerminalHandler) setupPtyAccess( return false } + // Send buffered output to the terminal before switching to direct mode. + // This provides scrollback history when reconnecting to an existing session. + // We batch all chunks into a single write for efficiency. + if chunks, ok := interactiveRunner.GetBuffer(processID); ok && len(chunks) > 0 { + var combined bytes.Buffer + for _, chunk := range chunks { + combined.WriteString(chunk.Data) + } + + if combined.Len() > 0 { + h.logger.Debug("sending buffered output to terminal", + zap.String("session_id", sessionID), + zap.Int("chunks", len(chunks)), + zap.Int("total_bytes", combined.Len())) + + if _, writeErr := wsw.Write(combined.Bytes()); writeErr != nil { + h.logger.Warn("failed to send buffered output", + zap.String("session_id", sessionID), + zap.Error(writeErr)) + } + } + } + // Set up direct output if err := interactiveRunner.SetDirectOutput(processID, wsw); err != nil { h.logger.Error("failed to set direct output", diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 92c76b0482..1fa492d906 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -13,12 +13,13 @@ import ( v1 "github.com/kandev/kandev/pkg/api/v1" "go.uber.org/zap" - "github.com/kandev/kandev/internal/worktree" + "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" + "github.com/kandev/kandev/internal/worktree" ) // WorktreeCleanup provides worktree cleanup on task deletion. @@ -2197,3 +2198,59 @@ func (s *Service) GetCumulativeDiff(ctx context.Context, sessionID string) (*mod Files: latestSnapshot.Files, }, nil } + +// GetWorkspaceInfoForSession returns workspace information for a task session. +// This implements the lifecycle.WorkspaceInfoProvider interface. +// The taskID parameter is optional - if empty, it will be looked up from the session. +func (s *Service) GetWorkspaceInfoForSession(ctx context.Context, taskID, sessionID string) (*lifecycle.WorkspaceInfo, error) { + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + return nil, fmt.Errorf("failed to get session %s: %w", sessionID, err) + } + if session == nil { + return nil, fmt.Errorf("session %s not found", sessionID) + } + + // Use session's TaskID if not provided + if taskID == "" { + taskID = session.TaskID + } + + // Get workspace path from the session's worktree + var workspacePath string + if len(session.Worktrees) > 0 { + workspacePath = session.Worktrees[0].WorktreePath + } + + // If no worktree, try to get from repository snapshot + if workspacePath == "" && session.RepositorySnapshot != nil { + if path, ok := session.RepositorySnapshot["path"].(string); ok { + workspacePath = path + } + } + + // Get agent ID from profile snapshot + var agentID string + if session.AgentProfileSnapshot != nil { + if id, ok := session.AgentProfileSnapshot["agent_id"].(string); ok { + agentID = id + } + } + + // Get ACP session ID from metadata + var acpSessionID string + if session.Metadata != nil { + if id, ok := session.Metadata["acp_session_id"].(string); ok { + acpSessionID = id + } + } + + return &lifecycle.WorkspaceInfo{ + TaskID: taskID, + SessionID: sessionID, + WorkspacePath: workspacePath, + AgentProfileID: session.AgentProfileID, + AgentID: agentID, + ACPSessionID: acpSessionID, + }, nil +} diff --git a/apps/web/components/task/task-center-panel.tsx b/apps/web/components/task/task-center-panel.tsx index c8756a03bd..816dae60cd 100644 --- a/apps/web/components/task/task-center-panel.tsx +++ b/apps/web/components/task/task-center-panel.tsx @@ -143,7 +143,7 @@ export const TaskCenterPanel = memo(function TaskCenterPanel({ {activeTaskId ? ( isPassthroughMode ? (
- +
) : ( diff --git a/apps/web/components/task/task-page-content.tsx b/apps/web/components/task/task-page-content.tsx index 3d6b45e743..5d45948a63 100644 --- a/apps/web/components/task/task-page-content.tsx +++ b/apps/web/components/task/task-page-content.tsx @@ -128,42 +128,23 @@ export function TaskPageContent({ }, []); useEffect(() => { - console.log('[TaskPageContent] Active session setup effect', { - initialTaskId: initialTask?.id, - activeTaskId, - initialSessionId, - sessionIdProp: sessionId, - taskSessionId, - }); if (!initialTask?.id) return; - // Always sync active task/session when SSR provides data. - // This ensures navigation to a new task/session updates the store. + // Sync SSR data to store on mount or when SSR props change. + // Don't include activeTaskId in deps - we only want to sync FROM SSR, not react to store changes. if (initialSessionId) { - console.log('[TaskPageContent] Setting active session', initialTask.id, initialSessionId); setActiveSession(initialTask.id, initialSessionId); - } else if (activeTaskId !== initialTask.id) { - console.log('[TaskPageContent] Setting active task', initialTask.id); + } else { setActiveTask(initialTask.id); } - }, [activeTaskId, initialSessionId, initialTask?.id, sessionId, taskSessionId, setActiveSession, setActiveTask]); + }, [initialTask?.id, initialSessionId, setActiveSession, setActiveTask]); useEffect(() => { - console.log('[TaskPageContent] Task fetch effect', { - activeTaskId, - initialTaskId: initialTask?.id, - taskDetailsId: taskDetails?.id, - shouldFetch: activeTaskId && taskDetails?.id !== activeTaskId, - }); if (!activeTaskId) return; if (taskDetails?.id === activeTaskId) return; - console.log('[TaskPageContent] Fetching task details for', activeTaskId); fetchTask(activeTaskId, { cache: 'no-store' }) - .then((response) => { - console.log('[TaskPageContent] Fetched task details', response?.id); - setTaskDetails(response); - }) + .then((response) => setTaskDetails(response)) .catch((error) => console.error('[TaskPageContent] Failed to load task details:', error)); - }, [activeTaskId, initialTask?.id, taskDetails?.id]); + }, [activeTaskId, taskDetails?.id]); const { repositories } = useRepositories(task?.workspace_id ?? null, Boolean(task?.workspace_id)); const effectiveRepositories = repositories.length ? repositories : initialRepositories; const repository = useMemo( diff --git a/apps/web/components/task/task-right-panel.tsx b/apps/web/components/task/task-right-panel.tsx index a41f64fa21..6c9eb6e0ba 100644 --- a/apps/web/components/task/task-right-panel.tsx +++ b/apps/web/components/task/task-right-panel.tsx @@ -281,7 +281,7 @@ const TaskRightPanel = memo(function TaskRightPanel({ topPanel, sessionId = null isStopping={isStoppingDev} /> ) : ( - + )}