fix(ai): bound silent stream recovery and surface a terminal CONNECTION_LOST state (#22486)

## Rationale Two gaps in the keep-alive recovery (`AgentChatStreamKeepAliveEffect`): 1. It only engages when `isStreaming` is already true — a socket that dies **before the first chunk** leaves the user waiting forever with no recovery path (CONFIRMED-high in the chat-stack audit; the window where Sentry shows failures concentrate). 2. When it does engage, it retries **silently forever** — a genuinely dead connection means an infinite spinner with the user none the wiser. ## Why this is the root cause, not a symptom patch Recovery must be gated on "a response is owed" — which since #22485 is `isStreaming || isAwaitingFirstChunk`, closing gap 1 with the state that actually models the window rather than a timer heuristic. For gap 2, unbounded retry hides a terminal condition; the fix is an honest state machine: 3 silent recoveries (resubscribe + refetch), then a client-only `CONNECTION_LOST` error. Two deliberate choices from the audit: - **No Retry button** on `CONNECTION_LOST` — it's semantically forced, not cosmetic: Retry calls `retryLastFailedTurn`, which requires a persisted `lastStreamError`; after a mere connection loss the server has no failed turn (the stream is likely still running or completed server-side), so Retry would deterministically throw `NO_FAILED_TURN_TO_RETRY`. - **Auto-clear instead of dead-end**: the moment events flow again (SSE reconnect, refetch delivering data), the `CONNECTION_LOST` error clears itself — the state is "connection lost", not "turn failed", and it self-heals when the connection returns. ## User impact A dead connection pre-first-token currently means waiting forever; mid-stream it means silent infinite recovery. Now: three quiet recovery attempts (which fix the transient cases invisibly), then a truthful message, which disappears on its own when connectivity returns — and the server-side answer is intact all along, delivered by the next successful refetch. ## Stack Based on #22485 (pending indicator) — reads the awaiting-first-chunk state. Chain: #22484 → #22485 → this. ## Test plan - [ ] CI green - [ ] Manual: kill the network pre-first-token → 3 recoveries → CONNECTION_LOST; restore network → error clears, transcript catches up https://claude.ai/code/session_01Lyi6zTema2FMVVh8MD6c38 --- _Generated by [Claude Code](https://claude.ai/code/session_01Lyi6zTema2FMVVh8MD6c38)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22486?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-07-02 21:20:45 +02:00
parent 77529191f8
commit 4a8679327b
4 changed files with 88 additions and 9 deletions
@@ -5,16 +5,24 @@ import { isDefined, isValidUuid } from 'twenty-shared/utils';
 import { AGENT_CHAT_REFETCH_MESSAGES_EVENT_NAME } from '@/ai/constants/AgentChatRefetchMessagesEventName';
 import { AGENT_CHAT_STREAM_LIVENESS_CHECK_INTERVAL_IN_MS } from '@/ai/constants/AgentChatStreamLivenessCheckIntervalInMs';
 import { AGENT_CHAT_STREAM_LIVENESS_TIMEOUT_IN_MS } from '@/ai/constants/AgentChatStreamLivenessTimeoutInMs';
+import { agentChatErrorComponentFamilyState } from '@/ai/states/agentChatErrorComponentFamilyState';
+import { agentChatIsAwaitingFirstChunkComponentFamilyState } from '@/ai/states/agentChatIsAwaitingFirstChunkComponentFamilyState';
 import { agentChatIsStreamingComponentFamilyState } from '@/ai/states/agentChatIsStreamingComponentFamilyState';
 import { agentChatStreamLastEventTimestampState } from '@/ai/states/agentChatStreamLastEventTimestampState';
+import { agentChatStreamRecoveryAttemptsState } from '@/ai/states/agentChatStreamRecoveryAttemptsState';
 import { agentChatStreamResubscribeNonceState } from '@/ai/states/agentChatStreamResubscribeNonceState';
 import { currentAiChatThreadState } from '@/ai/states/currentAiChatThreadState';
+import { AiChatErrorCode } from '@/ai/utils/aiChatErrorCode';
+import { createAiChatCodedError } from '@/ai/utils/createAiChatCodedError';
 import { useListenToBrowserEvent } from '@/browser-event/hooks/useListenToBrowserEvent';
+import { isGraphqlErrorOfType } from '~/utils/is-graphql-error-of-type.util';
 import { dispatchBrowserEvent } from '@/browser-event/utils/dispatchBrowserEvent';
 import { SSE_CLIENT_RECONNECTED_EVENT_NAME } from '@/sse-db-event/constants/SseClientReconnectedEventName';
 import { useAtomComponentFamilyStateCallbackState } from '@/ui/utilities/state/jotai/hooks/useAtomComponentFamilyStateCallbackState';
 import { useAtomStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomStateValue';

+const MAX_SILENT_RECOVERY_ATTEMPTS = 3;
+
 export const AgentChatStreamKeepAliveEffect = () => {
  const store = useStore();
  const currentAiChatThread = useAtomStateValue(currentAiChatThreadState);
@@ -22,23 +30,64 @@ export const AgentChatStreamKeepAliveEffect = () => {
  const isStreamingFamilyCallback = useAtomComponentFamilyStateCallbackState(
    agentChatIsStreamingComponentFamilyState,
  );
+  const isAwaitingFirstChunkFamilyCallback =
+    useAtomComponentFamilyStateCallbackState(
+      agentChatIsAwaitingFirstChunkComponentFamilyState,
+    );
+  const errorFamilyCallback = useAtomComponentFamilyStateCallbackState(
+    agentChatErrorComponentFamilyState,
+  );

  const hasActiveSubscription =
    isDefined(currentAiChatThread) && isValidUuid(currentAiChatThread);

-  const recoverStreamIfStreaming = useCallback(() => {
-    const isStreaming = store.get(
-      isStreamingFamilyCallback({ threadId: currentAiChatThread }),
+  const recoverStreamIfStalled = useCallback(() => {
+    const familyKey = { threadId: currentAiChatThread };
+    const isStreaming = store.get(isStreamingFamilyCallback(familyKey));
+    const isAwaitingFirstChunk = store.get(
+      isAwaitingFirstChunkFamilyCallback(familyKey),
    );

-    if (!isStreaming) {
+    if (!isStreaming && !isAwaitingFirstChunk) {
+      store.set(agentChatStreamRecoveryAttemptsState.atom, 0);
+
      return;
    }

+    const recoveryAttempts = store.get(
+      agentChatStreamRecoveryAttemptsState.atom,
+    );
+
+    if (recoveryAttempts >= MAX_SILENT_RECOVERY_ATTEMPTS) {
+      store.set(
+        errorFamilyCallback(familyKey),
+        createAiChatCodedError(
+          'Connection to the assistant was lost. Reload to see the response.',
+          AiChatErrorCode.CONNECTION_LOST,
+        ),
+      );
+      store.set(isStreamingFamilyCallback(familyKey), false);
+      store.set(isAwaitingFirstChunkFamilyCallback(familyKey), false);
+      store.set(agentChatStreamRecoveryAttemptsState.atom, 0);
+      store.set(agentChatStreamLastEventTimestampState.atom, Date.now());
+
+      return;
+    }
+
+    store.set(
+      agentChatStreamRecoveryAttemptsState.atom,
+      (attempts) => attempts + 1,
+    );
    store.set(agentChatStreamLastEventTimestampState.atom, Date.now());
    store.set(agentChatStreamResubscribeNonceState.atom, (nonce) => nonce + 1);
    dispatchBrowserEvent(AGENT_CHAT_REFETCH_MESSAGES_EVENT_NAME);
-  }, [store, isStreamingFamilyCallback, currentAiChatThread]);
+  }, [
+    store,
+    isStreamingFamilyCallback,
+    isAwaitingFirstChunkFamilyCallback,
+    errorFamilyCallback,
+    currentAiChatThread,
+  ]);

  useEffect(() => {
    if (!hasActiveSubscription) {
@@ -57,18 +106,38 @@ export const AgentChatStreamKeepAliveEffect = () => {
      const timeSinceLastEventInMs = Date.now() - lastTimestamp;

      if (timeSinceLastEventInMs <= AGENT_CHAT_STREAM_LIVENESS_TIMEOUT_IN_MS) {
+        store.set(agentChatStreamRecoveryAttemptsState.atom, 0);
+
+        const errorAtom = errorFamilyCallback({
+          threadId: currentAiChatThread,
+        });
+        const currentError = store.get(errorAtom);
+
+        if (
+          isDefined(currentError) &&
+          isGraphqlErrorOfType(currentError, AiChatErrorCode.CONNECTION_LOST)
+        ) {
+          store.set(errorAtom, null);
+        }
+
        return;
      }

-      recoverStreamIfStreaming();
+      recoverStreamIfStalled();
    }, AGENT_CHAT_STREAM_LIVENESS_CHECK_INTERVAL_IN_MS);

    return () => clearInterval(interval);
-  }, [hasActiveSubscription, store, recoverStreamIfStreaming]);
+  }, [
+    hasActiveSubscription,
+    store,
+    recoverStreamIfStalled,
+    errorFamilyCallback,
+    currentAiChatThread,
+  ]);

  useListenToBrowserEvent({
    eventName: SSE_CLIENT_RECONNECTED_EVENT_NAME,
-    onBrowserEvent: recoverStreamIfStreaming,
+    onBrowserEvent: recoverStreamIfStalled,
  });

  return null;
@@ -22,5 +22,9 @@ export const AiChatErrorRenderer = ({
    return <AiChatApiKeyNotConfiguredMessage />;
  }

+  if (isGraphqlErrorOfType(error, AiChatErrorCode.CONNECTION_LOST)) {
+    return <AiChatErrorMessage error={error} />;
+  }
+
  return <AiChatErrorMessage error={error} onRetry={onRetry} />;
 };
@@ -0,0 +1,6 @@
+import { createAtomState } from '@/ui/utilities/state/jotai/utils/createAtomState';
+
+export const agentChatStreamRecoveryAttemptsState = createAtomState<number>({
+  key: 'agentChatStreamRecoveryAttemptsState',
+  defaultValue: 0,
+});
@@ -1,5 +1,5 @@
-// Error codes matching backend AgentExceptionCode and BillingExceptionCode
 export const AiChatErrorCode = {
  BILLING_CREDITS_EXHAUSTED: 'BILLING_CREDITS_EXHAUSTED',
  API_KEY_NOT_CONFIGURED: 'API_KEY_NOT_CONFIGURED',
+  CONNECTION_LOST: 'CONNECTION_LOST',
 } as const;