diff --git a/packages/core/src/cloud-task/cloud-task-engine.ts b/packages/core/src/cloud-task/cloud-task-engine.ts index 05e297f779..abb90c5789 100644 --- a/packages/core/src/cloud-task/cloud-task-engine.ts +++ b/packages/core/src/cloud-task/cloud-task-engine.ts @@ -31,6 +31,12 @@ const SSE_RECONNECT_BASE_DELAY_MS = 500; const SSE_RECONNECT_FLAT_ATTEMPTS = 3; const SSE_RECONNECT_MAX_DELAY_MS = 30_000; const SSE_HEALTHY_CONNECTION_MS = 60_000; +// The backend emits a keepalive at least every ~25-30s (see SSE_KEEPALIVE_INTERVAL_MS in +// packages/agent). A half-open socket (laptop sleep, unplugged NIC, NAT rebind) neither errors +// nor EOFs, so `reader.read()` awaits forever with nothing to trigger reconnect. This timeout +// treats "no bytes at all for a few keepalive intervals" as a disconnect so it flows into the +// existing reconnect/backoff machinery instead of hanging the watcher indefinitely. +const SSE_IDLE_TIMEOUT_MS = 90_000; const EVENT_BATCH_FLUSH_MS = 16; const EVENT_BATCH_MAX_SIZE = 50; const SESSION_LOG_PAGE_LIMIT = 5_000; @@ -1256,12 +1262,75 @@ export class CloudTaskEngine extends TypedEventEmitter { const controller = new AbortController(); watcher.sseAbortController = controller; + let connectedAt = 0; + let streamWasEstablished = false; + let bytesReceived = 0; + let eventsReceived = 0; + let idleTimedOut = false; + let idleTimeoutHandle: ReturnType | null = null; + let idlePhase: "target_resolution" | "connection" | "stream" = + "target_resolution"; + + const clearIdleTimeout = () => { + if (idleTimeoutHandle) { + clearTimeout(idleTimeoutHandle); + idleTimeoutHandle = null; + } + }; + const armIdleTimeout = () => { + clearIdleTimeout(); + idleTimeoutHandle = setTimeout(() => { + idleTimedOut = true; + controller.abort(); + }, SSE_IDLE_TIMEOUT_MS); + }; + const recordIdleTimeout = (details: Record = {}) => { + const idleWatcher = this.watchers.get(key); + this.log.warn("Cloud task stream idle timeout, no bytes received", { + key, + phase: idlePhase, + idleTimeoutMs: SSE_IDLE_TIMEOUT_MS, + bytesReceived, + eventsReceived, + connectionDurationMs: streamWasEstablished + ? Date.now() - connectedAt + : 0, + ...details, + }); + if (idleWatcher) { + this.analytics.track(ANALYTICS_EVENTS.CLOUD_STREAM_IDLE_TIMEOUT, { + task_id: idleWatcher.taskId, + run_id: idleWatcher.runId, + team_id: idleWatcher.teamId, + idle_timeout_ms: SSE_IDLE_TIMEOUT_MS, + bytes_received: bytesReceived, + events_received: eventsReceived, + }); + } + }; + watcher.connStartedAt = 0; watcher.connDataEventsReceived = 0; // Resolve the read target once (proxy URL + token, or Django), reused across reconnects. if (!watcher.streamTargetResolved) { - await this.resolveStreamTarget(watcher); + armIdleTimeout(); + try { + await this.resolveStreamTarget(watcher, controller.signal); + } catch (error) { + if (!idleTimedOut) { + return; + } + recordIdleTimeout(); + await this.handleStreamCompletion(key, { + reconnectOnDisconnect: true, + reconnectError: error, + countReconnectAttempt: true, + }); + return; + } finally { + clearIdleTimeout(); + } const resolvedWatcher = this.watchers.get(key); if ( !resolvedWatcher || @@ -1340,12 +1409,11 @@ export class CloudTaskEngine extends TypedEventEmitter { // Track how long the body stayed open so healthy long-lived connections cut by churn // aren't penalized as failed reconnects (see SSE_HEALTHY_CONNECTION_MS). - let connectedAt = 0; - let streamWasEstablished = false; - let bytesReceived = 0; - let eventsReceived = 0; - + // Re-armed on every read that returns a value (data or keepalive bytes), so it only fires + // when the transport has gone completely silent, not merely between infrequent events. try { + idlePhase = "connection"; + armIdleTimeout(); // The proxy authenticates with the run-scoped Bearer token; the Django leg uses the session. const response = usingProxy ? await this.streamFetch(url.toString(), { @@ -1401,6 +1469,8 @@ export class CloudTaskEngine extends TypedEventEmitter { }); const reader = response.body.getReader(); + idlePhase = "stream"; + armIdleTimeout(); while (true) { const { done, value } = await reader.read(); @@ -1408,6 +1478,8 @@ export class CloudTaskEngine extends TypedEventEmitter { break; } + armIdleTimeout(); + if (!value) { continue; } @@ -1463,10 +1535,20 @@ export class CloudTaskEngine extends TypedEventEmitter { } catch (error) { this.flushLogBatch(key); - if (controller.signal.aborted) { + // An idle-timeout abort must fall through to the reconnect machinery below rather than + // return here like a deliberate cancel (disconnectSse/stopWatching), since nothing else + // will ever notice this connection went silent. + if (controller.signal.aborted && !idleTimedOut) { return; } + if (idleTimedOut) { + recordIdleTimeout({ + leg, + streamUrl: url.toString(), + }); + } + // Proxy-leg 401: the read token expired or its signing key rotated. Re-resolve to mint a // fresh token (or route back to Django) instead of failing. Django-leg 401 stays fatal below. const unauthorizedWatcher = this.watchers.get(key); @@ -1506,6 +1588,7 @@ export class CloudTaskEngine extends TypedEventEmitter { const isBackendError = error instanceof BackendStreamError; const wasHealthyStream = !isBackendError && + !idleTimedOut && streamWasEstablished && Date.now() - connectedAt >= SSE_HEALTHY_CONNECTION_MS; @@ -1548,6 +1631,7 @@ export class CloudTaskEngine extends TypedEventEmitter { countReconnectAttempt: !isBackendError && !wasHealthyStream, }); } finally { + clearIdleTimeout(); const currentWatcher = this.watchers.get(key); if (currentWatcher?.sseAbortController === controller) { currentWatcher.sseAbortController = null; @@ -2198,11 +2282,15 @@ export class CloudTaskEngine extends TypedEventEmitter { } } - private async resolveStreamTarget(watcher: WatcherState): Promise { + private async resolveStreamTarget( + watcher: WatcherState, + signal: AbortSignal, + ): Promise { const url = `${watcher.apiHost}/api/projects/${watcher.teamId}/tasks/${watcher.taskId}/runs/${watcher.runId}/stream_token/`; try { const response = await this.auth.authenticatedFetch(url, { method: "GET", + signal, }); if (!response.ok) { watcher.streamBaseUrl = null; @@ -2245,6 +2333,9 @@ export class CloudTaskEngine extends TypedEventEmitter { durableStream: watcher.durableStreamEnabled, }); } catch (error) { + if (signal.aborted) { + throw error; + } // Transient failure: leave unresolved so the next reconnect retries and falls back to Django. watcher.streamBaseUrl = null; watcher.streamReadToken = null; diff --git a/packages/core/src/cloud-task/cloud-task.test.ts b/packages/core/src/cloud-task/cloud-task.test.ts index 689abbda37..edd17ec0fa 100644 --- a/packages/core/src/cloud-task/cloud-task.test.ts +++ b/packages/core/src/cloud-task/cloud-task.test.ts @@ -94,6 +94,7 @@ async function waitFor( describe("CloudTaskEngine", () => { let service: CloudTaskEngine; + let analyticsMock: { track: ReturnType }; beforeEach(() => { const scopedLog = { @@ -103,7 +104,7 @@ describe("CloudTaskEngine", () => { error: vi.fn(), }; const loggerMock = { ...scopedLog, scope: vi.fn(() => scopedLog) }; - const analyticsMock = { track: vi.fn() }; + analyticsMock = { track: vi.fn() }; service = createCloudTaskEngine({ auth: mockAuthService as never, analytics: analyticsMock as never, @@ -2404,6 +2405,220 @@ describe("CloudTaskEngine", () => { ).toBe(false); }); + it("aborts and reconnects a stream that goes silent with no bytes or keepalives", async () => { + vi.useFakeTimers(); + + const updates: unknown[] = []; + service.on(CloudTaskEvent.Update, (payload) => updates.push(payload)); + + const makeInProgressRun = () => + createJsonResponse({ + id: "run-1", + status: "in_progress", + stage: null, + output: null, + error_message: null, + branch: "main", + updated_at: "2026-01-01T00:00:00Z", + }); + + mockNetFetch + .mockResolvedValueOnce(makeInProgressRun()) + .mockResolvedValueOnce( + createJsonResponse([], 200, { "X-Has-More": "false" }), + ) + .mockImplementation(() => Promise.resolve(makeInProgressRun())); + + // First connection hangs forever: no bytes, no error, no EOF, simulating a half-open + // socket (laptop sleep, NAT rebind). The second connection stays open so recovery is + // observable once the idle watchdog aborts the first. + let streamCall = 0; + const abortedFirstConnection = { value: false }; + mockStreamFetch.mockImplementation( + (_input: unknown, init?: RequestInit) => { + streamCall += 1; + if (streamCall === 1) { + const stream = new ReadableStream({ + start(controller) { + // Never enqueue or close on our own; the read() promise awaits forever until the + // idle watchdog aborts it below, mirroring how a real fetch's reader rejects once + // its AbortSignal fires. + init?.signal?.addEventListener("abort", () => { + abortedFirstConnection.value = true; + controller.error(new DOMException("Aborted", "AbortError")); + }); + }, + }); + return Promise.resolve( + new Response(stream, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + ); + } + const stream = new ReadableStream({ + start() {}, + }); + return Promise.resolve( + new Response(stream, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + ); + }, + ); + + service.watch({ + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }); + + await waitFor(() => mockStreamFetch.mock.calls.length === 1); + + // Nothing throws or EOFs; without the idle watchdog this would hang forever. + await vi.advanceTimersByTimeAsync(60_000); + expect(abortedFirstConnection.value).toBe(false); + + await vi.advanceTimersByTimeAsync(40_000); + await waitFor(() => abortedFirstConnection.value, 20_000); + await waitFor(() => mockStreamFetch.mock.calls.length >= 2, 20_000); + + expect( + analyticsMock.track.mock.calls.some( + ([eventName]) => eventName === "Cloud stream idle timeout", + ), + ).toBe(true); + + const watcher = ( + service as unknown as { + watchers: Map; + } + ).watchers.get("task-1:run-1"); + expect(watcher?.failed).toBe(false); + // Silence is a broken transport, not a healthy long-lived connection. It must consume the + // reconnect budget so a persistently silent endpoint eventually reaches the circuit breaker. + expect(watcher?.reconnectAttempts).toBe(1); + expect( + updates.some( + (u) => + typeof u === "object" && + u !== null && + (u as { kind?: string }).kind === "error", + ), + ).toBe(false); + }); + + it("times out while resolving the stream target and reconnects", async () => { + vi.useFakeTimers(); + + mockNetFetch + .mockResolvedValueOnce( + createJsonResponse({ id: "run-1", status: "in_progress" }), + ) + .mockResolvedValueOnce( + createJsonResponse([], 200, { "X-Has-More": "false" }), + ) + .mockResolvedValue( + createJsonResponse({ id: "run-1", status: "in_progress" }), + ); + + let tokenCall = 0; + const abortedResolution = { value: false }; + mockStreamTokenFetch.mockImplementation( + (_input: unknown, init?: RequestInit) => { + tokenCall += 1; + if (tokenCall === 1) { + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => { + abortedResolution.value = true; + reject(new DOMException("Aborted", "AbortError")); + }); + }); + } + return Promise.resolve( + createJsonResponse({ token: "test-token", stream_base_url: null }), + ); + }, + ); + mockStreamFetch.mockImplementation( + () => + new Promise(() => { + // The connection-phase watchdog owns this second pending request. + }), + ); + + service.watch({ + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }); + + await waitFor(() => mockStreamTokenFetch.mock.calls.length === 1); + await vi.advanceTimersByTimeAsync(100_000); + await waitFor(() => abortedResolution.value, 20_000); + await waitFor(() => mockStreamTokenFetch.mock.calls.length >= 2, 20_000); + + expect(mockStreamTokenFetch.mock.calls[0]?.[1]?.signal).toBeDefined(); + expect( + analyticsMock.track.mock.calls.some( + ([eventName]) => eventName === "Cloud stream idle timeout", + ), + ).toBe(true); + }); + + it("times out while waiting for stream response headers and reconnects", async () => { + vi.useFakeTimers(); + + mockNetFetch + .mockResolvedValueOnce( + createJsonResponse({ id: "run-1", status: "in_progress" }), + ) + .mockResolvedValueOnce( + createJsonResponse([], 200, { "X-Has-More": "false" }), + ) + .mockResolvedValue( + createJsonResponse({ id: "run-1", status: "in_progress" }), + ); + + let streamCall = 0; + const abortedConnection = { value: false }; + mockStreamFetch.mockImplementation( + (_input: unknown, init?: RequestInit) => { + streamCall += 1; + if (streamCall === 1) { + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => { + abortedConnection.value = true; + reject(new DOMException("Aborted", "AbortError")); + }); + }); + } + return Promise.resolve(createOpenSseResponse("")); + }, + ); + + service.watch({ + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }); + + await waitFor(() => mockStreamFetch.mock.calls.length === 1); + await vi.advanceTimersByTimeAsync(100_000); + await waitFor(() => abortedConnection.value, 20_000); + await waitFor(() => mockStreamFetch.mock.calls.length >= 2, 20_000); + + expect( + analyticsMock.track.mock.calls.some( + ([eventName]) => eventName === "Cloud stream idle timeout", + ), + ).toBe(true); + }); + it("stops a cloud run through the run cancel endpoint", async () => { mockNetFetch.mockResolvedValueOnce( createJsonResponse({ id: "run-1", status: "in_progress" }, 202), diff --git a/packages/shared/src/analytics-events.ts b/packages/shared/src/analytics-events.ts index eea1430b6f..82bdd241b7 100644 --- a/packages/shared/src/analytics-events.ts +++ b/packages/shared/src/analytics-events.ts @@ -327,6 +327,15 @@ export interface CloudStreamDisconnectedProperties { was_bootstrapping: boolean; } +export interface CloudStreamIdleTimeoutProperties { + task_id: string; + run_id: string; + team_id: number; + idle_timeout_ms: number; + bytes_received: number; + events_received: number; +} + // Permission events export interface PermissionRespondedProperties { task_id: string; @@ -1378,6 +1387,7 @@ export const ANALYTICS_EVENTS = { TASK_CREATION_FAILED: "Task creation failed", AGENT_SESSION_ERROR: "Agent session error", CLOUD_STREAM_DISCONNECTED: "Cloud stream disconnected", + CLOUD_STREAM_IDLE_TIMEOUT: "Cloud stream idle timeout", // Inbox events INBOX_VIEWED: "Inbox viewed", @@ -1556,6 +1566,7 @@ export type EventPropertyMap = { [ANALYTICS_EVENTS.TASK_CREATION_FAILED]: TaskCreationFailedProperties; [ANALYTICS_EVENTS.AGENT_SESSION_ERROR]: AgentSessionErrorProperties; [ANALYTICS_EVENTS.CLOUD_STREAM_DISCONNECTED]: CloudStreamDisconnectedProperties; + [ANALYTICS_EVENTS.CLOUD_STREAM_IDLE_TIMEOUT]: CloudStreamIdleTimeoutProperties; // Inbox events [ANALYTICS_EVENTS.INBOX_VIEWED]: InboxViewedProperties;