fix(tasks): recover from a dropped agent stream instead of dead-ending

frontend/src/lib/logic/apiStatusLogic.test.ts

@@ -80,6 +80,59 @@ describe('apiStatusLogic', () => {
         })
     })
 
+    describe('internet connection recovery', () => {
+        // Unmount so beforeUnmount clears the re-probe interval a set(true) starts.
+        afterEach(() => logic?.unmount())
+
+        it('sets the issue on a "Failed to fetch" error', async () => {
+            initKeaTests()
+            logic = apiStatusLogic()
+            logic.mount()
+
+            await expectLogic(logic, () => {
+                logic.actions.onApiResponse(undefined, new Error('Failed to fetch'))
+            })
+                .toDispatchActions(['setInternetConnectionIssue'])
+                .toMatchValues({ internetConnectionIssue: true })
+        })
+
+        it('clears the issue when a re-probe reaches the server', async () => {
+            useMocks({
+                get: {
+                    '/api/users/@me/': () => [200, MOCK_DEFAULT_USER],
+                },
+            })
+            initKeaTests()
+            logic = apiStatusLogic()
+            logic.mount()
+            logic.actions.setInternetConnectionIssue(true)
+
+            await expectLogic(logic, () => {
+                logic.actions.probeInternetConnection()
+            })
+                .toDispatchActions([logic.actionCreators.setInternetConnectionIssue(false)])
+                .toMatchValues({ internetConnectionIssue: false })
+        })
+
+        it('re-probes on a browser online event', async () => {
+            useMocks({
+                get: {
+                    '/api/users/@me/': () => [200, MOCK_DEFAULT_USER],
+                },
+            })
+            initKeaTests()
+            logic = apiStatusLogic()
+            logic.mount()
+            logic.actions.setInternetConnectionIssue(true)
+
+            await expectLogic(logic, () => {
+                window.dispatchEvent(new Event('online'))
+            })
+                .toDispatchActions(['probeInternetConnection'])
+                .toMatchValues({ internetConnectionIssue: false })
+        })
+    })
+
     describe('read-only impersonation 403 handling', () => {
         const READ_ONLY_DETAIL = 'This action is not allowed during read-only user impersonation.'
 

frontend/src/lib/logic/apiStatusLogic.ts

@@ -1,4 +1,4 @@
-import { MakeLogicType, actions, kea, listeners, path, reducers } from 'kea'
+import { MakeLogicType, actions, events, kea, listeners, path, reducers } from 'kea'
 
 import { lemonToast } from '@posthog/lemon-ui'
 
@@ -26,6 +26,9 @@ export interface apiStatusLogicActions {
     setInternetConnectionIssue: (issue: boolean) => {
         issue: boolean
     }
+    probeInternetConnection: () => {
+        value: true
+    }
     setTimeSensitiveAuthenticationRequired: (required: boolean | [onSuccess: () => void, onFailure: () => void]) => {
         required: boolean | [onSuccess: () => void, onFailure: () => void]
     }
@@ -39,11 +42,17 @@ export interface apiStatusLogicActions {
 
 export type apiStatusLogicType = MakeLogicType<apiStatusLogicValues, apiStatusLogicActions>
 
+// How often to re-probe the server while the connection-issue banner is up. Nothing else clears the
+// banner on an SSE-only page (there may be no further ordinary API calls), so the probe is the only
+// path back for those users.
+const INTERNET_REPROBE_INTERVAL_MS = 5000
+
 export const apiStatusLogic = kea<apiStatusLogicType>([
     path(['lib', 'apiStatusLogic']),
     actions({
         onApiResponse: (response?: Response, error?: any) => ({ response, error }),
         setInternetConnectionIssue: (issue: boolean) => ({ issue }),
+        probeInternetConnection: true,
         setTimeSensitiveAuthenticationRequired: (
             required: boolean | [onSuccess: () => void, onFailure: () => void]
         ) => ({
@@ -189,5 +198,54 @@ export const apiStatusLogic = kea<apiStatusLogicType>([
                 }
             }
         },
+        setInternetConnectionIssue: ({ issue }) => {
+            // The banner used to clear only on the next successful API response. On an SSE-driven page
+            // that call may never come, so drive recovery here instead: while the issue is up, re-probe
+            // the server on a timer and clear as soon as it answers.
+            if (issue) {
+                if (cache.reprobeInterval === undefined) {
+                    cache.reprobeInterval = window.setInterval(() => {
+                        actions.probeInternetConnection()
+                    }, INTERNET_REPROBE_INTERVAL_MS)
+                }
+            } else if (cache.reprobeInterval !== undefined) {
+                window.clearInterval(cache.reprobeInterval)
+                cache.reprobeInterval = undefined
+            }
+        },
+        probeInternetConnection: async () => {
+            if (!values.internetConnectionIssue) {
+                return
+            }
+            try {
+                await api.get('api/users/@me/')
+                // onApiResponse clears on an ok response; clear here too to cover a resolved response.
+                actions.setInternetConnectionIssue(false)
+            } catch (error: any) {
+                // A bare 'Failed to fetch' means still unreachable — keep the banner for the next tick.
+                // Any other error carries an HTTP status, so we reached the server: the connection is back.
+                if (error?.message !== 'Failed to fetch') {
+                    actions.setInternetConnectionIssue(false)
+                }
+            }
+        },
+    })),
+    events(({ actions, cache }) => ({
+        afterMount: () => {
+            // A browser 'online' event only fires for a true offline→online transition (not for a
+            // CORS/server 'Failed to fetch' while navigator stays online), so it complements the timer
+            // rather than replacing it. Probe rather than clear blindly — only reaching the server clears.
+            cache.onOnline = () => actions.probeInternetConnection()
+            window.addEventListener('online', cache.onOnline)
+        },
+        beforeUnmount: () => {
+            if (cache.onOnline) {
+                window.removeEventListener('online', cache.onOnline)
+            }
+            if (cache.reprobeInterval !== undefined) {
+                window.clearInterval(cache.reprobeInterval)
+                cache.reprobeInterval = undefined
+            }
+        },
     })),
 ])

products/posthog_ai/frontend/components/RunAlertActivity.test.tsx

@@ -1,6 +1,6 @@
 import '@testing-library/jest-dom'
 
-import { cleanup, render, screen } from '@testing-library/react'
+import { cleanup, fireEvent, render, screen } from '@testing-library/react'
 
 import { RunAlertActivity } from './RunAlertActivity'
 
@@ -24,4 +24,18 @@ describe('RunAlertActivity', () => {
         expect(screen.getByText(title)).toBeInTheDocument()
         expect(screen.getByText('boom')).toBeInTheDocument()
     })
+
+    it('offers an in-place retry when onRetry is set', () => {
+        const onRetry = jest.fn()
+        render(<RunAlertActivity kind="connection_failed" message="boom" onRetry={onRetry} />)
+
+        const retry = screen.getByText('Retry')
+        fireEvent.click(retry)
+        expect(onRetry).toHaveBeenCalledTimes(1)
+    })
+
+    it('shows no retry button without onRetry', () => {
+        render(<RunAlertActivity kind="connection_failed" message="boom" />)
+        expect(screen.queryByText('Retry')).not.toBeInTheDocument()
+    })
 })

products/posthog_ai/frontend/components/RunAlertActivity.tsx

@@ -1,5 +1,5 @@
-import { IconWarning } from '@posthog/icons'
-import { Spinner } from '@posthog/lemon-ui'
+import { IconRefresh, IconWarning } from '@posthog/icons'
+import { LemonButton, Spinner } from '@posthog/lemon-ui'
 
 import { MarkdownMessage } from '../messages/MarkdownMessage'
 import type { RunAlertKind, RunConnectionState } from '../types/streamTypes'
@@ -8,6 +8,8 @@ import { Activity } from './ActivityPrimitives'
 interface RunAlertActivityProps extends RunConnectionState {
     /** Stable id for the underlying `Activity` (drives markdown substep ids). Defaults per kind. */
     id?: string
+    /** `connection_failed`: when set, the card shows a Retry button that reopens the stream in place. */
+    onRetry?: () => void
 }
 
 const TITLES: Record<RunAlertKind, string> = {
@@ -27,7 +29,14 @@ const TITLES: Record<RunAlertKind, string> = {
  * `Activity` body auto-collapses, so the detail message rides the always-visible `children` region (mirrors
  * `ToolActivity`'s failed-error pattern) rather than the collapsible `details`.
  */
-export function RunAlertActivity({ kind, id, attempt, maxAttempts, message }: RunAlertActivityProps): JSX.Element {
+export function RunAlertActivity({
+    kind,
+    id,
+    attempt,
+    maxAttempts,
+    message,
+    onRetry,
+}: RunAlertActivityProps): JSX.Element {
     const activityId = id ?? `run-alert-${kind}`
 
     if (kind === 'reconnecting') {
@@ -57,6 +66,17 @@ export function RunAlertActivity({ kind, id, attempt, maxAttempts, message }: Ru
                     <MarkdownMessage content={message} id={`${activityId}-message`} />
                 </div>
             ) : null}
+            {onRetry ? (
+                <LemonButton
+                    type="secondary"
+                    size="small"
+                    icon={<IconRefresh />}
+                    onClick={onRetry}
+                    data-attr="run-stream-retry"
+                >
+                    Retry
+                </LemonButton>
+            ) : null}
         </Activity>
     )
 }

products/posthog_ai/frontend/components/ThreadView.tsx

@@ -1,4 +1,4 @@
-import { useValues } from 'kea'
+import { useActions, useValues } from 'kea'
 import { type ReactNode, memo, useCallback, useEffect, useMemo, useState } from 'react'
 
 import { inStorybookTestRunner } from 'lib/utils/dom'
@@ -269,11 +269,17 @@ const ThreadFooter = memo(function ThreadFooter({
     // `runConnectionState` is self-subscribed here (like `currentProgress`) so the frequently-updating
     // reconnect attempt counter stays isolated to this leaf and never destabilizes `ThreadView`'s footer.
     const { currentProgress, runConnectionState } = useValues(runStreamLogic)
+    const { retryConnection } = useActions(runStreamLogic)
     // `gap-1.5` matches the thread's inter-row gap (`VirtualizedThread`'s `gap` default) so stacked footer
     // items keep the same vertical rhythm as the thread.
     return (
         <div className="flex flex-col gap-1.5">
-            {showConnectionStatus && runConnectionState && <RunAlertActivity {...runConnectionState} />}
+            {showConnectionStatus && runConnectionState && (
+                <RunAlertActivity
+                    {...runConnectionState}
+                    onRetry={runConnectionState.retryable ? retryConnection : undefined}
+                />
+            )}
             {showThinking && <ThinkingIndicator progress={currentProgress} phase={thinkingPhase} />}
             {pullRequestUrl && <PullRequestCard prUrl={pullRequestUrl} branch={prBranch} />}
             {showContextUsage && <ContextUsageBar />}

products/posthog_ai/frontend/logics/runStreamLogic.test.ts

@@ -2930,6 +2930,53 @@ describe('runStreamLogic', () => {
             })
         })
 
+        it('captures an exception on a retryable exhaustion so error tracking sees it', () => {
+            const exceptionSpy = jest.spyOn(posthog, 'captureException').mockImplementation(() => undefined as any)
+
+            logic.actions.handleStreamError({ errorTitle: 'Cloud stream failed', retryable: true })
+
+            expect(exceptionSpy).toHaveBeenCalledTimes(1)
+            expect(exceptionSpy.mock.calls[0][0]).toBeInstanceOf(Error)
+            exceptionSpy.mockRestore()
+        })
+
+        it('does not capture an exception for a non-retryable failure', () => {
+            const exceptionSpy = jest.spyOn(posthog, 'captureException').mockImplementation(() => undefined as any)
+
+            logic.actions.handleStreamError({ errorTitle: 'No current project', retryable: false })
+
+            expect(exceptionSpy).not.toHaveBeenCalled()
+            exceptionSpy.mockRestore()
+        })
+
+        it('marks a retryable connection_failed banner as retryable', () => {
+            logic.actions.handleStreamError({ errorTitle: 'Cloud stream failed', retryable: true })
+            expect(logic.values.runConnectionState).toEqual(
+                expect.objectContaining({ kind: 'connection_failed', retryable: true })
+            )
+        })
+
+        it('reopens the stream from the cursor on retryConnection and resets the reconnect budget', async () => {
+            jest.spyOn(api.tasks.runs, 'get').mockResolvedValue({ status: 'in_progress' } as any)
+            jest.spyOn(posthog, 'capture').mockImplementation(() => undefined as any)
+            jest.spyOn(posthog, 'captureException').mockImplementation(() => undefined as any)
+
+            logic.actions.openSseForRun({ taskId: 'task-1', runId: 'run-1', traceId: 'trace-1' })
+            logic.actions.sseReconnecting(MAX_SSE_RECONNECT_ATTEMPTS)
+            logic.actions.sseDropped()
+            await flushPromises()
+            expect(logic.values.sseStatus).toEqual('error')
+
+            await expectLogic(logic, () => {
+                logic.actions.retryConnection()
+            }).toDispatchActions([
+                logic.actionCreators.openSseForRun({ taskId: 'task-1', runId: 'run-1', startLatest: true }),
+            ])
+
+            expect(logic.values.reconnectAttempt).toEqual(0)
+            expect(logic.values.cumulativeReconnectAttempt).toEqual(0)
+        })
+
         it('retries the snapshot before teardown and reports was_bootstrapping=true on exhaustion', async () => {
             const captureSpy = jest.spyOn(posthog, 'capture').mockImplementation(() => undefined as any)
             jest.spyOn(api.tasks.runs, 'get').mockResolvedValue({ status: 'in_progress' } as any)

products/posthog_ai/frontend/logics/runStreamLogic.ts

@@ -1465,6 +1465,9 @@ export interface runStreamLogicActions {
     closeSse: () => {
         value: true
     }
+    retryConnection: () => {
+        value: true
+    }
     handleStreamError: (envelope: StreamErrorEnvelope) => StreamErrorEnvelope
     handleTerminalStatus: (status: {
         errorMessage?: string | null
@@ -1709,6 +1712,11 @@ export const runStreamLogic = kea<runStreamLogicType>([
         /** Internal: the live run history snapshot finished loading or was intentionally skipped. */
         bootstrapLogReady: true,
         closeSse: true,
+        /**
+         * User-initiated retry after the reconnect budget ran out: reopens the SSE for the active run
+         * from the last-seen cursor and hands recovery back to the reconnect loop.
+         */
+        retryConnection: true,
         /**
          * The conversations/open POST is in flight — drives the optimistic "spinning up" indicator
          * before any SSE state exists. The caller (maxThreadLogic) flips it on before the POST and off
@@ -1866,6 +1874,8 @@ export const runStreamLogic = kea<runStreamLogicType>([
                 // A successful (re)connection clears the counter; bootstrapping a run starts fresh.
                 sseOpened: () => 0,
                 bootstrapRun: () => 0,
+                // A manual retry starts the per-drop budget over.
+                retryConnection: () => 0,
                 reset: () => 0,
             },
         ],
@@ -1877,6 +1887,8 @@ export const runStreamLogic = kea<runStreamLogicType>([
             {
                 sseReconnecting: (state) => state + 1,
                 bootstrapRun: () => 0,
+                // A manual retry starts the cumulative budget over.
+                retryConnection: () => 0,
                 reset: () => 0,
             },
         ],
@@ -2326,7 +2338,11 @@ export const runStreamLogic = kea<runStreamLogicType>([
                     const detail = bootstrapError
                         ? [bootstrapError.errorTitle, bootstrapError.errorMessage].filter(Boolean).join(' — ')
                         : undefined
-                    return { kind: 'connection_failed', message: detail || undefined }
+                    return {
+                        kind: 'connection_failed',
+                        message: detail || undefined,
+                        retryable: bootstrapError?.retryable ?? false,
+                    }
                 }
                 return null
             },
@@ -3022,6 +3038,33 @@ export const runStreamLogic = kea<runStreamLogicType>([
                 was_bootstrapping: cache.isBootstrapping === true,
                 execution_type: 'sandbox',
             })
+            // A retryable failure means the reconnect budget ran out while the run was probably still
+            // alive in the sandbox — the client just stopped watching. The analytics event above alone
+            // left this invisible in error tracking, so also capture an exception to make it measurable.
+            if (retryable) {
+                posthog.captureException(new Error(`Task run stream gave up: ${errorTitle}`), {
+                    conversation_id: props.conversationId,
+                    trace_id: values.traceId,
+                    run_id: activeRun?.runId,
+                    task_id: activeRun?.taskId,
+                    reconnect_attempts: values.reconnectAttempt,
+                    cumulative_reconnect_attempts: values.cumulativeReconnectAttempt,
+                })
+            }
+        },
+        retryConnection: () => {
+            const activeRun = cache.activeRun as { taskId: string; runId: string } | undefined
+            if (activeRun) {
+                // Resume from the last-seen cursor (openSseForRun reads cache.lastEventId) so the backend
+                // replays only the frames after it — no gap, no re-broadcast. The reducers cleared the
+                // reconnect budget, so the backoff loop owns recovery again from a fresh count.
+                actions.openSseForRun({ taskId: activeRun.taskId, runId: activeRun.runId, startLatest: true })
+                return
+            }
+            // The stream never opened (the bootstrap fetch itself failed) — re-run bootstrap from scratch.
+            if (values.bootstrappedTaskId && values.bootstrappedRunId) {
+                actions.bootstrapRun({ taskId: values.bootstrappedTaskId, runId: values.bootstrappedRunId })
+            }
         },
         closeSse: () => {
             cache.activeRun = undefined