diff --git a/packages/cloud-agent-sdk/src/service-state.test.ts b/packages/cloud-agent-sdk/src/service-state.test.ts index 9ac91419f2..59c474ed90 100644 --- a/packages/cloud-agent-sdk/src/service-state.test.ts +++ b/packages/cloud-agent-sdk/src/service-state.test.ts @@ -1738,6 +1738,44 @@ describe('createServiceState', () => { }); }); + it('terminal delivery failure resolves a stale preparing status', () => { + const state = createServiceState(makeConfig()); + + state.process({ + type: 'cloud.status', + cloudStatus: { type: 'preparing', message: 'Setting up environment...' }, + }); + state.process({ + type: 'cloud.message.failed', + messageId: 'm1', + error: 'Environment preparation failed', + reason: 'exhausted', + }); + + expect(state.getCloudStatus()).toEqual({ + type: 'error', + message: 'Environment preparation failed', + }); + }); + + it('an interrupt during preparation clears the preparing status', () => { + const state = createServiceState(makeConfig()); + + state.process({ + type: 'cloud.status', + cloudStatus: { type: 'preparing', message: 'Setting up environment...' }, + }); + state.process({ + type: 'cloud.message.failed', + messageId: 'm1', + error: 'The message was interrupted', + reason: 'interrupted', + }); + + expect(state.getCloudStatus()).toBeNull(); + expect(state.getStatus()).toEqual({ type: 'interrupted' }); + }); + it('cloud.message.failed with reason=interrupted settles the session', () => { const state = createServiceState(makeConfig()); diff --git a/packages/cloud-agent-sdk/src/service-state.ts b/packages/cloud-agent-sdk/src/service-state.ts index 0d0eb46625..5832f51751 100644 --- a/packages/cloud-agent-sdk/src/service-state.ts +++ b/packages/cloud-agent-sdk/src/service-state.ts @@ -591,6 +591,14 @@ function createServiceState(config: ServiceStateConfig): ServiceState { ...(event.attempts !== undefined ? { attempts: event.attempts } : {}), }; pendingMessages.set(event.messageId, deliveryState); + // A preparation failure can arrive as a terminal message-delivery event + // without a separate preparing event. Do not leave the composer showing + // "Setting up environment" forever in that case. An interrupt is the user + // cancelling, not a failure, so it clears the stale status instead of + // raising an error banner. + if (cloudStatus?.type === 'preparing') { + cloudStatus = event.reason === 'interrupted' ? null : { type: 'error', message: event.error }; + } if (event.reason === 'interrupted') { activity = { type: 'idle' }; status = { type: 'interrupted' }; diff --git a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts index b5c6fec16b..49a6357fbd 100644 --- a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts +++ b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts @@ -3551,7 +3551,7 @@ export class CloudAgentSession extends DurableObject { { ...plan, preparation: { attemptId: recorder.attemptId } }, { onProgress: (step, message) => { - recorder.onProgress(step, message); + if (!recorder.onProgress(step, message)) return; this.broadcastVolatileEvent({ executionId: eventSourceId, sessionId, @@ -3567,6 +3567,7 @@ export class CloudAgentSession extends DurableObject { if (!readyResult.success) { throw new Error(readyResult.error ?? 'Failed to record session readiness'); } + recorder.finalize({ status: 'completed' }); }, onAccepted: delivery => this.recordRuntimeAcceptedMessage(plan, delivery), } diff --git a/services/cloud-agent-next/src/session/preparation-progress.test.ts b/services/cloud-agent-next/src/session/preparation-progress.test.ts index 25134a5eb8..d74f14d6a0 100644 --- a/services/cloud-agent-next/src/session/preparation-progress.test.ts +++ b/services/cloud-agent-next/src/session/preparation-progress.test.ts @@ -143,6 +143,19 @@ describe('createPreparationProgressRecorder', () => { expect(eventQueries.findByEntityPrefix('preparation/attempt/')).toEqual([]); }); + it('ignores progress received after the attempt was finalized', () => { + const eventQueries = createMemoryEventQueries(); + const broadcasts: StoredEvent[] = []; + const recorder = createRecorder(eventQueries, broadcasts); + + recorder.onProgress('sandbox_provision', 'Provisioning sandbox…'); + recorder.finalize({ status: 'failed', safeError: 'Environment preparation failed' }); + broadcasts.length = 0; + + expect(recorder.onProgress('cloning', 'Cloning repository…')).toBe(false); + expect(broadcasts).toEqual([]); + }); + it('finalize settles an attempt the wrapper continued but never terminated', () => { const eventQueries = createMemoryEventQueries(); const broadcasts: StoredEvent[] = []; diff --git a/services/cloud-agent-next/src/session/preparation-progress.ts b/services/cloud-agent-next/src/session/preparation-progress.ts index e905df3408..0876967417 100644 --- a/services/cloud-agent-next/src/session/preparation-progress.ts +++ b/services/cloud-agent-next/src/session/preparation-progress.ts @@ -19,7 +19,7 @@ import { export type PreparationProgressRecorder = { readonly attemptId: string; /** Translate a legacy (step, message) progress callback into v2 events. */ - onProgress(step: string, message: string): void; + onProgress(step: string, message: string): boolean; /** * Drive the attempt to a terminal state if it is still running. A no-op * when no preparation happened or the wrapper already finished the attempt. @@ -72,7 +72,7 @@ export function createPreparationProgressRecorder(options: { if (materializePreparationEvent(eventQueries, stored, data)) broadcast(stored); } - function onProgress(step: string, message: string): void { + function onProgress(step: string, message: string): boolean { const key = step as PreparingStep; if (!readPreparationAttempt(eventQueries, attemptId)) { emit('workspace_setup', 'Preparing environment', { action: 'attempt_started' }); @@ -91,10 +91,12 @@ export function createPreparationProgressRecorder(options: { activeStep = { id: stepId, key }; } emit(key, message, { action: 'step_progress', stepId, detail: message }); + return readPreparationAttempt(eventQueries, attemptId)?.status === 'running'; } function finalize(outcome: PreparationOutcome): void { activeStep = undefined; + if (!readPreparationAttempt(eventQueries, attemptId)) return; for (const event of finalizePreparationAttempt(eventQueries, attemptId, { ...outcome, timestamp: now(), diff --git a/services/cloud-agent-next/test/integration/session/execute-directly-failure.test.ts b/services/cloud-agent-next/test/integration/session/execute-directly-failure.test.ts index 0c62b7405d..5d089f1754 100644 --- a/services/cloud-agent-next/test/integration/session/execute-directly-failure.test.ts +++ b/services/cloud-agent-next/test/integration/session/execute-directly-failure.test.ts @@ -158,6 +158,134 @@ describe('executeDirectly failure handling', () => { expect(result.wrapperRuntimeState.noOutputDeadlineAt).toBeGreaterThan(result.staleDeadline); }); + /** + * A held delivery (the previous wrapper batch is still finalizing) never + * reaches preparation: the message stays queued and is retried moments later. + * Settling a preparation attempt from that outcome would flash a spurious + * "Environment preparation failed" card between the two tries. + */ + async function drainWithRuntimeResult( + keySuffix: string, + result: { success: false; code: string; error: string }, + options: { emitProgress?: boolean; reportWorkspaceReady?: boolean } = {} + ) { + const userId = `user_exec_direct_${keySuffix}`; + const sessionId = `agent_exec_direct_${keySuffix}`; + const stub = env.CLOUD_AGENT_SESSION.get( + env.CLOUD_AGENT_SESSION.idFromName(`${userId}:${sessionId}`) + ); + + return runInDurableObject(stub, async (instance, state) => { + (instance as any).agentRuntime = { + send: async (_plan: unknown, hooks: any) => { + if (options.emitProgress) { + hooks?.onProgress?.('sandbox_provision', 'Provisioning sandbox…'); + } + if (options.reportWorkspaceReady) { + await hooks?.onWorkspaceReady?.({ + workspacePath: `/workspace/${userId}/sessions/${sessionId}`, + sandboxId: 'usr-123456789abc', + sessionHome: `/home/${sessionId}`, + branchName: `session/${sessionId}`, + kiloSessionId: 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + }); + } + return result; + }, + requestSnapshot: async () => {}, + interruptWrapper: async () => ({ commandSent: false }), + sendPing: () => {}, + keepSandboxAlive: async () => {}, + }; + + await registerReadySession(instance, { + sessionId, + userId, + orgId: `org_exec_direct_${keySuffix}`, + kiloSessionId: 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + prompt: 'initial prompt', + mode: 'code', + model: 'test-model', + kilocodeToken: `token-${keySuffix}`, + }); + + await instance.admitSubmittedMessage( + queueUserMessageInput({ + userId, + prompt: 'do some work', + messageId: 'msg_018f1e2d3c4bHeldMsgAbCdEfG', + }) + ); + await instance.alarm(); + + const db = drizzle(state.storage, { logger: false }); + const eventQueries = createEventQueries(db, state.storage.sql); + const attemptRows = eventQueries.findByEntityPrefix('preparation/attempt/'); + return { + pending: await listPendingSessionMessages(instance.ctx.storage), + preparationEvents: attemptRows, + attemptStatuses: attemptRows + .map(row => parsePreparationAttemptStatus(row.payload)) + .filter((status): status is string => status !== null), + }; + }); + } + + it('a held delivery leaves no preparation attempt behind', async () => { + const result = await drainWithRuntimeResult('held', { + success: false, + code: 'WRAPPER_FINALIZING', + error: 'Wrapper batch is finalizing', + }); + + expect(result.preparationEvents).toEqual([]); + // Held, not failed: the message is still queued for the next drain. + expect(result.pending.map(message => message.messageId)).toEqual([ + 'msg_018f1e2d3c4bHeldMsgAbCdEfG', + ]); + }); + + it('a hold raised after preparation started terminalizes that attempt', async () => { + const result = await drainWithRuntimeResult( + 'held-after-progress', + { success: false, code: 'WRAPPER_FINALIZING', error: 'Wrapper batch is finalizing' }, + { emitProgress: true } + ); + + // The attempt exists (progress was observed) and must not be left running, + // or clients stay in the preparing state forever. + expect(result.attemptStatuses).toEqual(['failed']); + }); + + function parsePreparationAttemptStatus(payload: unknown): string | null { + const parsed = JSON.parse(String(payload)) as { attempt?: { status?: string } }; + return parsed.attempt?.status ?? null; + } + + it('a terminal delivery failure before preparation leaves no attempt behind', async () => { + const result = await drainWithRuntimeResult('terminal', { + success: false, + code: 'SANDBOX_CAPABILITY_UNAVAILABLE', + error: 'Sandbox capability unavailable', + }); + + expect(result.preparationEvents).toEqual([]); + }); + + it('a delivery failure after workspace readiness leaves the attempt completed', async () => { + const result = await drainWithRuntimeResult( + 'failure-after-ready', + { + success: false, + code: 'WRAPPER_START_FAILED', + error: 'Prompt dispatch failed', + }, + { emitProgress: true, reportWorkspaceReady: true } + ); + + expect(result.attemptStatuses).toEqual(['completed']); + }); + it('queued flush pre-start failure retries cleanly with the original execution and message ids', async () => { const userId = 'user_exec_direct_fail'; const sessionId = 'agent_exec_direct_fail'; diff --git a/services/cloud-agent-next/test/unit/wrapper/reconnection.test.ts b/services/cloud-agent-next/test/unit/wrapper/reconnection.test.ts index a3e815648e..b3cc231c6c 100644 --- a/services/cloud-agent-next/test/unit/wrapper/reconnection.test.ts +++ b/services/cloud-agent-next/test/unit/wrapper/reconnection.test.ts @@ -19,6 +19,9 @@ import { import { CODE_REVIEW_PERMISSION_REJECTION_MESSAGE, createConnectionManager, + INGEST_INITIAL_CONNECT_TIMEOUT_MS, + INGEST_RECONNECT_CONNECT_TIMEOUT_MS, + RECONNECT_TOTAL_BUDGET_MS, openIngestProgressChannel, type ConnectionCallbacks, } from '../../../wrapper/src/connection.js'; @@ -575,33 +578,137 @@ describe('ingest WS reconnection', () => { // Test: reconnection fails after all attempts // ------------------------------------------------------------------------- - it('calls onDisconnect after all reconnection attempts fail', async () => { + it('keeps retrying reconnect handshakes until one opens', async () => { const manager = createManager(); const ws = await openConnection(manager); ws.simulateClose(1006); expect(manager.isReconnecting()).toBe(true); - // Backoff delays: 1s, 2s, 4s (3 attempts) - const delays = [1_000, 2_000, 4_000]; + const delays = [1_000, 2_000, 4_000, 8_000]; + for (const delay of delays) { + await vi.advanceTimersByTimeAsync(delay); + MockWebSocket.latest!.simulateError(); + await vi.advanceTimersByTimeAsync(0); + } - for (let i = 0; i < delays.length; i++) { - expect(callbacks.onReconnecting).toHaveBeenCalledWith(i + 1); - await vi.advanceTimersByTimeAsync(delays[i]); + expect(callbacks.onDisconnect).not.toHaveBeenCalled(); + expect(manager.isReconnecting()).toBe(true); + expect(callbacks.onReconnecting).toHaveBeenCalledWith(5); + + await vi.advanceTimersByTimeAsync(8_000); + const laterWs = MockWebSocket.latest!; + laterWs.simulateOpen(); + await vi.advanceTimersByTimeAsync(0); + + expect(callbacks.onReconnected).toHaveBeenCalled(); + expect(manager.isReconnecting()).toBe(false); + }); - // New WS created — simulate error so openIngestWs rejects - const attemptWs = MockWebSocket.latest!; - attemptWs.simulateError(); + it('gives up and reports a disconnect once the reconnect budget is exhausted', async () => { + const manager = createManager(); + const ws = await openConnection(manager); - // Let the rejection propagate and next attempt to schedule + ws.simulateClose(1006); + + // The DO rejects a stale run/connection fence forever, so every handshake + // fails immediately. Burn past the wall-clock budget. The iteration cap is + // only there so an unbounded retry loop fails the test instead of hanging + // it: the budget divided by the capped delay is the real bound. + const maxAttempts = Math.ceil(RECONNECT_TOTAL_BUDGET_MS / 8_000) + 10; + let delay = 1_000; + let attempts = 0; + while (manager.isReconnecting() && attempts < maxAttempts) { + await vi.advanceTimersByTimeAsync(delay); + MockWebSocket.latest!.simulateError(); await vi.advanceTimersByTimeAsync(0); + delay = Math.min(delay * 2, 8_000); + attempts++; } - // After 3 failures, onDisconnect should fire expect(callbacks.onDisconnect).toHaveBeenCalledWith( 'ingest websocket closed (reconnection failed)' ); expect(manager.isReconnecting()).toBe(false); + + // A give-up before the budget elapsed would reintroduce the bug this + // ceiling replaces: a single hung handshake alone costs 90s. + expect(vi.getTimerCount()).toBe(0); + }); + + it('is still reconnecting well before the budget elapses', async () => { + const manager = createManager(); + const ws = await openConnection(manager); + + ws.simulateClose(1006); + + let elapsed = 0; + let delay = 1_000; + while (elapsed + delay < RECONNECT_TOTAL_BUDGET_MS / 2) { + await vi.advanceTimersByTimeAsync(delay); + MockWebSocket.latest!.simulateError(); + await vi.advanceTimersByTimeAsync(0); + elapsed += delay; + delay = Math.min(delay * 2, 8_000); + } + + expect(callbacks.onDisconnect).not.toHaveBeenCalled(); + expect(manager.isReconnecting()).toBe(true); + }); + + it('keeps a reconnect handshake open past the initial-connect timeout', async () => { + const manager = createManager(); + const ws = await openConnection(manager); + + ws.simulateClose(1006); + await vi.advanceTimersByTimeAsync(1_000); + const reconnectWs = MockWebSocket.latest!; + expect(reconnectWs).not.toBe(ws); + + await vi.advanceTimersByTimeAsync(INGEST_INITIAL_CONNECT_TIMEOUT_MS); + expect(callbacks.onDisconnect).not.toHaveBeenCalled(); + expect(manager.isReconnecting()).toBe(true); + expect(MockWebSocket.latest).toBe(reconnectWs); + + reconnectWs.simulateOpen(); + await vi.advanceTimersByTimeAsync(0); + + expect(callbacks.onReconnected).toHaveBeenCalled(); + expect(manager.isReconnecting()).toBe(false); + }); + + it('times out a hung reconnect handshake and starts the next attempt', async () => { + const manager = createManager(); + const ws = await openConnection(manager); + + ws.simulateClose(1006); + await vi.advanceTimersByTimeAsync(1_000); + const firstReconnectWs = MockWebSocket.latest!; + + await vi.advanceTimersByTimeAsync(INGEST_RECONNECT_CONNECT_TIMEOUT_MS - 1); + expect(MockWebSocket.latest).toBe(firstReconnectWs); + expect(callbacks.onDisconnect).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + await vi.advanceTimersByTimeAsync(0); + expect(firstReconnectWs.readyState).toBe(MockWebSocket.CLOSED); + + await vi.advanceTimersByTimeAsync(2_000); + const secondReconnectWs = MockWebSocket.latest!; + expect(secondReconnectWs).not.toBe(firstReconnectWs); + expect(manager.isReconnecting()).toBe(true); + expect(callbacks.onDisconnect).not.toHaveBeenCalled(); + }); + + it('still fail-fasts the initial ingest handshake', async () => { + const manager = createManager(); + const openPromise = manager.open(); + const initialWs = MockWebSocket.latest!; + const rejected = expect(openPromise).rejects.toThrow('Timed out before open'); + + await vi.advanceTimersByTimeAsync(INGEST_INITIAL_CONNECT_TIMEOUT_MS); + await rejected; + expect(initialWs.readyState).toBe(MockWebSocket.CLOSED); }); // ------------------------------------------------------------------------- @@ -811,8 +918,8 @@ describe('ingest WS reconnection', () => { ws.simulateClose(1006); // Track when each new WS instance is created by checking instance count - // Backoff: attempt 1 = 1s, attempt 2 = 2s, attempt 3 = 4s - const delays = [1_000, 2_000, 4_000]; + // Backoff: 1s, 2s, 4s, then capped at 8s + const delays = [1_000, 2_000, 4_000, 8_000, 8_000]; for (let i = 0; i < delays.length; i++) { const countBefore = MockWebSocket.instances.length; diff --git a/services/cloud-agent-next/wrapper/src/connection.ts b/services/cloud-agent-next/wrapper/src/connection.ts index d906b3e3d0..09b7c329a5 100644 --- a/services/cloud-agent-next/wrapper/src/connection.ts +++ b/services/cloud-agent-next/wrapper/src/connection.ts @@ -303,17 +303,30 @@ export function buildIngestConnectionFailureMessage( return `Failed to connect to ingest: ${details.wsUrl} (${INGEST_CONNECTION_FAILURE_HINTS[details.reason]}${closeDetails})`; } -/** Maximum number of reconnection attempts before giving up. - * 3 attempts ≈ 7s total (1+2+4), fitting within the DO's 10s grace period. */ -const MAX_RECONNECT_ATTEMPTS = 3; -/** Base delay for exponential backoff (1 second) */ +/** Base delay for exponential backoff between reconnect handshakes. */ const RECONNECT_BASE_DELAY_MS = 1_000; +/** Cap so a long Durable Object blackout does not grow the gap without bound. */ +const RECONNECT_MAX_DELAY_MS = 8_000; +/** Wall-clock budget for a whole reconnect campaign. Some rejections are + * permanent for this wrapper identity — the DO answers a stale run/connection + * fence with 409 forever — and a WebSocket upgrade failure exposes no status + * code to classify, so the ceiling is time. Generous enough to ride out a long + * blackout (several hung `INGEST_RECONNECT_CONNECT_TIMEOUT_MS` handshakes), + * after which the wrapper reports a disconnect so the Kilo turn is aborted and + * the batch drains instead of retrying into a socket nobody will accept. */ +export const RECONNECT_TOTAL_BUDGET_MS = 5 * 60_000; /** Maximum time to wait for the SDK SSE `/event` handshake before aborting. * The wrapper talks to kilo on loopback, so handshakes typically complete in * <5ms; a 5s budget covers kilo startup hiccups without blocking `open()` on * a silently stuck HTTP stream. */ const SUBSCRIBE_HANDSHAKE_TIMEOUT_MS = 5_000; -const INGEST_INITIAL_CONNECT_TIMEOUT_MS = 10_000; +/** Fail-fast budget for the first ingest handshake (bad WORKER_URL / network). */ +export const INGEST_INITIAL_CONNECT_TIMEOUT_MS = 10_000; +/** Recycle a hung reconnect handshake. Does not abort the Kilo turn — the + * wrapper keeps trying until `close()`. A blocked DO can accept an already- + * started socket tens of seconds late; aborting at 10s produced zombie ingest + * sockets the wrapper had already given up on. */ +export const INGEST_RECONNECT_CONNECT_TIMEOUT_MS = 90_000; const MODEL_NOT_FOUND_DIAGNOSTICS_TIMEOUT_MS = 1_000; function buildIngestWebSocketUrl(session: NonNullable): string { @@ -487,6 +500,7 @@ export function createConnectionManager( let closedByUs = false; let reconnecting = false; let reconnectAttempt = 0; + let reconnectStartedAt = 0; let reconnectTimer: ReturnType | null = null; let generation = 0; @@ -707,8 +721,14 @@ export function createConnectionManager( * @param expectedGeneration If provided, the connection is only accepted when * `generation` still matches. This prevents a stale reconnect from assigning * `ingestWs` and flushing buffered events after `close()` was called. + * @param connectTimeoutMs How long to wait for `onopen` before aborting this + * handshake. Initial connect stays short; reconnects use a longer budget so + * a stalled DO can still accept the in-flight socket. */ - async function openIngestWs(expectedGeneration?: number): Promise { + async function openIngestWs( + expectedGeneration?: number, + connectTimeoutMs: number = INGEST_INITIAL_CONNECT_TIMEOUT_MS + ): Promise { const session = state.currentSession; if (!session) { throw new Error('Cannot open ingest WS: no session context'); @@ -838,7 +858,7 @@ export function createConnectionManager( /* ignore */ } } - }, INGEST_INITIAL_CONNECT_TIMEOUT_MS); + }, connectTimeoutMs); }); } @@ -1290,6 +1310,7 @@ export function createConnectionManager( if (reconnecting) return; reconnecting = true; reconnectAttempt = 0; + reconnectStartedAt = Date.now(); scheduleReconnect(); } @@ -1297,6 +1318,7 @@ export function createConnectionManager( logToFile(`reconnected successfully on attempt ${reconnectAttempt}`); reconnecting = false; reconnectAttempt = 0; + reconnectStartedAt = 0; // Re-store ingest WS in state (event subscription abort controller unchanged) const existingAbort = state.sseAbortController; if (ingestWs && existingAbort) { @@ -1325,23 +1347,29 @@ export function createConnectionManager( } function scheduleReconnect(): void { - reconnectAttempt++; - if (reconnectAttempt > MAX_RECONNECT_ATTEMPTS) { - logToFile(`reconnection failed after ${MAX_RECONNECT_ATTEMPTS} attempts — giving up`); + const elapsed = Date.now() - reconnectStartedAt; + if (elapsed >= RECONNECT_TOTAL_BUDGET_MS) { + logToFile( + `reconnection failed after ${reconnectAttempt} attempts over ${elapsed}ms — giving up` + ); reconnecting = false; reconnectAttempt = 0; callbacks.onDisconnect('ingest websocket closed (reconnection failed)'); return; } - const delay = RECONNECT_BASE_DELAY_MS * Math.pow(2, reconnectAttempt - 1); - logToFile(`reconnect attempt ${reconnectAttempt}/${MAX_RECONNECT_ATTEMPTS} in ${delay}ms`); + reconnectAttempt++; + const delay = Math.min( + RECONNECT_BASE_DELAY_MS * 2 ** (reconnectAttempt - 1), + RECONNECT_MAX_DELAY_MS + ); + logToFile(`reconnect attempt ${reconnectAttempt} in ${delay}ms`); callbacks.onReconnecting?.(reconnectAttempt); reconnectTimer = setTimeout(() => { reconnectTimer = null; const gen = generation; - openIngestWs(gen) + openIngestWs(gen, INGEST_RECONNECT_CONNECT_TIMEOUT_MS) .then(() => { if (gen !== generation) { discardStaleReconnect(); @@ -1365,6 +1393,7 @@ export function createConnectionManager( } reconnecting = false; reconnectAttempt = 0; + reconnectStartedAt = 0; } return {