diff --git a/apps/backend/internal/agent/lifecycle/manager_execution.go b/apps/backend/internal/agent/lifecycle/manager_execution.go index 94ac77563b..410a3dc4fe 100644 --- a/apps/backend/internal/agent/lifecycle/manager_execution.go +++ b/apps/backend/internal/agent/lifecycle/manager_execution.go @@ -2,6 +2,7 @@ package lifecycle import ( "context" + "errors" "fmt" "time" @@ -12,6 +13,10 @@ import ( "github.com/kandev/kandev/internal/common/appctx" ) +// ErrSessionWorkspaceNotReady indicates the task session exists but does not yet +// have a resolved workspace path (typically while worktree preparation is in progress). +var ErrSessionWorkspaceNotReady = errors.New("session workspace not ready") + // EnsureWorkspaceExecutionForSession ensures an agentctl execution exists for a specific task session. // This is used when the frontend provides a session ID (e.g., from URL path /task/[id]/[sessionId]). // If an execution already exists for the session, it returns it. Otherwise, it creates a new execution @@ -128,7 +133,7 @@ func (m *Manager) createExecutionFromSessionInfo(ctx context.Context, sessionID } if info.WorkspacePath == "" { - return nil, fmt.Errorf("session %s has no workspace path configured", sessionID) + return nil, fmt.Errorf("%w: session %s has no workspace path configured", ErrSessionWorkspaceNotReady, sessionID) } if info.TaskID == "" { diff --git a/apps/backend/internal/agent/lifecycle/manager_execution_test.go b/apps/backend/internal/agent/lifecycle/manager_execution_test.go new file mode 100644 index 0000000000..8b3ed5774c --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/manager_execution_test.go @@ -0,0 +1,33 @@ +package lifecycle + +import ( + "errors" + "fmt" + "testing" +) + +func TestErrSessionWorkspaceNotReady_ErrorsIs(t *testing.T) { + // The production code wraps ErrSessionWorkspaceNotReady with fmt.Errorf("%w", ...). + // The terminal handler uses errors.Is to detect this sentinel and trigger retry logic. + // This test ensures the wrapping chain stays detectable. + + wrapped := fmt.Errorf("%w: session test-session has no workspace path configured", ErrSessionWorkspaceNotReady) + + if !errors.Is(wrapped, ErrSessionWorkspaceNotReady) { + t.Errorf("expected errors.Is(wrapped, ErrSessionWorkspaceNotReady) to be true") + } + + // Double-wrapped (as done in ensurePassthroughExecutionReady timeout path) + doubleWrapped := fmt.Errorf("%w: timed out after 30s", ErrSessionWorkspaceNotReady) + if !errors.Is(doubleWrapped, ErrSessionWorkspaceNotReady) { + t.Errorf("expected errors.Is(doubleWrapped, ErrSessionWorkspaceNotReady) to be true") + } +} + +func TestErrSessionWorkspaceNotReady_UnrelatedError(t *testing.T) { + unrelated := fmt.Errorf("some other error: %w", errors.New("connection timeout")) + + if errors.Is(unrelated, ErrSessionWorkspaceNotReady) { + t.Errorf("expected errors.Is to be false for unrelated error") + } +} diff --git a/apps/backend/internal/agent/lifecycle/testing_support.go b/apps/backend/internal/agent/lifecycle/testing_support.go new file mode 100644 index 0000000000..e82fabc583 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/testing_support.go @@ -0,0 +1,7 @@ +package lifecycle + +// ExecutionStoreForTesting returns the manager's execution store. +// Intended for use by test packages that need to inject executions. +func (m *Manager) ExecutionStoreForTesting() *ExecutionStore { + return m.executionStore +} diff --git a/apps/backend/internal/gateway/websocket/terminal_handler.go b/apps/backend/internal/gateway/websocket/terminal_handler.go index b1992fc991..50c787f283 100644 --- a/apps/backend/internal/gateway/websocket/terminal_handler.go +++ b/apps/backend/internal/gateway/websocket/terminal_handler.go @@ -5,6 +5,8 @@ import ( "bytes" "context" "encoding/json" + "errors" + "fmt" "io" "net/http" "net/url" @@ -102,6 +104,12 @@ func checkWebSocketOrigin(r *http.Request) bool { // First byte 0x01 indicates resize, followed by JSON {cols, rows}. const resizeCommandByte = 0x01 +const ( + passthroughReadyTimeout = 30 * time.Second + shellExecutionTimeout = 15 * time.Second + pollInterval = 500 * time.Millisecond +) + // ResizePayload is the JSON payload for resize commands. type ResizePayload struct { Cols uint16 `json:"cols"` @@ -223,17 +231,26 @@ func (h *TerminalHandler) shouldUseWorkspaceShell(ctx context.Context, sessionID // resume flow recreates the execution asynchronously; this bridges the gap so the // terminal doesn't fall back to a local shell. func (h *TerminalHandler) waitForRemoteExecution(ctx context.Context, sessionID string) (*lifecycle.AgentExecution, bool) { + return h.waitForRemoteExecutionWithTimeout(ctx, sessionID, 30*time.Second) +} + +func (h *TerminalHandler) waitForRemoteExecutionWithTimeout( + ctx context.Context, + sessionID string, + timeout time.Duration, +) (*lifecycle.AgentExecution, bool) { // Fast path: execution already exists. if execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID); exists { return execution, true } - h.logger.Info("waiting for remote execution to become available", - zap.String("session_id", sessionID)) + h.logger.Info("waiting for execution to become available", + zap.String("session_id", sessionID), + zap.Duration("timeout", timeout)) - ticker := time.NewTicker(500 * time.Millisecond) + ticker := time.NewTicker(pollInterval) defer ticker.Stop() - timer := time.NewTimer(30 * time.Second) + timer := time.NewTimer(timeout) defer timer.Stop() for { @@ -241,12 +258,13 @@ func (h *TerminalHandler) waitForRemoteExecution(ctx context.Context, sessionID case <-ctx.Done(): return nil, false case <-timer.C: - h.logger.Warn("timed out waiting for remote execution", - zap.String("session_id", sessionID)) + h.logger.Warn("timed out waiting for execution", + zap.String("session_id", sessionID), + zap.Duration("timeout", timeout)) return nil, false case <-ticker.C: if execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID); exists { - h.logger.Info("remote execution became available", + h.logger.Info("execution became available", zap.String("session_id", sessionID), zap.String("execution_id", execution.ID)) return execution, true @@ -386,11 +404,15 @@ func (h *TerminalHandler) handleAgentPassthroughWS( // 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) + execution, err := h.ensurePassthroughExecutionReady(c.Request.Context(), sessionID) if err != nil { h.logger.Warn("failed to ensure passthrough execution", zap.String("session_id", sessionID), zap.Error(err)) + if errors.Is(err, lifecycle.ErrSessionWorkspaceNotReady) { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()}) + return + } c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } @@ -433,6 +455,44 @@ func (h *TerminalHandler) handleAgentPassthroughWS( h.runTerminalBridge(conn, sessionID, processID, interactiveRunner, wsw) } +func (h *TerminalHandler) ensurePassthroughExecutionReady(ctx context.Context, sessionID string) (*lifecycle.AgentExecution, error) { + execution, err := h.lifecycleMgr.EnsurePassthroughExecution(ctx, sessionID) + if err == nil { + return execution, nil + } + if !errors.Is(err, lifecycle.ErrSessionWorkspaceNotReady) { + return nil, err + } + + h.logger.Info("waiting for passthrough workspace to become ready", + zap.String("session_id", sessionID), + zap.Duration("timeout", passthroughReadyTimeout)) + + timeoutCtx, cancel := context.WithTimeout(ctx, passthroughReadyTimeout) + defer cancel() + + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + + for { + select { + case <-timeoutCtx.Done(): + return nil, fmt.Errorf("%w: timed out after %s", lifecycle.ErrSessionWorkspaceNotReady, passthroughReadyTimeout) + case <-ticker.C: + execution, err := h.lifecycleMgr.EnsurePassthroughExecution(timeoutCtx, sessionID) + if err == nil { + h.logger.Info("passthrough workspace became ready", + zap.String("session_id", sessionID), + zap.String("execution_id", execution.ID)) + return execution, nil + } + if !errors.Is(err, lifecycle.ErrSessionWorkspaceNotReady) { + return nil, err + } + } + } +} + // runTerminalBridge handles WebSocket I/O and manages the PTY connection. // It waits for the first resize to trigger process start, then sets up bidirectional I/O. func (h *TerminalHandler) runTerminalBridge( @@ -789,16 +849,31 @@ func (h *TerminalHandler) resolvePreferredShell(ctx context.Context) string { return shell } -// resolveWorkingDir returns the workspace path for the given session, if available. -func (h *TerminalHandler) resolveWorkingDir(sessionID string) string { +// resolveWorkingDir returns the workspace path for the given session. +// It waits briefly for execution/workspace readiness to avoid launching shells in the wrong cwd. +func (h *TerminalHandler) resolveWorkingDir(ctx context.Context, sessionID string) (string, error) { execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID) - if !exists { - h.logger.Info("handleUserShellWS: no execution found, using empty working directory") - return "" + if exists && execution.WorkspacePath != "" { + h.logger.Info("handleUserShellWS: got working directory from execution", + zap.String("working_dir", execution.WorkspacePath)) + return execution.WorkspacePath, nil } - h.logger.Info("handleUserShellWS: got working directory from execution", + + h.logger.Info("handleUserShellWS: waiting for execution workspace path", + zap.String("session_id", sessionID), + zap.Duration("timeout", shellExecutionTimeout)) + + execution, exists = h.waitForRemoteExecutionWithTimeout(ctx, sessionID, shellExecutionTimeout) + if !exists || execution == nil { + return "", fmt.Errorf("session %s execution is not ready yet", sessionID) + } + if execution.WorkspacePath == "" { + return "", fmt.Errorf("%w: session %s has no workspace path configured", lifecycle.ErrSessionWorkspaceNotReady, sessionID) + } + + h.logger.Info("handleUserShellWS: got working directory after wait", zap.String("working_dir", execution.WorkspacePath)) - return execution.WorkspacePath + return execution.WorkspacePath, nil } // resolveShellLabel returns the label and initial command for a user shell, derived @@ -841,7 +916,13 @@ func (h *TerminalHandler) startUserShellProcess( zap.String("initial_command", initialCommand)) preferredShell := h.resolvePreferredShell(c.Request.Context()) - workingDir := h.resolveWorkingDir(sessionID) + workingDir, err := h.resolveWorkingDir(c.Request.Context(), sessionID) + if err != nil { + h.logger.Warn("handleUserShellWS: workspace path not ready", + zap.String("session_id", sessionID), + zap.Error(err)) + return "", http.StatusServiceUnavailable, err.Error() + } opts := &process.UserShellOptions{Label: label, InitialCommand: initialCommand} diff --git a/apps/backend/internal/gateway/websocket/terminal_handler_wait_test.go b/apps/backend/internal/gateway/websocket/terminal_handler_wait_test.go new file mode 100644 index 0000000000..ca8a10b14d --- /dev/null +++ b/apps/backend/internal/gateway/websocket/terminal_handler_wait_test.go @@ -0,0 +1,124 @@ +package websocket + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/agent/lifecycle" + "github.com/kandev/kandev/internal/agent/registry" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// mockEventBus satisfies bus.EventBus for test setup. +type mockEventBus struct{} + +func (m *mockEventBus) Publish(_ context.Context, _ string, _ *bus.Event) error { return nil } +func (m *mockEventBus) Subscribe(_ string, _ bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} +func (m *mockEventBus) QueueSubscribe(_, _ string, _ bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} +func (m *mockEventBus) Request(_ context.Context, _ string, _ *bus.Event, _ time.Duration) (*bus.Event, error) { + return nil, nil +} +func (m *mockEventBus) Close() {} +func (m *mockEventBus) IsConnected() bool { return true } + +// mockProfileResolver satisfies lifecycle.ProfileResolver. +type mockProfileResolver struct{} + +func (m *mockProfileResolver) ResolveProfile(_ context.Context, profileID string) (*lifecycle.AgentProfileInfo, error) { + return &lifecycle.AgentProfileInfo{ProfileID: profileID}, nil +} + +// mockCredsMgr satisfies lifecycle.CredentialsManager. +type mockCredsMgr struct{} + +func (m *mockCredsMgr) GetCredentialValue(_ context.Context, _ string) (string, error) { + return "", nil +} + +func newTestTerminalHandler(t *testing.T) (*TerminalHandler, *lifecycle.ExecutionStore) { + t.Helper() + + log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + reg := registry.NewRegistry(log) + reg.LoadDefaults() + + mgr := lifecycle.NewManager( + reg, &mockEventBus{}, nil, &mockCredsMgr{}, &mockProfileResolver{}, nil, + lifecycle.ExecutorFallbackWarn, "", log, + ) + + handler := &TerminalHandler{ + lifecycleMgr: mgr, + logger: log, + } + + return handler, mgr.ExecutionStoreForTesting() +} + +func TestWaitForRemoteExecutionWithTimeout_FastPath(t *testing.T) { + handler, store := newTestTerminalHandler(t) + + store.Add(&lifecycle.AgentExecution{ + ID: "exec-1", + SessionID: "session-1", + Status: v1.AgentStatusRunning, + }) + + got, ok := handler.waitForRemoteExecutionWithTimeout(context.Background(), "session-1", 1*time.Second) + if !ok { + t.Fatal("expected fast path to find execution") + } + if got.ID != "exec-1" { + t.Errorf("expected exec-1, got %s", got.ID) + } +} + +func TestWaitForRemoteExecutionWithTimeout_PollSuccess(t *testing.T) { + handler, store := newTestTerminalHandler(t) + + // Add execution after a short delay (simulates async workspace setup) + go func() { + time.Sleep(200 * time.Millisecond) + store.Add(&lifecycle.AgentExecution{ + ID: "exec-delayed", + SessionID: "session-1", + Status: v1.AgentStatusRunning, + }) + }() + + got, ok := handler.waitForRemoteExecutionWithTimeout(context.Background(), "session-1", 5*time.Second) + if !ok { + t.Fatal("expected polling to find execution") + } + if got.ID != "exec-delayed" { + t.Errorf("expected exec-delayed, got %s", got.ID) + } +} + +func TestWaitForRemoteExecutionWithTimeout_Timeout(t *testing.T) { + handler, _ := newTestTerminalHandler(t) + + _, ok := handler.waitForRemoteExecutionWithTimeout(context.Background(), "session-missing", 600*time.Millisecond) + if ok { + t.Fatal("expected timeout, got execution") + } +} + +func TestWaitForRemoteExecutionWithTimeout_ContextCancelled(t *testing.T) { + handler, _ := newTestTerminalHandler(t) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, ok := handler.waitForRemoteExecutionWithTimeout(ctx, "session-1", 5*time.Second) + if ok { + t.Fatal("expected context cancellation to return false") + } +} diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 3d7aab79c0..c3ba05bb72 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -1,6 +1,7 @@ package models import ( + "errors" "maps" "time" @@ -8,6 +9,9 @@ import ( v1 "github.com/kandev/kandev/pkg/api/v1" ) +// ErrExecutorRunningNotFound is returned when no executor running record exists for a session. +var ErrExecutorRunningNotFound = errors.New("executor running not found") + // ListMessagesOptions defines pagination options for listing messages type ListMessagesOptions struct { Limit int diff --git a/apps/backend/internal/task/repository/sqlite/executor.go b/apps/backend/internal/task/repository/sqlite/executor.go index 1912b308e5..2f86727218 100644 --- a/apps/backend/internal/task/repository/sqlite/executor.go +++ b/apps/backend/internal/task/repository/sqlite/executor.go @@ -310,7 +310,7 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI &running.UpdatedAt, ) if err == sql.ErrNoRows { - return nil, fmt.Errorf("executor running not found for session: %s", sessionID) + return nil, fmt.Errorf("%w for session: %s", models.ErrExecutorRunningNotFound, sessionID) } if err != nil { return nil, err diff --git a/apps/backend/internal/task/service/service_turns.go b/apps/backend/internal/task/service/service_turns.go index 98118d37d0..90bb3e77e4 100644 --- a/apps/backend/internal/task/service/service_turns.go +++ b/apps/backend/internal/task/service/service_turns.go @@ -246,9 +246,15 @@ func (s *Service) GetWorkspaceInfoForSession(ctx context.Context, taskID, sessio // Populate executor info for correct runtime selection and remote reconnection running, err := s.executors.GetExecutorRunningBySessionID(ctx, sessionID) if err != nil { - s.logger.Warn("failed to get executor running for session", - zap.String("session_id", sessionID), - zap.Error(err)) + if isExecutorRunningNotFoundError(err) { + s.logger.Debug("executor running not ready for session", + zap.String("session_id", sessionID), + zap.Error(err)) + } else { + s.logger.Warn("failed to get executor running for session", + zap.String("session_id", sessionID), + zap.Error(err)) + } } else if running != nil { info.RuntimeName = running.Runtime info.AgentExecutionID = running.AgentExecutionID @@ -267,3 +273,10 @@ func (s *Service) GetWorkspaceInfoForSession(ctx context.Context, taskID, sessio return info, nil } + +func isExecutorRunningNotFoundError(err error) bool { + if err == nil { + return false + } + return errors.Is(err, models.ErrExecutorRunningNotFound) +} diff --git a/apps/backend/internal/task/service/service_turns_helpers_test.go b/apps/backend/internal/task/service/service_turns_helpers_test.go new file mode 100644 index 0000000000..b899b762ec --- /dev/null +++ b/apps/backend/internal/task/service/service_turns_helpers_test.go @@ -0,0 +1,52 @@ +package service + +import ( + "errors" + "fmt" + "testing" + + "github.com/kandev/kandev/internal/task/models" +) + +func TestIsExecutorRunningNotFoundError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + { + name: "nil error", + err: nil, + want: false, + }, + { + name: "sentinel error directly", + err: models.ErrExecutorRunningNotFound, + want: true, + }, + { + name: "wrapped sentinel error", + err: fmt.Errorf("%w for session: abc-123", models.ErrExecutorRunningNotFound), + want: true, + }, + { + name: "double-wrapped sentinel error", + err: fmt.Errorf("outer: %w", fmt.Errorf("%w for session: abc-123", models.ErrExecutorRunningNotFound)), + want: true, + }, + { + name: "unrelated error", + err: errors.New("connection refused"), + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := isExecutorRunningNotFoundError(tt.err) + if got != tt.want { + t.Errorf("isExecutorRunningNotFoundError(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} diff --git a/apps/web/components/task/passthrough-terminal.tsx b/apps/web/components/task/passthrough-terminal.tsx index 1f35cd9568..449e6f36f9 100644 --- a/apps/web/components/task/passthrough-terminal.tsx +++ b/apps/web/components/task/passthrough-terminal.tsx @@ -145,7 +145,7 @@ export function PassthroughTerminal(props: PassthroughTerminalProps) {
- {mode === "agent" ? "Starting agent..." : "Connecting..."} + {mode === "agent" ? "Preparing workspace..." : "Connecting terminal..."}
diff --git a/apps/web/components/task/use-passthrough-terminal.test.ts b/apps/web/components/task/use-passthrough-terminal.test.ts new file mode 100644 index 0000000000..f9cc063e7a --- /dev/null +++ b/apps/web/components/task/use-passthrough-terminal.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest"; +import { reconnectDelayMs } from "./ws-reconnect"; + +describe("reconnectDelayMs", () => { + it("returns 300ms for attempt 0", () => { + expect(reconnectDelayMs(0)).toBe(300); + }); + + it("doubles delay for each attempt", () => { + expect(reconnectDelayMs(0)).toBe(300); + expect(reconnectDelayMs(1)).toBe(600); + expect(reconnectDelayMs(2)).toBe(1200); + expect(reconnectDelayMs(3)).toBe(2400); + expect(reconnectDelayMs(4)).toBe(4800); + }); + + it("caps at 5000ms", () => { + expect(reconnectDelayMs(5)).toBe(5000); + }); + + it("caps attempt at 5 so high values stay at 5000ms", () => { + expect(reconnectDelayMs(10)).toBe(5000); + expect(reconnectDelayMs(100)).toBe(5000); + }); +}); diff --git a/apps/web/components/task/use-passthrough-terminal.ts b/apps/web/components/task/use-passthrough-terminal.ts index 2f3af2edaf..1ae4880881 100644 --- a/apps/web/components/task/use-passthrough-terminal.ts +++ b/apps/web/components/task/use-passthrough-terminal.ts @@ -6,6 +6,7 @@ import { Unicode11Addon } from "@xterm/addon-unicode11"; import { WebLinksAddon } from "@xterm/addon-web-links"; import { WebglAddon } from "@xterm/addon-webgl"; import { getTerminalTheme } from "@/lib/theme/terminal-theme"; +import { startReconnectLoop } from "./ws-reconnect"; // Debug flag - set to true to see detailed logs const DEBUG = false; @@ -272,6 +273,7 @@ type ConnectWebSocketOptions = { isMountedCheck: () => boolean; onTimeout: (id: ReturnType) => void; onConnected: () => void; + onSocketClose: (event: CloseEvent) => void; }; function connectWebSocket({ @@ -287,6 +289,7 @@ function connectWebSocket({ isMountedCheck, onTimeout, onConnected, + onSocketClose, }: ConnectWebSocketOptions) { if (attachAddonRef.current) { attachAddonRef.current.dispose(); @@ -341,12 +344,17 @@ function connectWebSocket({ attachAddonRef.current.dispose(); attachAddonRef.current = null; } + onSocketClose(event); }; ws.onerror = (error) => { log("WebSocket error:", error); }; } +// reconnectDelayMs and startReconnectLoop are in ws-reconnect.ts +// Re-export reconnectDelayMs for tests that import from this module. +export { reconnectDelayMs } from "./ws-reconnect"; + export function useWebSocketConnection({ taskId, sessionId, @@ -390,33 +398,22 @@ export function useWebSocketConnection({ log("Resetting terminal buffer for session", sessionId); terminal.reset(); - let isMounted = true; - let connectTimeout: ReturnType | null = null; - let settleTimeout: ReturnType | null = null; - connectTimeout = setTimeout(() => { - if (!isMounted) return; - connectWebSocket({ - sessionId, - wsBaseUrl, - mode, - terminalId, - label, - terminal, - fitAndResize, - wsRef, - attachAddonRef, - isMountedCheck: () => isMounted, - onTimeout: (id) => { - settleTimeout = id; - }, - onConnected, - }); - }, 150); + const stopReconnectLoop = startReconnectLoop({ + sessionId, + wsBaseUrl, + mode, + terminalId, + label, + terminal, + fitAndResize, + wsRef, + attachAddonRef, + onConnected, + connectWebSocket, + }); return () => { log("WebSocket cleanup"); - isMounted = false; - if (connectTimeout) clearTimeout(connectTimeout); - if (settleTimeout) clearTimeout(settleTimeout); + stopReconnectLoop(); if (attachAddonRef.current) { attachAddonRef.current.dispose(); attachAddonRef.current = null; diff --git a/apps/web/components/task/ws-reconnect.ts b/apps/web/components/task/ws-reconnect.ts new file mode 100644 index 0000000000..255a92dc08 --- /dev/null +++ b/apps/web/components/task/ws-reconnect.ts @@ -0,0 +1,115 @@ +import { Terminal } from "@xterm/xterm"; +import { AttachAddon } from "@xterm/addon-attach"; +import { log } from "./use-passthrough-terminal"; + +const STABLE_CONNECTION_MS = 500; + +export function reconnectDelayMs(attempt: number): number { + const cappedAttempt = Math.min(attempt, 5); + return Math.min(5000, 300 * 2 ** cappedAttempt); +} + +type ConnectWebSocketFn = (opts: { + sessionId: string; + wsBaseUrl: string; + mode: "agent" | "shell"; + terminalId: string | undefined; + label: string | undefined; + terminal: Terminal; + fitAndResize: (force?: boolean) => void; + wsRef: React.MutableRefObject; + attachAddonRef: React.MutableRefObject; + isMountedCheck: () => boolean; + onTimeout: (id: ReturnType) => void; + onConnected: () => void; + onSocketClose: (event: CloseEvent) => void; +}) => void; + +export type ReconnectLoopOptions = { + sessionId: string; + wsBaseUrl: string; + mode: "agent" | "shell"; + terminalId: string | undefined; + label: string | undefined; + terminal: Terminal; + fitAndResize: (force?: boolean) => void; + wsRef: React.MutableRefObject; + attachAddonRef: React.MutableRefObject; + onConnected: () => void; + connectWebSocket: ConnectWebSocketFn; +}; + +export function startReconnectLoop({ + sessionId, + wsBaseUrl, + mode, + terminalId, + label, + terminal, + fitAndResize, + wsRef, + attachAddonRef, + onConnected, + connectWebSocket, +}: ReconnectLoopOptions): () => void { + let isMounted = true; + let connectTimeout: ReturnType | null = null; + let settleTimeout: ReturnType | null = null; + let stableOpenTimeout: ReturnType | null = null; + let retryAttempt = 0; + + const scheduleConnect = (delayMs: number) => { + if (!isMounted) return; + if (connectTimeout) clearTimeout(connectTimeout); + connectTimeout = setTimeout(() => { + if (!isMounted) return; + connectWebSocket({ + sessionId, + wsBaseUrl, + mode, + terminalId, + label, + terminal, + fitAndResize, + wsRef, + attachAddonRef, + isMountedCheck: () => isMounted, + onTimeout: (id) => { + settleTimeout = id; + }, + onConnected: () => { + if (stableOpenTimeout) clearTimeout(stableOpenTimeout); + stableOpenTimeout = setTimeout(() => { + retryAttempt = 0; + stableOpenTimeout = null; + }, STABLE_CONNECTION_MS); + onConnected(); + }, + onSocketClose: (event) => { + if (!isMounted) return; + if (stableOpenTimeout) { + clearTimeout(stableOpenTimeout); + stableOpenTimeout = null; + } + const nextDelay = reconnectDelayMs(retryAttempt); + retryAttempt += 1; + log("Scheduling reconnect", { + attempt: retryAttempt, + delayMs: nextDelay, + code: event.code, + }); + scheduleConnect(nextDelay); + }, + }); + }, delayMs); + }; + + scheduleConnect(150); + + return () => { + isMounted = false; + if (connectTimeout) clearTimeout(connectTimeout); + if (settleTimeout) clearTimeout(settleTimeout); + if (stableOpenTimeout) clearTimeout(stableOpenTimeout); + }; +}