From 9463ddac9f2eebe4bea8523d969061a77d4f4512 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Mon, 26 Jan 2026 23:52:28 +0000 Subject: [PATCH] feat: remove npm warns from terminal + fix terminal render on refresh (#104) --- .../internal/agent/lifecycle/manager.go | 88 +++++++ .../server/process/interactive_runner.go | 230 +++++++++++++++++- .../agentctl/server/process/runner.go | 28 ++- .../agentctl/server/process/runner_test.go | 77 ++++++ .../gateway/websocket/terminal_handler.go | 153 ++++++------ 5 files changed, 495 insertions(+), 81 deletions(-) diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index a36475fde9..b7ca0e8eec 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -2576,6 +2576,7 @@ func (m *Manager) handlePassthroughOutput(output *agentctltypes.ProcessOutput) { // handlePassthroughStatus handles status updates from a passthrough process and publishes to the event bus. // This is called when running in standalone mode without a WorkspaceTracker. +// When the process exits while a WebSocket is connected, it attempts auto-restart with rate limiting. func (m *Manager) handlePassthroughStatus(status *agentctltypes.ProcessStatusUpdate) { if status == nil { return @@ -2602,6 +2603,93 @@ func (m *Manager) handlePassthroughStatus(status *agentctltypes.ProcessStatusUpd } m.eventPublisher.PublishProcessStatus(execution, clientStatus) + + // Check if process exited and should be auto-restarted + // Run asynchronously to allow the old process to be cleaned up first + if status.Status == agentctltypes.ProcessStatusExited || status.Status == agentctltypes.ProcessStatusFailed { + go m.handlePassthroughExit(execution, status) + } +} + +// handlePassthroughExit handles auto-restart logic when a passthrough process exits. +// This function is called asynchronously to allow the old process to be cleaned up first. +func (m *Manager) handlePassthroughExit(execution *AgentExecution, status *agentctltypes.ProcessStatusUpdate) { + const restartDelay = 500 * time.Millisecond + const cleanupDelay = 100 * time.Millisecond // Wait for old process cleanup + + // Wait a bit for the old process to be cleaned up from the process map + time.Sleep(cleanupDelay) + + sessionID := execution.SessionID + + interactiveRunner := m.GetInteractiveRunner() + if interactiveRunner == nil { + m.logger.Debug("no interactive runner available for auto-restart", + zap.String("session_id", sessionID)) + return + } + + // Check if WebSocket is still connected (use session-level tracking which survives process deletion) + if !interactiveRunner.HasActiveWebSocketBySession(sessionID) { + m.logger.Debug("no active WebSocket, skipping auto-restart", + zap.String("session_id", sessionID)) + return + } + + exitCode := 0 + if status.ExitCode != nil { + exitCode = *status.ExitCode + } + + m.logger.Info("passthrough process exited with active WebSocket, attempting auto-restart", + zap.String("session_id", sessionID), + zap.Int("exit_code", exitCode)) + + // Send restart notification to terminal (use session-level to survive process deletion) + restartMsg := "\r\n\x1b[33m[Agent exited. Restarting...]\x1b[0m\r\n" + if err := interactiveRunner.WriteToDirectOutputBySession(sessionID, []byte(restartMsg)); err != nil { + m.logger.Debug("failed to write restart message to terminal", + zap.String("session_id", sessionID), + zap.Error(err)) + } + + // Delay before restart + time.Sleep(restartDelay) + + // Check WebSocket is still connected after delay (use session-level tracking) + if !interactiveRunner.HasActiveWebSocketBySession(sessionID) { + m.logger.Debug("WebSocket disconnected during restart delay, aborting", + zap.String("session_id", sessionID)) + return + } + + // Attempt restart + ctx := context.Background() + if err := m.ResumePassthroughSession(ctx, sessionID); err != nil { + m.logger.Error("failed to auto-restart passthrough session", + zap.String("session_id", sessionID), + zap.Error(err)) + + // Send error message to terminal + errorMsg := fmt.Sprintf("\r\n\x1b[31m[Restart failed: %s]\x1b[0m\r\n", err.Error()) + if writeErr := interactiveRunner.WriteToDirectOutputBySession(sessionID, []byte(errorMsg)); writeErr != nil { + m.logger.Debug("failed to write restart error message to terminal", + zap.String("session_id", sessionID), + zap.Error(writeErr)) + } + return + } + + // Connect the session's existing WebSocket to the new process + if interactiveRunner.ConnectSessionWebSocket(execution.PassthroughProcessID) { + m.logger.Info("passthrough session auto-restarted and reconnected WebSocket", + zap.String("session_id", sessionID), + zap.String("new_process_id", execution.PassthroughProcessID)) + } else { + m.logger.Warn("passthrough session restarted but failed to reconnect WebSocket", + zap.String("session_id", sessionID), + zap.String("new_process_id", execution.PassthroughProcessID)) + } } // GetInteractiveRunner returns the interactive runner for passthrough mode. diff --git a/apps/backend/internal/agentctl/server/process/interactive_runner.go b/apps/backend/internal/agentctl/server/process/interactive_runner.go index cced627de0..52c51eb960 100644 --- a/apps/backend/internal/agentctl/server/process/interactive_runner.go +++ b/apps/backend/internal/agentctl/server/process/interactive_runner.go @@ -92,6 +92,9 @@ type interactiveProcess struct { directOutput DirectOutputWriter directOutputMu sync.RWMutex + // WebSocket tracking - tracks whether a WebSocket is actively connected + hasActiveWebSocket bool + // Lifecycle stopOnce sync.Once stopSignal chan struct{} @@ -112,6 +115,13 @@ type StatusCallback func(status *types.ProcessStatusUpdate) // AgentStateCallback is called when agent TUI state changes (working, waiting, etc.). type AgentStateCallback func(sessionID string, state AgentState) +// sessionWebSocket tracks a WebSocket connection at the session level. +// This allows the WebSocket to survive process restarts. +type sessionWebSocket struct { + writer DirectOutputWriter + mu sync.RWMutex +} + // InteractiveRunner manages interactive PTY-based processes with stdin support. type InteractiveRunner struct { logger *logger.Logger @@ -124,6 +134,10 @@ type InteractiveRunner struct { mu sync.RWMutex processes map[string]*interactiveProcess + + // Session-level WebSocket tracking - survives process restarts + sessionWsMu sync.RWMutex + sessionWs map[string]*sessionWebSocket } // NewInteractiveRunner creates a new interactive process runner. @@ -133,6 +147,7 @@ func NewInteractiveRunner(workspaceTracker *WorkspaceTracker, log *logger.Logger workspaceTracker: workspaceTracker, bufferMaxBytes: bufferMaxBytes, processes: make(map[string]*interactiveProcess), + sessionWs: make(map[string]*sessionWebSocket), } } @@ -662,13 +677,21 @@ func (r *InteractiveRunner) readOutput(proc *interactiveProcess) { } } + // Always buffer output for scrollback restoration on reconnect + chunk := ProcessOutputChunk{ + Stream: "stdout", + Data: dataStr, + Timestamp: time.Now().UTC(), + } + proc.buffer.append(chunk) + // Check if we have a direct output writer (binary WebSocket mode) proc.directOutputMu.RLock() directWriter := proc.directOutput proc.directOutputMu.RUnlock() if directWriter != nil { - // Direct mode: write raw bytes to the WebSocket, skip event bus + // Direct mode: write raw bytes to the WebSocket if _, writeErr := directWriter.Write(data); writeErr != nil { r.logger.Debug("direct output write error", zap.String("process_id", proc.info.ID), @@ -676,13 +699,7 @@ func (r *InteractiveRunner) readOutput(proc *interactiveProcess) { // Don't return - the writer might have closed but process continues } } else { - // Normal mode: buffer and stream output via event bus - chunk := ProcessOutputChunk{ - Stream: "stdout", - Data: dataStr, - Timestamp: time.Now().UTC(), - } - proc.buffer.append(chunk) + // No WebSocket connected: also publish via event bus r.publishOutput(proc, chunk) } @@ -922,6 +939,7 @@ func (p *interactiveProcess) snapshot(includeOutput bool) InteractiveProcessInfo // SetDirectOutput sets a direct output writer for a process. // When set, PTY output bypasses the event bus and goes directly to this writer. // This is used for the dedicated binary WebSocket in passthrough mode. +// Also tracks the WebSocket at the session level to survive process restarts. // Returns error if process not found. func (r *InteractiveRunner) SetDirectOutput(processID string, writer DirectOutputWriter) error { proc, ok := r.get(processID) @@ -929,36 +947,119 @@ func (r *InteractiveRunner) SetDirectOutput(processID string, writer DirectOutpu return fmt.Errorf("process not found: %s", processID) } + sessionID := proc.info.SessionID + proc.directOutputMu.Lock() proc.directOutput = writer + proc.hasActiveWebSocket = true proc.directOutputMu.Unlock() + // Also track at session level for restart support + r.sessionWsMu.Lock() + r.sessionWs[sessionID] = &sessionWebSocket{writer: writer} + r.sessionWsMu.Unlock() + r.logger.Info("direct output set for process", zap.String("process_id", processID), - zap.String("session_id", proc.info.SessionID)) + zap.String("session_id", sessionID)) return nil } // ClearDirectOutput clears the direct output writer for a process. // Output will return to the normal event bus path. +// Also clears the session-level WebSocket tracking. func (r *InteractiveRunner) ClearDirectOutput(processID string) error { proc, ok := r.get(processID) if !ok { - return fmt.Errorf("process not found: %s", processID) + // Process may have been deleted - still try to clear session tracking + r.logger.Debug("process not found during ClearDirectOutput, trying to clear by session", + zap.String("process_id", processID)) + return nil } + sessionID := proc.info.SessionID + proc.directOutputMu.Lock() proc.directOutput = nil + proc.hasActiveWebSocket = false proc.directOutputMu.Unlock() + // Also clear session-level tracking + r.sessionWsMu.Lock() + delete(r.sessionWs, sessionID) + r.sessionWsMu.Unlock() + r.logger.Info("direct output cleared for process", zap.String("process_id", processID), - zap.String("session_id", proc.info.SessionID)) + zap.String("session_id", sessionID)) return nil } +// ClearDirectOutputBySession clears the direct output for a session. +// This is used when the terminal WebSocket disconnects. +func (r *InteractiveRunner) ClearDirectOutputBySession(sessionID string) { + // Clear session-level tracking + r.sessionWsMu.Lock() + delete(r.sessionWs, sessionID) + r.sessionWsMu.Unlock() + + // Also clear from any process with this session + r.mu.RLock() + for _, proc := range r.processes { + if proc.info.SessionID == sessionID { + proc.directOutputMu.Lock() + proc.directOutput = nil + proc.hasActiveWebSocket = false + proc.directOutputMu.Unlock() + } + } + r.mu.RUnlock() + + r.logger.Info("direct output cleared for session", + zap.String("session_id", sessionID)) +} + +// ConnectSessionWebSocket connects an existing session WebSocket to a process. +// This is called when a new process starts for a session that already has an active WebSocket. +// Returns true if a WebSocket was connected. +func (r *InteractiveRunner) ConnectSessionWebSocket(processID string) bool { + proc, ok := r.get(processID) + if !ok { + return false + } + + sessionID := proc.info.SessionID + + r.sessionWsMu.RLock() + sessWs, exists := r.sessionWs[sessionID] + r.sessionWsMu.RUnlock() + + if !exists || sessWs == nil { + return false + } + + sessWs.mu.RLock() + writer := sessWs.writer + sessWs.mu.RUnlock() + + if writer == nil { + return false + } + + proc.directOutputMu.Lock() + proc.directOutput = writer + proc.hasActiveWebSocket = true + proc.directOutputMu.Unlock() + + r.logger.Info("connected existing session WebSocket to new process", + zap.String("process_id", processID), + zap.String("session_id", sessionID)) + + return true +} + // GetPtyWriter returns a writer for sending input to the PTY. // This is used for the dedicated binary WebSocket in passthrough mode. func (r *InteractiveRunner) GetPtyWriter(processID string) (io.Writer, error) { @@ -981,6 +1082,113 @@ func (r *InteractiveRunner) GetPtyWriter(processID string) (io.Writer, error) { return proc.ptmx, nil } +// GetPtyWriterBySession returns a writer for sending input to the PTY for a session. +// This is used to reconnect after a process restart. +func (r *InteractiveRunner) GetPtyWriterBySession(sessionID string) (io.Writer, string, error) { + r.mu.RLock() + var proc *interactiveProcess + for _, p := range r.processes { + if p.info.SessionID == sessionID { + proc = p + break + } + } + r.mu.RUnlock() + + if proc == nil { + return nil, "", fmt.Errorf("no process found for session: %s", sessionID) + } + + proc.mu.Lock() + defer proc.mu.Unlock() + + if !proc.started { + return nil, proc.info.ID, fmt.Errorf("process not started yet - waiting for terminal dimensions") + } + + if proc.ptmx == nil { + return nil, proc.info.ID, fmt.Errorf("PTY not available") + } + + return proc.ptmx, proc.info.ID, nil +} + +// HasActiveWebSocket checks if a process has an active WebSocket connection. +// This is used to determine if auto-restart should be attempted on process exit. +func (r *InteractiveRunner) HasActiveWebSocket(processID string) bool { + proc, ok := r.get(processID) + if !ok { + return false + } + + proc.directOutputMu.RLock() + defer proc.directOutputMu.RUnlock() + return proc.hasActiveWebSocket +} + +// HasActiveWebSocketBySession checks if a session has an active WebSocket connection. +// Uses session-level tracking which survives process restarts. +func (r *InteractiveRunner) HasActiveWebSocketBySession(sessionID string) bool { + r.sessionWsMu.RLock() + sessWs, exists := r.sessionWs[sessionID] + r.sessionWsMu.RUnlock() + + if !exists || sessWs == nil { + return false + } + + sessWs.mu.RLock() + hasWriter := sessWs.writer != nil + sessWs.mu.RUnlock() + + return hasWriter +} + +// WriteToDirectOutput writes data directly to the WebSocket output for a process. +// This is used to send messages like restart notifications to the terminal. +// Returns error if process not found or no direct output is set. +func (r *InteractiveRunner) WriteToDirectOutput(processID string, data []byte) error { + proc, ok := r.get(processID) + if !ok { + return fmt.Errorf("process not found: %s", processID) + } + + proc.directOutputMu.RLock() + directWriter := proc.directOutput + proc.directOutputMu.RUnlock() + + if directWriter == nil { + return fmt.Errorf("no direct output writer set for process: %s", processID) + } + + _, err := directWriter.Write(data) + return err +} + +// WriteToDirectOutputBySession writes data directly to the WebSocket output for a session. +// This is used to send messages like restart notifications to the terminal. +// Uses session-level tracking which survives process restarts. +func (r *InteractiveRunner) WriteToDirectOutputBySession(sessionID string, data []byte) error { + r.sessionWsMu.RLock() + sessWs, exists := r.sessionWs[sessionID] + r.sessionWsMu.RUnlock() + + if !exists || sessWs == nil { + return fmt.Errorf("no WebSocket found for session: %s", sessionID) + } + + sessWs.mu.RLock() + writer := sessWs.writer + sessWs.mu.RUnlock() + + if writer == nil { + return fmt.Errorf("no direct output writer set for session: %s", sessionID) + } + + _, err := writer.Write(data) + return err +} + // 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 { diff --git a/apps/backend/internal/agentctl/server/process/runner.go b/apps/backend/internal/agentctl/server/process/runner.go index 8cc15bfbad..71848d99d8 100644 --- a/apps/backend/internal/agentctl/server/process/runner.go +++ b/apps/backend/internal/agentctl/server/process/runner.go @@ -640,10 +640,16 @@ func (p *commandProcess) snapshot(includeOutput bool) ProcessInfo { func mergeEnv(env map[string]string) []string { base := make(map[string]string, len(os.Environ())+len(env)) - // Parse parent environment into map + // Parse parent environment into map, filtering out npm config variables + // that cause warnings when passed to npx commands for _, entry := range os.Environ() { if eq := strings.IndexByte(entry, '='); eq >= 0 { - base[entry[:eq]] = entry[eq+1:] + key := entry[:eq] + // Filter out npm-related environment variables that cause warnings + if isNpmEnvVar(key) { + continue + } + base[key] = entry[eq+1:] } } @@ -659,3 +665,21 @@ func mergeEnv(env map[string]string) []string { } return merged } + +// isNpmEnvVar returns true if the key is an npm-related environment variable +// that should be filtered out to prevent warnings in npx commands. +func isNpmEnvVar(key string) bool { + npmPrefixes := []string{ + "npm_config_", + "npm_package_", + "npm_lifecycle_", + "npm_execpath", + "npm_node_execpath", + } + for _, prefix := range npmPrefixes { + if strings.HasPrefix(key, prefix) { + return true + } + } + return false +} diff --git a/apps/backend/internal/agentctl/server/process/runner_test.go b/apps/backend/internal/agentctl/server/process/runner_test.go index 08dc286169..ba4e4b481e 100644 --- a/apps/backend/internal/agentctl/server/process/runner_test.go +++ b/apps/backend/internal/agentctl/server/process/runner_test.go @@ -149,3 +149,80 @@ func TestStripANSI(t *testing.T) { }) } } + +func TestIsNpmEnvVar(t *testing.T) { + tests := []struct { + key string + expected bool + }{ + // Should filter + {"npm_config_registry", true}, + {"npm_config_npm-globalconfig", true}, + {"npm_config__jsr-registry", true}, + {"npm_config_verify-deps-before-run", true}, + {"npm_config_dir", true}, + {"npm_package_name", true}, + {"npm_package_version", true}, + {"npm_lifecycle_event", true}, + {"npm_execpath", true}, + {"npm_node_execpath", true}, + + // Should keep + {"PATH", false}, + {"HOME", false}, + {"USER", false}, + {"SHELL", false}, + {"ANTHROPIC_API_KEY", false}, + {"NPM_TOKEN", false}, // Uppercase, different format + {"NPMRC", false}, // Not npm_ prefix + {"npm_not_a_config", false}, // Doesn't match any prefix + } + + for _, tt := range tests { + t.Run(tt.key, func(t *testing.T) { + result := isNpmEnvVar(tt.key) + if result != tt.expected { + t.Errorf("isNpmEnvVar(%q) = %v, want %v", tt.key, result, tt.expected) + } + }) + } +} + +func TestMergeEnv_FiltersNpmVars(t *testing.T) { + // Set some test npm env vars that should be filtered + t.Setenv("npm_config_test_var", "should_be_filtered") + t.Setenv("npm_package_name", "test-package") + + // Set a normal var that should be kept + t.Setenv("TEST_NORMAL_VAR", "should_be_kept") + + result := mergeEnv(map[string]string{ + "CUSTOM_VAR": "custom_value", + }) + + // Convert to map for easier checking + resultMap := make(map[string]string) + for _, entry := range result { + if eq := strings.IndexByte(entry, '='); eq >= 0 { + resultMap[entry[:eq]] = entry[eq+1:] + } + } + + // Check that npm vars are filtered + if _, exists := resultMap["npm_config_test_var"]; exists { + t.Error("npm_config_test_var should have been filtered") + } + if _, exists := resultMap["npm_package_name"]; exists { + t.Error("npm_package_name should have been filtered") + } + + // Check that normal vars are kept + if resultMap["TEST_NORMAL_VAR"] != "should_be_kept" { + t.Error("TEST_NORMAL_VAR should have been kept") + } + + // Check that custom vars are added + if resultMap["CUSTOM_VAR"] != "custom_value" { + t.Error("CUSTOM_VAR should have been added") + } +} diff --git a/apps/backend/internal/gateway/websocket/terminal_handler.go b/apps/backend/internal/gateway/websocket/terminal_handler.go index 830f4c8455..b5a67ecda8 100644 --- a/apps/backend/internal/gateway/websocket/terminal_handler.go +++ b/apps/backend/internal/gateway/websocket/terminal_handler.go @@ -212,7 +212,6 @@ func (h *TerminalHandler) runTerminalBridge( ) { var ptyWriter io.Writer var directOutputSet bool - var lastCols, lastRows uint16 // Channel to signal cleanup to any spawned goroutines done := make(chan struct{}) @@ -221,14 +220,8 @@ func (h *TerminalHandler) runTerminalBridge( // Signal any background goroutines to stop close(done) - // Clean up: clear direct output, close WebSocket writer, close connection - if directOutputSet { - if err := interactiveRunner.ClearDirectOutput(processID); err != nil { - h.logger.Debug("failed to clear direct output", - zap.String("session_id", sessionID), - zap.Error(err)) - } - } + // Clean up: clear direct output at session level (process may have been restarted with new ID) + interactiveRunner.ClearDirectOutputBySession(sessionID) _ = wsw.Close() _ = conn.Close() @@ -258,78 +251,52 @@ func (h *TerminalHandler) runTerminalBridge( // Check for resize command (first byte 0x01) if data[0] == resizeCommandByte { - cols, rows := h.handleResize(data[1:], sessionID, interactiveRunner) - if cols > 0 && rows > 0 { - lastCols, lastRows = cols, rows - } - - // After resize, try to set up PTY access if not already done + // Set up PTY access BEFORE handling resize if not already done. + // This ensures buffered output is sent to the terminal before the resize + // triggers a TUI redraw (which would overwrite the scrollback). if !directOutputSet { if h.setupPtyAccess(sessionID, processID, interactiveRunner, wsw, &ptyWriter) { directOutputSet = true - - // Trigger a resize to force TUI redraw after reconnect - // This ensures the TUI app redraws with full colors/styles - // We resize to different dimensions then back to force a real redraw - // (some TUI apps ignore same-size resizes) - if lastCols > 2 && lastRows > 2 { - // Delay to ensure direct output is fully set up - time.Sleep(50 * time.Millisecond) - - // Resize to smaller dimensions to trigger a real change - // Use a more significant change (2 cols/rows) for apps that might ignore small changes - if err := interactiveRunner.ResizeBySession(sessionID, lastCols-2, lastRows-2); err != nil { - h.logger.Debug("failed to trigger intermediate resize", - zap.String("session_id", sessionID), - zap.Error(err)) - } - - // Longer delay then resize back to actual dimensions - time.Sleep(100 * time.Millisecond) - if err := interactiveRunner.ResizeBySession(sessionID, lastCols, lastRows); err != nil { - h.logger.Debug("failed to trigger final resize", - zap.String("session_id", sessionID), - zap.Error(err)) - } else { - h.logger.Info("triggered TUI redraw via resize", - zap.String("session_id", sessionID), - zap.Uint16("cols", lastCols), - zap.Uint16("rows", lastRows)) - } - - // Some TUI apps (like opencode) need a second redraw trigger - // after a longer delay to fully render - go func(sid string, cols, rows uint16, doneCh <-chan struct{}) { - select { - case <-doneCh: - return - case <-time.After(300 * time.Millisecond): - } - if err := interactiveRunner.ResizeBySession(sid, cols-1, rows); err == nil { - select { - case <-doneCh: - return - case <-time.After(50 * time.Millisecond): - } - _ = interactiveRunner.ResizeBySession(sid, cols, rows) - } - }(sessionID, lastCols, lastRows, done) - } } } + + // Now handle the resize - this sends SIGWINCH which triggers TUI redraw + h.handleResize(data[1:], sessionID, interactiveRunner) continue } // Regular input - write to PTY if available if ptyWriter != nil { if _, err := ptyWriter.Write(data); err != nil { - h.logger.Debug("PTY write error", + h.logger.Debug("PTY write error, attempting reconnect", zap.String("session_id", sessionID), zap.Error(err)) - return + + // Try to reconnect to a new process (may have been restarted) + newWriter, newProcID := h.attemptReconnect(sessionID, interactiveRunner, wsw, done) + if newWriter != nil { + ptyWriter = newWriter + processID = newProcID + directOutputSet = true + + // Retry write with new writer + if _, err := ptyWriter.Write(data); err != nil { + h.logger.Debug("PTY write error after reconnect", + zap.String("session_id", sessionID), + zap.Error(err)) + continue + } + h.detectInputSubmission(sessionID, data) + } else { + h.logger.Debug("reconnect failed, PTY unavailable", + zap.String("session_id", sessionID)) + // Don't return - keep trying on subsequent inputs + continue + } + } else { + // Detect Enter key (user submitted input) - notify lifecycle manager + h.detectInputSubmission(sessionID, data) } - // Detect Enter key (user submitted input) - notify lifecycle manager - h.detectInputSubmission(sessionID, data) } else { // PTY not ready yet - try to set it up if h.setupPtyAccess(sessionID, processID, interactiveRunner, wsw, &ptyWriter) { @@ -345,7 +312,7 @@ func (h *TerminalHandler) runTerminalBridge( h.logger.Debug("PTY write error after setup", zap.String("session_id", sessionID), zap.Error(err)) - return + continue } // Detect Enter key (user submitted input) - notify lifecycle manager h.detectInputSubmission(sessionID, data) @@ -358,6 +325,56 @@ func (h *TerminalHandler) runTerminalBridge( } } +// attemptReconnect tries to reconnect to a new process after the old one exited. +// Returns the new PTY writer and process ID if successful, or nil if not. +func (h *TerminalHandler) attemptReconnect( + sessionID string, + interactiveRunner *process.InteractiveRunner, + wsw *wsWriter, + done <-chan struct{}, +) (io.Writer, string) { + const maxRetries = 20 // 20 retries + const retryInterval = 100 * time.Millisecond // 100ms between retries = 2s total + + for i := 0; i < maxRetries; i++ { + select { + case <-done: + // Bridge is shutting down + return nil, "" + default: + } + + // Try to get the PTY writer for the session (may be a new process) + writer, newProcID, err := interactiveRunner.GetPtyWriterBySession(sessionID) + if err == nil && writer != nil { + // Found a new process - set up direct output + if err := interactiveRunner.SetDirectOutput(newProcID, wsw); err != nil { + h.logger.Debug("failed to set direct output during reconnect", + zap.String("session_id", sessionID), + zap.String("process_id", newProcID), + zap.Error(err)) + } else { + h.logger.Info("reconnected to new process after restart", + zap.String("session_id", sessionID), + zap.String("process_id", newProcID)) + return writer, newProcID + } + } + + // Wait before retrying + select { + case <-done: + return nil, "" + case <-time.After(retryInterval): + } + } + + h.logger.Debug("reconnect attempts exhausted", + zap.String("session_id", sessionID), + zap.Int("attempts", maxRetries)) + return nil, "" +} + // detectInputSubmission checks if the input contains Enter key and notifies // the lifecycle manager to update the execution state to Running. func (h *TerminalHandler) detectInputSubmission(sessionID string, data []byte) {