fix: wait for workspace readiness in terminal connections (#411)

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 == "" {

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")
+	}
+}

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
+}

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}
 

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")
+	}
+}

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

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

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)
+}

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)
+			}
+		})
+	}
+}

apps/web/components/task/passthrough-terminal.tsx

@@ -145,7 +145,7 @@ export function PassthroughTerminal(props: PassthroughTerminalProps) {
           <div className="flex flex-col items-center gap-3 text-muted-foreground">
             <GridSpinner />
             <span className="text-sm">
-              {mode === "agent" ? "Starting agent..." : "Connecting..."}
+              {mode === "agent" ? "Preparing workspace..." : "Connecting terminal..."}
             </span>
           </div>
         </div>

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);
+  });
+});

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<typeof setTimeout>) => 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<typeof setTimeout> | null = null;
-    let settleTimeout: ReturnType<typeof setTimeout> | 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;

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<WebSocket | null>;
+  attachAddonRef: React.MutableRefObject<AttachAddon | null>;
+  isMountedCheck: () => boolean;
+  onTimeout: (id: ReturnType<typeof setTimeout>) => 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<WebSocket | null>;
+  attachAddonRef: React.MutableRefObject<AttachAddon | null>;
+  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<typeof setTimeout> | null = null;
+  let settleTimeout: ReturnType<typeof setTimeout> | null = null;
+  let stableOpenTimeout: ReturnType<typeof setTimeout> | 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);
+  };
+}