diff --git a/frontend/src/lib/logic/apiStatusLogic.test.ts b/frontend/src/lib/logic/apiStatusLogic.test.ts index 6deacf7044cc..a352544817ea 100644 --- a/frontend/src/lib/logic/apiStatusLogic.test.ts +++ b/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.' diff --git a/frontend/src/lib/logic/apiStatusLogic.ts b/frontend/src/lib/logic/apiStatusLogic.ts index 8d55e5391962..8c367fe854b2 100644 --- a/frontend/src/lib/logic/apiStatusLogic.ts +++ b/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 +// 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([ 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([ } } }, + 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 + } + }, })), ]) diff --git a/products/posthog_ai/frontend/components/RunAlertActivity.test.tsx b/products/posthog_ai/frontend/components/RunAlertActivity.test.tsx index b216fdff9c4b..07d4ce1f5618 100644 --- a/products/posthog_ai/frontend/components/RunAlertActivity.test.tsx +++ b/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() + + const retry = screen.getByText('Retry') + fireEvent.click(retry) + expect(onRetry).toHaveBeenCalledTimes(1) + }) + + it('shows no retry button without onRetry', () => { + render() + expect(screen.queryByText('Retry')).not.toBeInTheDocument() + }) }) diff --git a/products/posthog_ai/frontend/components/RunAlertActivity.tsx b/products/posthog_ai/frontend/components/RunAlertActivity.tsx index 22b9b3a93020..57f4cbd89061 100644 --- a/products/posthog_ai/frontend/components/RunAlertActivity.tsx +++ b/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 = { @@ -27,7 +29,14 @@ const TITLES: Record = { * `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 ) : null} + {onRetry ? ( + } + onClick={onRetry} + data-attr="run-stream-retry" + > + Retry + + ) : null} ) } diff --git a/products/posthog_ai/frontend/components/ThreadView.tsx b/products/posthog_ai/frontend/components/ThreadView.tsx index 2463365a1895..0c890d0ab455 100644 --- a/products/posthog_ai/frontend/components/ThreadView.tsx +++ b/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 (
- {showConnectionStatus && runConnectionState && } + {showConnectionStatus && runConnectionState && ( + + )} {showThinking && } {pullRequestUrl && } {showContextUsage && } diff --git a/products/posthog_ai/frontend/logics/runStreamLogic.test.ts b/products/posthog_ai/frontend/logics/runStreamLogic.test.ts index 62ff44960afd..e9cda947db27 100644 --- a/products/posthog_ai/frontend/logics/runStreamLogic.test.ts +++ b/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) diff --git a/products/posthog_ai/frontend/logics/runStreamLogic.ts b/products/posthog_ai/frontend/logics/runStreamLogic.ts index 867e998fe5b2..5916f38b06b7 100644 --- a/products/posthog_ai/frontend/logics/runStreamLogic.ts +++ b/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([ /** 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([ // 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([ { 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([ 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([ 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 diff --git a/products/posthog_ai/frontend/types/streamTypes.ts b/products/posthog_ai/frontend/types/streamTypes.ts index 0f59831a4c65..6ba8b15d7c39 100644 --- a/products/posthog_ai/frontend/types/streamTypes.ts +++ b/products/posthog_ai/frontend/types/streamTypes.ts @@ -39,6 +39,8 @@ export interface RunConnectionState { maxAttempts?: number /** The failed kinds: the error/crash detail to surface. */ message?: string + /** `connection_failed`: the drop is recoverable, so the card offers an in-place retry. */ + retryable?: boolean } /**