feat(app-expo): conversation WS warmup, background pool, lifecycle

Prefetch opening over WebSocket from the conversations list before navigation, with prepared-session handoff into the chat screen. Add a single-slot background pool so leaving chat (in-app) keeps the last session socket with UI callbacks stripped; dispose on app background and reconnect after resume when the chat screen is mounted. Tear down pooled sockets on logout, purge, and conversation delete. RealtimeSession supports attachUiCallbacks and idempotent dispose, and the chat composer hides the connection notice while connecting if assistant history already exists. Fix pause handler wiring in the conversation screen. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-06 14:54:47 +08:00
parent 7ad52fce89
commit 5dac3efd52
9 changed files with 381 additions and 30 deletions
--- a/app-expo/src/app/(main)/conversation/[id].tsx
+++ b/app-expo/src/app/(main)/conversation/[id].tsx
@@ -1084,6 +1084,11 @@ export default function ConversationScreen() {
  );

  const { data: messages } = useMessages(id);
+  const hasAssistantInHistory = useMemo(
+    () => (messages ?? []).some((m) => m.senderType === 'assistant'),
+    [messages],
+  );
+
  const ttsGate = useRef(createTtsPlaybackGate());
  const {
    enqueue,
@@ -1186,6 +1191,7 @@ export default function ConversationScreen() {
    [enqueueExclusive],
  );

+  const handlePauseAssistantPlayback = useCallback(() => {
    pausePlayback();
  }, [pausePlayback]);

@@ -1448,7 +1454,9 @@ export default function ConversationScreen() {
        ? t('connectionConnecting')
        : t('connectionDisconnected');
  const showConnectionBadge = __DEV__;
-  const showConnectionNotice = connectionState !== 'connected';
+  const showConnectionNotice =
+    connectionState !== 'connected' &&
+    !(connectionState === 'connecting' && hasAssistantInHistory);
  const connectionNoticeText =
    connectionState === 'connecting'
      ? t('chatUnavailableConnecting')
--- a/app-expo/src/app/(tabs)/index.tsx
+++ b/app-expo/src/app/(tabs)/index.tsx
@@ -22,6 +22,11 @@ import { Text } from '@/components/ui/text';
 import { NetworkError } from '@/core/api/types';
 import { useTypography } from '@/core/typography-context';
 import { conversationApi } from '@/features/conversation/api';
+import {
+  prefetchConversationMessages,
+  warmupConversationOpening,
+} from '@/features/conversation/entry-warmup';
+import { abandonPreparedRealtimeSession } from '@/features/conversation/prepared-session-registry';
 import {
  useConversations,
  useCreateConversation,
@@ -88,9 +93,11 @@ function GreetingCardSkeleton() {
 function ConversationCard({
  item,
  onPress,
+  disabled,
 }: {
  item: ConversationListItem;
  onPress: () => void;
+  disabled?: boolean;
 }) {
  const { t } = useTranslation('conversation');
  const typography = useTypography();
@@ -140,6 +147,7 @@ function ConversationCard({
    <Pressable
      className="flex-row items-start gap-6 rounded-xl border border-border bg-card p-6 active:bg-muted"
      onPress={onPress}
+      disabled={disabled}
    >
      {renderAvatar()}
      <View className="min-w-0 flex-1 gap-2">
@@ -177,9 +185,11 @@ function ConversationCard({
 function SwipeableConversationCard({
  item,
  onPress,
+  disabled,
 }: {
  item: ConversationListItem;
  onPress: () => void;
+  disabled?: boolean;
 }) {
  const { t } = useTranslation('conversation');
  const deleteConversation = useDeleteConversation();
@@ -215,7 +225,7 @@ function SwipeableConversationCard({

  return (
    <Swipeable renderRightActions={renderRightActions} friction={2}>
-      <ConversationCard item={item} onPress={onPress} />
+      <ConversationCard item={item} onPress={onPress} disabled={disabled} />
    </Swipeable>
  );
 }
@@ -277,6 +287,7 @@ export default function ConversationsScreen() {
  const { data: conversations = [], isLoading } = useConversations();
  const createConversation = useCreateConversation();
  const createOnceGuardRef = useRef(false);
+  const [isEnteringChat, setIsEnteringChat] = useState(false);
  const [nowMs, setNowMs] = useState(() => Date.now());

  const isEmpty = conversations.length === 0;
@@ -311,8 +322,40 @@ export default function ConversationsScreen() {
    };
  }, []);

+  const navigateToConversation = async (
+    conversationId: string,
+    needsOpeningWarmup: boolean,
+  ) => {
+    setIsEnteringChat(true);
+    try {
+      if (needsOpeningWarmup) {
+        await warmupConversationOpening(queryClient, conversationId);
+      } else {
+        await prefetchConversationMessages(queryClient, conversationId);
+      }
+      router.push(`/(main)/conversation/${conversationId}`);
+    } catch (err) {
+      if (needsOpeningWarmup) {
+        abandonPreparedRealtimeSession(conversationId);
+      }
+      const msg =
+        err instanceof NetworkError
+          ? t('createError')
+          : err instanceof Error
+            ? err.message
+            : t('createError');
+      Alert.alert(t('chatTitle'), msg, [{ text: t('confirm') }]);
+    } finally {
+      setIsEnteringChat(false);
+    }
+  };
+
  const handleCreateConversation = () => {
-    if (createConversation.isPending || createOnceGuardRef.current) {
+    if (
+      createConversation.isPending ||
+      createOnceGuardRef.current ||
+      isEnteringChat
+    ) {
      return;
    }
    createOnceGuardRef.current = true;
@@ -325,8 +368,11 @@ export default function ConversationsScreen() {
        });
        const reuseId = findReusableEmptyConversationId(fresh ?? []);
        if (reuseId) {
+          try {
+            await navigateToConversation(reuseId, true);
+          } finally {
            createOnceGuardRef.current = false;
-          router.push(`/(main)/conversation/${reuseId}`);
+          }
          return;
        }
      } catch {
@@ -334,9 +380,12 @@ export default function ConversationsScreen() {
      }

      createConversation.mutate(undefined, {
-        onSuccess: (result) => {
+        onSuccess: async (result) => {
+          try {
+            await navigateToConversation(result.id, true);
+          } finally {
            createOnceGuardRef.current = false;
-          router.push(`/(main)/conversation/${result.id}`);
+          }
        },
        onError: (err) => {
          createOnceGuardRef.current = false;
@@ -357,15 +406,15 @@ export default function ConversationsScreen() {
  const handleResumeLatestConversation = () => {
    const toResume = findTodayConversationToResume(conversations, nowMs);
    if (toResume) {
-      router.push(`/(main)/conversation/${toResume.id}`);
+      void navigateToConversation(toResume.id, false);
      return;
    }
    // 当日没有可继续的会话（例如会话始于昨日）：与「打个招呼」一致，复用当日空会话或新建
    handleCreateConversation();
  };

-  const handleConversationPress = (id: string) => {
-    router.push(`/(main)/conversation/${id}`);
+  const handleConversationPress = (item: ConversationListItem) => {
+    void navigateToConversation(item.id, !item.hasUserMessage);
  };

  return (
@@ -401,7 +450,7 @@ export default function ConversationsScreen() {
            <Pressable
              className="items-center gap-6 rounded-2xl bg-muted/30 p-10 active:opacity-90"
              onPress={handleCreateConversation}
-              disabled={createConversation.isPending}
+              disabled={createConversation.isPending || isEnteringChat}
            >
              <Icon as={MessageCirclePlus} className="text-primary" size={40} />
              <View className="items-center gap-4">
@@ -424,6 +473,9 @@ export default function ConversationsScreen() {
              <Pressable
                className="items-center gap-6 rounded-2xl bg-muted/30 p-6 active:opacity-90"
                onPress={handleResumeLatestConversation}
+                disabled={
+                  isEnteringChat || createConversation.isPending
+                }
              >
                <Icon
                  as={MessageCirclePlus}
@@ -473,7 +525,8 @@ export default function ConversationsScreen() {
                    <SwipeableConversationCard
                      key={item.id}
                      item={item}
-                      onPress={() => handleConversationPress(item.id)}
+                      disabled={isEnteringChat}
+                      onPress={() => handleConversationPress(item)}
                    />
                  ))}
                </View>
--- a/app-expo/src/features/auth/hooks.ts
+++ b/app-expo/src/features/auth/hooks.ts
@@ -4,6 +4,7 @@ import { useCallback } from 'react';

 import { AuthError } from '@/core/api/types';
 import { tokenManager } from '@/core/auth/token-manager';
+import { disposeAllBackgroundConversationWs } from '@/features/conversation/conversation-ws-background-pool';

 import { authApi } from './api';
 import type {
@@ -235,6 +236,7 @@ export function useLogout() {
      }
    },
    onSettled: async () => {
+      disposeAllBackgroundConversationWs();
      await tokenManager.clearTokens();
      queryClient.clear();
      queryClient.setQueryData(authKeys.tokenCheck, false);
--- a/app-expo/src/features/conversation/conversation-ws-background-pool.ts
+++ b/app-expo/src/features/conversation/conversation-ws-background-pool.ts
@@ -0,0 +1,88 @@
+import type { QueryClient } from '@tanstack/react-query';
+import { AppState, type AppStateStatus } from 'react-native';
+
+import { RealtimeSession } from './realtime-session';
+
+type Slot = { conversationId: string; session: RealtimeSession };
+
+let slot: Slot | null = null;
+
+/** 与常见聊天 App 一致：仅当应用进入 background 时断开长连（避免后台挂 socket）；inactive 不处理以减少控制中心等短暂打断 */
+let backgroundUnsubscribe: (() => void) | null = null;
+
+function installBackgroundLifecycleOnce(): void {
+  if (backgroundUnsubscribe) return;
+  const sub = AppState.addEventListener('change', (next: AppStateStatus) => {
+    if (next === 'background') {
+      disposeAllBackgroundConversationWs();
+    }
+  });
+  backgroundUnsubscribe = () => sub.remove();
+}
+
+function disposeSlot(): void {
+  if (!slot) return;
+  slot.session.dispose();
+  slot = null;
+}
+
+const offScreenUi = {
+  onStreamingText: () => {},
+  onTtsSegment: () => {},
+  onError: () => {},
+  onStateChange: () => {},
+};
+
+/** 离屏：保持 WebSocket，去掉 UI 回调，避免列表页播 TTS 或对已卸载组件 setState */
+export function releaseConversationWsUi(session: RealtimeSession): void {
+  session.attachUiCallbacks({
+    onStreamingText: offScreenUi.onStreamingText,
+    onTtsSegment: offScreenUi.onTtsSegment,
+    onError: offScreenUi.onError,
+    onStateChange: offScreenUi.onStateChange,
+  });
+}
+
+/** 删除会话等场景：关闭对应长连 */
+export function disposeBackgroundConversationWs(conversationId: string): void {
+  if (slot?.conversationId === conversationId) {
+    disposeSlot();
+  }
+}
+
+/** 登出 / 清账号：关闭池中连接 */
+export function disposeAllBackgroundConversationWs(): void {
+  disposeSlot();
+}
+
+/**
+ * 单槽：仅保留「最近进入」的一个会话长连。换会话会dispose旧槽；同会话返回池中实例。
+ */
+export function acquireBackgroundConversationWs(
+  conversationId: string,
+  queryClient: QueryClient,
+  prepared: RealtimeSession | null,
+): RealtimeSession {
+  installBackgroundLifecycleOnce();
+  if (prepared) {
+    if (
+      slot &&
+      (slot.conversationId !== conversationId || slot.session !== prepared)
+    ) {
+      disposeSlot();
+    }
+    slot = { conversationId, session: prepared };
+    return prepared;
+  }
+
+  if (slot?.conversationId === conversationId) {
+    void slot.session.connect();
+    return slot.session;
+  }
+
+  disposeSlot();
+  const session = new RealtimeSession({ conversationId, queryClient });
+  slot = { conversationId, session };
+  void session.connect();
+  return session;
+}
--- a/app-expo/src/features/conversation/entry-warmup.ts
+++ b/app-expo/src/features/conversation/entry-warmup.ts
@@ -0,0 +1,93 @@
+import type { QueryClient } from '@tanstack/react-query';
+
+import { conversationMessagesRepository } from './conversation-messages-repository';
+import { conversationKeys } from './query-keys';
+import { registerPreparedRealtimeSession } from './prepared-session-registry';
+import { RealtimeSession } from './realtime-session';
+import type { MessageItem } from './types';
+
+const OPENING_WARMUP_TIMEOUT_MS = 50_000;
+const CACHE_POLL_MS = 120;
+
+function cacheHasAssistantMessage(
+  queryClient: QueryClient,
+  conversationId: string,
+): boolean {
+  const data = queryClient.getQueryData<MessageItem[]>(
+    conversationKeys.messages(conversationId),
+  );
+  return (data ?? []).some((m) => m.senderType === 'assistant');
+}
+
+function waitForAssistantInCache(
+  queryClient: QueryClient,
+  conversationId: string,
+  timeoutMs: number,
+): Promise<boolean> {
+  return new Promise((resolve) => {
+    const deadline = Date.now() + timeoutMs;
+    const schedule = () => {
+      if (cacheHasAssistantMessage(queryClient, conversationId)) {
+        resolve(true);
+        return;
+      }
+      if (Date.now() >= deadline) {
+        resolve(false);
+        return;
+      }
+      setTimeout(schedule, CACHE_POLL_MS);
+    };
+    schedule();
+  });
+}
+
+export async function prefetchConversationMessages(
+  queryClient: QueryClient,
+  conversationId: string,
+): Promise<void> {
+  await queryClient.prefetchQuery({
+    queryKey: conversationKeys.messages(conversationId),
+    queryFn: () => conversationMessagesRepository.loadMessages(conversationId),
+  });
+}
+
+/**
+ * 在会话列表阶段连接 WS 并等待首条助手开场写入 React Query；成功后挂起会话供聊天页接棒。
+ * 超时或失败则 dispose，由聊天页自行重连（服务端若已写入 history 不会重复开场）。
+ */
+export async function warmupConversationOpening(
+  queryClient: QueryClient,
+  conversationId: string,
+): Promise<void> {
+  if (cacheHasAssistantMessage(queryClient, conversationId)) {
+    await prefetchConversationMessages(queryClient, conversationId);
+    return;
+  }
+
+  const session = new RealtimeSession({
+    conversationId,
+    queryClient,
+  });
+
+  session.attachUiCallbacks({
+    onStreamingText: () => {},
+    onTtsSegment: () => {},
+    onError: () => {},
+    onStateChange: () => {},
+  });
+
+  await session.connect();
+
+  const ok = await waitForAssistantInCache(
+    queryClient,
+    conversationId,
+    OPENING_WARMUP_TIMEOUT_MS,
+  );
+
+  if (ok) {
+    registerPreparedRealtimeSession(conversationId, session);
+    await prefetchConversationMessages(queryClient, conversationId);
+  } else {
+    session.dispose();
+  }
+}
--- a/app-expo/src/features/conversation/hooks.ts
+++ b/app-expo/src/features/conversation/hooks.ts
@@ -1,17 +1,25 @@
 import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
 import { File, Paths } from 'expo-file-system';
 import { useCallback, useEffect, useRef, useState } from 'react';
+import { AppState, type AppStateStatus } from 'react-native';

 import type { WsConnectionState } from '@/core/ws/types';

 import { conversationApi } from './api';
+import {
+  acquireBackgroundConversationWs,
+  disposeAllBackgroundConversationWs,
+  disposeBackgroundConversationWs,
+  releaseConversationWsUi,
+} from './conversation-ws-background-pool';
 import { conversationMessagesRepository } from './conversation-messages-repository';
 import { conversationKeys } from './query-keys';
+import { takePreparedRealtimeSession } from './prepared-session-registry';
 import {
-  RealtimeSession,
  type ErrorCallback,
  type StreamingTextCallback,
  type TtsSegmentPayload,
+  type RealtimeSession,
 } from './realtime-session';
 import {
  type ConversationListItem,
@@ -126,6 +134,7 @@ export function useDeleteConversation() {
    mutationFn: (conversationId: string) =>
      conversationApi.delete(conversationId),
    onSuccess: async (_, conversationId) => {
+      disposeBackgroundConversationWs(conversationId);
      await voiceSegmentStore.clearConversation(conversationId);
      queryClient.setQueryData<ConversationListItem[]>(
        conversationKeys.lists(),
@@ -197,6 +206,13 @@ export function useRealtimeSession({
 }: UseRealtimeSessionOptions): RealtimeSessionState {
  const queryClient = useQueryClient();
  const sessionRef = useRef<RealtimeSession | null>(null);
+  const uiRef = useRef({
+    handleStreamingText: (() => {}) as StreamingTextCallback,
+    handleError: (() => {}) as ErrorCallback,
+    onTtsSegment: undefined as
+      | ((payload: TtsSegmentPayload) => void)
+      | undefined,
+  });

  const [connectionState, setConnectionState] =
    useState<WsConnectionState>('disconnected');
@@ -205,6 +221,10 @@ export function useRealtimeSession({
  const [awaitingAssistantReply, setAwaitingAssistantReply] = useState(false);
  const [error, setError] = useState<string | null>(null);

+  const [foregroundResumeGeneration, setForegroundResumeGeneration] =
+    useState(0);
+  const needsResumeAfterBackgroundRef = useRef(false);
+
  const handleStreamingText: StreamingTextCallback = useCallback(
    (text, isComplete) => {
      if (text.trim().length > 0) {
@@ -225,36 +245,60 @@ export function useRealtimeSession({
    setError(message);
  }, []);

+  uiRef.current.handleStreamingText = handleStreamingText;
+  uiRef.current.handleError = handleError;
+  uiRef.current.onTtsSegment = onTtsSegment;
+
  useEffect(() => {
    if (!enabled || !conversationId) return;

-    const session = new RealtimeSession({
+    const sub = AppState.addEventListener('change', (next: AppStateStatus) => {
+      if (next === 'background') {
+        needsResumeAfterBackgroundRef.current = true;
+        disposeAllBackgroundConversationWs();
+        sessionRef.current = null;
+        setConnectionState('disconnected');
+        setStreamingMessage(null);
+        setAwaitingAssistantReply(false);
+      } else if (next === 'active' && needsResumeAfterBackgroundRef.current) {
+        needsResumeAfterBackgroundRef.current = false;
+        setForegroundResumeGeneration((g) => g + 1);
+      }
+    });
+
+    return () => sub.remove();
+  }, [enabled, conversationId]);
+
+  useEffect(() => {
+    if (!enabled || !conversationId) return;
+
+    const prepared = takePreparedRealtimeSession(conversationId);
+    const session = acquireBackgroundConversationWs(
      conversationId,
      queryClient,
-      onStreamingText: handleStreamingText,
-      onTtsSegment,
-      onError: handleError,
+      prepared,
+    );
+
+    session.attachUiCallbacks({
+      onStreamingText: (text, isComplete) => {
+        uiRef.current.handleStreamingText(text, isComplete);
+      },
+      onTtsSegment: (payload) => uiRef.current.onTtsSegment?.(payload),
+      onError: (message, code) => uiRef.current.handleError(message, code),
      onStateChange: setConnectionState,
    });

    sessionRef.current = session;
-    session.connect();
+    setConnectionState(session.getConnectionState());

    return () => {
-      session.dispose();
+      releaseConversationWsUi(session);
      sessionRef.current = null;
      setConnectionState('disconnected');
      setStreamingMessage(null);
      setAwaitingAssistantReply(false);
    };
-  }, [
-    conversationId,
-    enabled,
-    queryClient,
-    handleStreamingText,
-    handleError,
-    onTtsSegment,
-  ]);
+  }, [conversationId, enabled, queryClient, foregroundResumeGeneration]);

  const sendText = useCallback(
    (text: string) => {
--- a/app-expo/src/features/conversation/prepared-session-registry.ts
+++ b/app-expo/src/features/conversation/prepared-session-registry.ts
@@ -0,0 +1,33 @@
+import type { RealtimeSession } from './realtime-session';
+
+const preparedByConversationId = new Map<string, RealtimeSession>();
+
+/** 列表页预热完成后挂起会话，聊天页挂载时接棒并删除登记 */
+export function registerPreparedRealtimeSession(
+  conversationId: string,
+  session: RealtimeSession,
+): void {
+  const old = preparedByConversationId.get(conversationId);
+  if (old && old !== session) {
+    old.dispose();
+  }
+  preparedByConversationId.set(conversationId, session);
+}
+
+/** 取出即视为消费；若无则返回 null */
+export function takePreparedRealtimeSession(
+  conversationId: string,
+): RealtimeSession | null {
+  const session = preparedByConversationId.get(conversationId);
+  if (!session) return null;
+  preparedByConversationId.delete(conversationId);
+  return session;
+}
+
+/** 预热成功后导航失败时释放挂起连接，避免僵尸 WebSocket */
+export function abandonPreparedRealtimeSession(conversationId: string): void {
+  const session = preparedByConversationId.get(conversationId);
+  if (!session) return;
+  preparedByConversationId.delete(conversationId);
+  session.dispose();
+}
--- a/app-expo/src/features/conversation/realtime-session.ts
+++ b/app-expo/src/features/conversation/realtime-session.ts
@@ -53,12 +53,14 @@ export class RealtimeSession {
  private onStreamingText?: StreamingTextCallback;
  private onTtsSegment?: (payload: TtsSegmentPayload) => void;
  private onError?: ErrorCallback;
+  private uiStateListener?: WsStateListener;
  private unsubEvent: (() => void) | null = null;
  private unsubState: (() => void) | null = null;

  private streamingBuffer = '';
  /** 单段回复且服务端带 `assistant_message_id` 时用于落缓存 id */
  private pendingAssistantMessageId: string | null = null;
+  private destroyed = false;

  constructor(options: RealtimeSessionOptions) {
    this.client = new WsClient(options.conversationId);
@@ -67,11 +69,35 @@ export class RealtimeSession {
    this.onStreamingText = options.onStreamingText;
    this.onTtsSegment = options.onTtsSegment;
    this.onError = options.onError;
+    this.uiStateListener = options.onStateChange;

    this.unsubEvent = this.client.onEvent(this.handleEvent);

-    if (options.onStateChange) {
-      this.unsubState = this.client.onStateChange(options.onStateChange);
+    this.unsubState = this.client.onStateChange((state) => {
+      this.uiStateListener?.(state);
+    });
+  }
+
+  /** 列表预热接棒或刷新 UI 订阅时替换回调，不重建 WebSocket */
+  attachUiCallbacks(options: {
+    onStreamingText?: StreamingTextCallback;
+    onTtsSegment?: (payload: TtsSegmentPayload) => void;
+    onError?: ErrorCallback;
+    onStateChange?: WsStateListener;
+  }): void {
+    if (this.destroyed) return;
+    if (options.onStreamingText !== undefined) {
+      this.onStreamingText = options.onStreamingText;
+    }
+    if (options.onTtsSegment !== undefined) {
+      this.onTtsSegment = options.onTtsSegment;
+    }
+    if (options.onError !== undefined) {
+      this.onError = options.onError;
+    }
+    if (options.onStateChange !== undefined) {
+      this.uiStateListener = options.onStateChange;
+      options.onStateChange(this.client.getState());
    }
  }

@@ -84,6 +110,8 @@ export class RealtimeSession {
  }

  dispose(): void {
+    if (this.destroyed) return;
+    this.destroyed = true;
    this.flushStreamingBufferIfPending();
    this.unsubEvent?.();
    this.unsubState?.();
--- a/app-expo/src/features/profile/hooks.ts
+++ b/app-expo/src/features/profile/hooks.ts
@@ -2,6 +2,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
 import { router } from 'expo-router';

 import { tokenManager } from '@/core/auth/token-manager';
+import { disposeAllBackgroundConversationWs } from '@/features/conversation/conversation-ws-background-pool';
 import { authKeys } from '@/features/auth/hooks';

 import { profileApi } from './api';
@@ -97,6 +98,7 @@ export function usePurgeUserData() {
  return useMutation({
    mutationFn: (body: PurgeUserDataRequest) => profileApi.purgeUserData(body),
    onSuccess: async () => {
+      disposeAllBackgroundConversationWs();
      await tokenManager.clearTokens();
      queryClient.clear();
      queryClient.setQueryData(authKeys.tokenCheck, false);