From eb082019fbe1ab3f11e65041ecf737c9fdd6512f Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Fri, 31 Jul 2026 15:22:27 +0000 Subject: [PATCH 1/3] fix(cloud-task): abort SSE reads that go silent, so watchers reconnect instead of hanging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stream-read loop in `cloud-task-engine.ts` awaited `reader.read()` with no idle deadline. A half-open socket (laptop sleep, unplugged NIC, NAT rebind) never throws and never EOFs, so the read just hangs forever — the watcher stays "connected" and never reaches the existing reconnect/backoff/error machinery. The only way out was force-quitting the app. This adds an idle watchdog that re-arms on every byte received (data or keepalive) and aborts the connection if the stream goes fully silent for 90s (a few keepalive intervals). The abort now flows into the existing reconnect logic instead of being treated as an intentional cancel, and a new `Cloud stream idle timeout` analytics event makes this failure mode visible for the first time. Generated-By: PostHog Code Task-Id: 2fd7edf4-3fde-47ab-b7e5-3564b2f6b246 --- .../core/src/cloud-task/cloud-task-engine.ts | 58 +++++++++- .../core/src/cloud-task/cloud-task.test.ts | 108 +++++++++++++++++- packages/shared/src/analytics-events.ts | 11 ++ 3 files changed, 175 insertions(+), 2 deletions(-) diff --git a/packages/core/src/cloud-task/cloud-task-engine.ts b/packages/core/src/cloud-task/cloud-task-engine.ts index 05e297f779..c2da23b8b7 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; @@ -1344,6 +1350,24 @@ export class CloudTaskEngine extends TypedEventEmitter { let streamWasEstablished = false; let bytesReceived = 0; let eventsReceived = 0; + let idleTimedOut = false; + let idleTimeoutHandle: ReturnType | null = null; + + const clearIdleTimeout = () => { + if (idleTimeoutHandle) { + clearTimeout(idleTimeoutHandle); + idleTimeoutHandle = null; + } + }; + // 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. + const armIdleTimeout = () => { + clearIdleTimeout(); + idleTimeoutHandle = setTimeout(() => { + idleTimedOut = true; + controller.abort(); + }, SSE_IDLE_TIMEOUT_MS); + }; try { // The proxy authenticates with the run-scoped Bearer token; the Django leg uses the session. @@ -1401,6 +1425,7 @@ export class CloudTaskEngine extends TypedEventEmitter { }); const reader = response.body.getReader(); + armIdleTimeout(); while (true) { const { done, value } = await reader.read(); @@ -1408,6 +1433,8 @@ export class CloudTaskEngine extends TypedEventEmitter { break; } + armIdleTimeout(); + if (!value) { continue; } @@ -1463,10 +1490,38 @@ 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) { + const idleWatcher = this.watchers.get(key); + this.log.warn("Cloud task stream idle timeout, no bytes received", { + key, + leg, + streamUrl: url.toString(), + idleTimeoutMs: SSE_IDLE_TIMEOUT_MS, + bytesReceived, + eventsReceived, + connectionDurationMs: streamWasEstablished + ? Date.now() - connectedAt + : 0, + }); + 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, + }); + } + } + // 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); @@ -1548,6 +1603,7 @@ export class CloudTaskEngine extends TypedEventEmitter { countReconnectAttempt: !isBackendError && !wasHealthyStream, }); } finally { + clearIdleTimeout(); const currentWatcher = this.watchers.get(key); if (currentWatcher?.sseAbortController === controller) { currentWatcher.sseAbortController = 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..35083b7470 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,111 @@ 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 and delivers a + // keepalive so recovery is observable once the idle watchdog aborts the first. + let streamCall = 0; + const encoder = new TextEncoder(); + 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(controller) { + controller.enqueue( + encoder.encode('event: keepalive\ndata: {"type":"keepalive"}\n\n'), + ); + }, + }); + 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); + expect( + updates.some( + (u) => + typeof u === "object" && + u !== null && + (u as { kind?: string }).kind === "error", + ), + ).toBe(false); + }); + 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; From 7137b92132e6848fdaf80e8606f315208a4c71a7 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Fri, 31 Jul 2026 16:30:03 +0100 Subject: [PATCH 2/3] fix(cloud-task): format idle stream recovery test Generated-By: PostHog Code Task-Id: 92b56d91-0d15-416f-8ab9-2b32e3cb3e34 --- .../core/src/cloud-task/cloud-task.test.ts | 54 ++++++++++--------- 1 file changed, 29 insertions(+), 25 deletions(-) diff --git a/packages/core/src/cloud-task/cloud-task.test.ts b/packages/core/src/cloud-task/cloud-task.test.ts index 35083b7470..929f07bdee 100644 --- a/packages/core/src/cloud-task/cloud-task.test.ts +++ b/packages/core/src/cloud-task/cloud-task.test.ts @@ -2435,18 +2435,35 @@ describe("CloudTaskEngine", () => { let streamCall = 0; const encoder = new TextEncoder(); const abortedFirstConnection = { value: false }; - mockStreamFetch.mockImplementation((_input: unknown, init?: RequestInit) => { - streamCall += 1; - if (streamCall === 1) { + 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(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")); - }); + controller.enqueue( + encoder.encode( + 'event: keepalive\ndata: {"type":"keepalive"}\n\n', + ), + ); }, }); return Promise.resolve( @@ -2455,21 +2472,8 @@ describe("CloudTaskEngine", () => { headers: { "Content-Type": "text/event-stream" }, }), ); - } - const stream = new ReadableStream({ - start(controller) { - controller.enqueue( - encoder.encode('event: keepalive\ndata: {"type":"keepalive"}\n\n'), - ); - }, - }); - return Promise.resolve( - new Response(stream, { - status: 200, - headers: { "Content-Type": "text/event-stream" }, - }), - ); - }); + }, + ); service.watch({ taskId: "task-1", From 54a524b1b73063affc6f06277d39750fe10bc1b6 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Fri, 31 Jul 2026 17:04:51 +0100 Subject: [PATCH 3/3] fix(cloud-task): cover silent stream setup Generated-By: PostHog Code Task-Id: 92b56d91-0d15-416f-8ab9-2b32e3cb3e34 --- .../core/src/cloud-task/cloud-task-engine.ts | 119 ++++++++++------ .../core/src/cloud-task/cloud-task.test.ts | 127 ++++++++++++++++-- 2 files changed, 193 insertions(+), 53 deletions(-) diff --git a/packages/core/src/cloud-task/cloud-task-engine.ts b/packages/core/src/cloud-task/cloud-task-engine.ts index c2da23b8b7..abb90c5789 100644 --- a/packages/core/src/cloud-task/cloud-task-engine.ts +++ b/packages/core/src/cloud-task/cloud-task-engine.ts @@ -1262,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 || @@ -1346,30 +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; - let idleTimedOut = false; - let idleTimeoutHandle: ReturnType | null = null; - - const clearIdleTimeout = () => { - if (idleTimeoutHandle) { - clearTimeout(idleTimeoutHandle); - idleTimeoutHandle = null; - } - }; // 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. - const armIdleTimeout = () => { - clearIdleTimeout(); - idleTimeoutHandle = setTimeout(() => { - idleTimedOut = true; - controller.abort(); - }, SSE_IDLE_TIMEOUT_MS); - }; - 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(), { @@ -1425,6 +1469,7 @@ export class CloudTaskEngine extends TypedEventEmitter { }); const reader = response.body.getReader(); + idlePhase = "stream"; armIdleTimeout(); while (true) { @@ -1498,28 +1543,10 @@ export class CloudTaskEngine extends TypedEventEmitter { } if (idleTimedOut) { - const idleWatcher = this.watchers.get(key); - this.log.warn("Cloud task stream idle timeout, no bytes received", { - key, + recordIdleTimeout({ leg, streamUrl: url.toString(), - idleTimeoutMs: SSE_IDLE_TIMEOUT_MS, - bytesReceived, - eventsReceived, - connectionDurationMs: streamWasEstablished - ? Date.now() - connectedAt - : 0, }); - 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, - }); - } } // Proxy-leg 401: the read token expired or its signing key rotated. Re-resolve to mint a @@ -1561,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; @@ -2254,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; @@ -2301,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 929f07bdee..edd17ec0fa 100644 --- a/packages/core/src/cloud-task/cloud-task.test.ts +++ b/packages/core/src/cloud-task/cloud-task.test.ts @@ -2430,10 +2430,9 @@ describe("CloudTaskEngine", () => { .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 and delivers a - // keepalive so recovery is observable once the idle watchdog aborts the first. + // 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 encoder = new TextEncoder(); const abortedFirstConnection = { value: false }; mockStreamFetch.mockImplementation( (_input: unknown, init?: RequestInit) => { @@ -2458,13 +2457,7 @@ describe("CloudTaskEngine", () => { ); } const stream = new ReadableStream({ - start(controller) { - controller.enqueue( - encoder.encode( - 'event: keepalive\ndata: {"type":"keepalive"}\n\n', - ), - ); - }, + start() {}, }); return Promise.resolve( new Response(stream, { @@ -2500,10 +2493,13 @@ describe("CloudTaskEngine", () => { const watcher = ( service as unknown as { - watchers: Map; + 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) => @@ -2514,6 +2510,115 @@ describe("CloudTaskEngine", () => { ).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),