fix(frontend): recover the chat stream from an expired access token (#3028)

* fix(frontend): recover the chat stream from an expired access token

The SSE chat request runs on raw fetch, outside the axios instance, so the
401 refresh flow in utils/request.ts never applied to it. An expired token
left the user staring at an orphaned message bubble and a dead-end
"HTTP 401" toast, with no re-login prompt until some unrelated polling
request happened to hit 401 seconds later.

Extract the refresh into a shared, de-duplicated helper and have the stream
client use it: on a 401 handshake, refresh once and replay the request.
The replay cannot duplicate a message because auth rejects before the
handler runs, and nothing has been rendered yet. Embed visitors are
excluded since their token cannot be refreshed by the SPA.

* fix(frontend): keep a minted session after a post-refresh SSE 401

A second chat handshake 401 is not proof the session is gone — tenant
lookup also returns 401 — so wiping the just-refreshed tokens logged
users out unlike axios. Identify StreamAuthError by name, pin replay
to the original abort controller, and cover the shared refresh queue
with unit tests.

---------

Co-authored-by: wizardchen <wizardchen@tencent.com>

frontend/src/api/chat/streame.ts

@@ -7,8 +7,12 @@ import {
   sanitizeStreamRequestBody,
   type StreamRequestMeta,
 } from '@/utils/chatRequestDebug';
-
-
+import {
+  StreamAuthError,
+  isStreamAuthError,
+  refreshAccessTokenShared,
+  runStreamWithAuthRetry,
+} from '@/utils/authRefresh';
 
 interface StreamOptions {
   // 请求方法 (默认POST)
@@ -38,6 +42,7 @@ export function useStream() {
   // 启动流式请求
   const startStream = async (params: { session_id: any; query: any; knowledge_base_ids?: string[]; knowledge_ids?: string[]; tag_ids?: string[]; agent_enabled?: boolean; agent_id?: string; agent_source_tenant_id?: string | number; web_search_enabled?: boolean; summary_model_id?: string; mcp_service_ids?: string[]; skill_names?: string[]; mentioned_items?: Array<{id: string; name: string; type: string; kb_type?: string; kb_id?: string; kb_name?: string; service_id?: string; skill_name?: string}>; images?: Array<{data: string}>; attachment_uploads?: Array<{data: string; file_name: string; file_size: number}>; attachment_ids?: string[]; suggestion_attribution?: { suggestion_set_id: string; question_id: string }; method: string; url: string; embed_token?: string; embed_session_sig?: string; embed_visitor_id?: string }) => {
     const myGeneration = ++streamGeneration
+    const streamAbort = controller
     // 重置状态
     output.value = '';
     error.value = null;
@@ -148,11 +153,14 @@ export function useStream() {
         sentAt: Date.now(),
       };
       
-      await fetchEventSource(url, {
+      // Wrapped so an expired access token can be refreshed and the request
+      // replayed once. Nothing has been streamed to the UI yet when the
+      // handshake 401s, so the replay is invisible to the user.
+      const runStream = (authToken: string) => fetchEventSource(url, {
         method: params.method,
         headers: {
           "Content-Type": "application/json",
-          "Authorization": embedToken ? `Embed ${embedToken}` : `Bearer ${token}`,
+          "Authorization": embedToken ? `Embed ${embedToken}` : `Bearer ${authToken}`,
           "Accept-Language": i18n.global.locale?.value || localStorage.getItem('locale') || 'zh-CN',
           "X-Request-ID": requestID,
           ...(!embedToken && tenantIdHeader ? { "X-Tenant-ID": tenantIdHeader } : {}),
@@ -163,10 +171,12 @@ export function useStream() {
           params.method == "POST"
             ? JSON.stringify(postBody)
             : null,
-        signal: controller.signal,
+        signal: streamAbort.signal,
         openWhenHidden: true,
 
         onopen: async (res) => {
+          // 401 is recoverable (refresh + replay); everything else is not.
+          if (res.status === 401) throw new StreamAuthError(res.status);
           if (!res.ok) throw new Error(`HTTP ${res.status}`);
           console.log(`[TTFB] response:headers request_id=${requestID} elapsed_ms=${(performance.now() - sentAt).toFixed(1)}`);
           isLoading.value = false;
@@ -190,6 +200,7 @@ export function useStream() {
         },
 
         onerror: (err) => {
+          if (isStreamAuthError(err)) throw err;
           throw new Error(`${i18n.global.t('error.streamFailed')}: ${err}`);
         },
 
@@ -197,6 +208,20 @@ export function useStream() {
           stopStream();
         },
       });
+
+      await runStreamWithAuthRetry({
+        run: runStream,
+        initialToken: token,
+        isEmbed: Boolean(embedToken),
+        isCurrent: () => myGeneration === streamGeneration && !streamAbort.signal.aborted,
+        refreshAccessToken: () => refreshAccessTokenShared({
+          messages: {
+            pleaseRelogin: i18n.global.t('error.pleaseRelogin'),
+            tokenRefreshFailed: i18n.global.t('error.tokenRefreshFailed'),
+          },
+        }),
+        reloginMessage: i18n.global.t('error.pleaseRelogin'),
+      });
     } catch (err) {
       error.value = err instanceof Error ? err.message : String(err)
       stopStream()

frontend/src/utils/authRefresh.test.ts

@@ -0,0 +1,218 @@
+import assert from 'node:assert/strict'
+import { readFileSync } from 'node:fs'
+import test from 'node:test'
+
+import {
+  StreamAuthError,
+  forceReloginRedirect,
+  isStreamAuthError,
+  refreshAccessTokenShared,
+  resetAuthRefreshStateForTests,
+  runStreamWithAuthRetry,
+} from './authRefresh.ts'
+
+type Store = Record<string, string>
+
+function installBrowser(pathname = '/chat') {
+  const store: Store = {}
+  Object.defineProperty(globalThis, 'localStorage', {
+    value: {
+      getItem: (key: string) => (key in store ? store[key] : null),
+      setItem: (key: string, value: string) => {
+        store[key] = String(value)
+      },
+      removeItem: (key: string) => {
+        delete store[key]
+      },
+    },
+    configurable: true,
+    writable: true,
+  })
+  const location = { pathname, href: `http://localhost${pathname}` }
+  Object.defineProperty(globalThis, 'window', {
+    value: { location },
+    configurable: true,
+    writable: true,
+  })
+  return { store, location }
+}
+
+test.afterEach(() => {
+  resetAuthRefreshStateForTests()
+})
+
+test('isStreamAuthError matches the class and name-only wrappers', () => {
+  assert.equal(isStreamAuthError(new StreamAuthError(401)), true)
+  assert.equal(isStreamAuthError({ name: 'StreamAuthError', message: 'HTTP 401' }), true)
+  assert.equal(isStreamAuthError(new Error('HTTP 401')), false)
+  assert.equal(isStreamAuthError('HTTP 401'), false)
+})
+
+test('concurrent 401s share a single refresh call', async () => {
+  const { store } = installBrowser()
+  store.weknora_refresh_token = 'rt-1'
+  let calls = 0
+  let release!: (value: { success: true; data: { token: string; refreshToken: string } }) => void
+  const pending = new Promise<{ success: true; data: { token: string; refreshToken: string } }>(
+    (resolve) => {
+      release = resolve
+    },
+  )
+  const refresh = async () => {
+    calls += 1
+    return pending
+  }
+
+  const first = refreshAccessTokenShared({ refresh })
+  await Promise.resolve()
+  const second = refreshAccessTokenShared({ refresh })
+  assert.equal(calls, 1)
+
+  release({ success: true, data: { token: 'access-2', refreshToken: 'rt-2' } })
+  assert.deepEqual(await Promise.all([first, second]), ['access-2', 'access-2'])
+  assert.equal(store.weknora_token, 'access-2')
+  assert.equal(store.weknora_refresh_token, 'rt-2')
+  assert.equal(calls, 1)
+})
+
+test('missing refresh token clears credentials, redirects, and throws an Error', async () => {
+  const { store, location } = installBrowser()
+  store.weknora_token = 'expired'
+  store.weknora_user = '{}'
+  store.weknora_selected_tenant_id = '9'
+
+  await assert.rejects(
+    () => refreshAccessTokenShared({
+      refresh: async () => {
+        throw new Error('should not refresh')
+      },
+      messages: { pleaseRelogin: 'please-relogin', tokenRefreshFailed: 'refresh-failed' },
+    }),
+    (err: unknown) => err instanceof Error && err.message === 'please-relogin',
+  )
+  assert.equal(store.weknora_token, undefined)
+  assert.equal(store.weknora_selected_tenant_id, undefined)
+  assert.equal(location.href, '/login')
+})
+
+test('a failed refresh clears the newly irrelevant session and redirects', async () => {
+  const { store, location } = installBrowser()
+  store.weknora_refresh_token = 'rt-dead'
+  store.weknora_token = 'expired'
+  store.weknora_selected_tenant_id = '9'
+
+  await assert.rejects(
+    () => refreshAccessTokenShared({
+      refresh: async () => ({ success: false, message: 'revoked' }),
+    }),
+    /revoked/,
+  )
+  assert.equal(store.weknora_refresh_token, undefined)
+  assert.equal(store.weknora_selected_tenant_id, undefined)
+  assert.equal(location.href, '/login')
+})
+
+test('runStreamWithAuthRetry refreshes once and replays with the new token', async () => {
+  installBrowser()
+  const tokens: string[] = []
+  const result = await runStreamWithAuthRetry({
+    initialToken: 'expired',
+    isEmbed: false,
+    isCurrent: () => true,
+    refreshAccessToken: async () => 'fresh',
+    reloginMessage: 'please-relogin',
+    run: async (token) => {
+      tokens.push(token)
+      if (token === 'expired') throw new StreamAuthError(401)
+      return `ok:${token}`
+    },
+  })
+  assert.deepEqual(tokens, ['expired', 'fresh'])
+  assert.equal(result, 'ok:fresh')
+})
+
+test('embed visitors are not refreshed or redirected', async () => {
+  let refreshed = 0
+  await assert.rejects(
+    () => runStreamWithAuthRetry({
+      initialToken: 'embed-token',
+      isEmbed: true,
+      isCurrent: () => true,
+      refreshAccessToken: async () => {
+        refreshed += 1
+        return 'nope'
+      },
+      reloginMessage: 'please-relogin',
+      run: async () => {
+        throw new StreamAuthError(401)
+      },
+    }),
+    (err: unknown) => err instanceof StreamAuthError,
+  )
+  assert.equal(refreshed, 0)
+})
+
+test('a second handshake 401 after a successful refresh does not wipe tokens', async () => {
+  const { store, location } = installBrowser()
+  store.weknora_token = 'expired'
+  store.weknora_refresh_token = 'rt-1'
+  location.href = 'http://localhost/chat'
+
+  await assert.rejects(
+    () => runStreamWithAuthRetry({
+      initialToken: 'expired',
+      isEmbed: false,
+      isCurrent: () => true,
+      refreshAccessToken: async () => {
+        const token = await refreshAccessTokenShared({
+          refresh: async () => ({
+            success: true,
+            data: { token: 'fresh', refreshToken: 'rt-2' },
+          }),
+        })
+        return token
+      },
+      reloginMessage: 'please-relogin',
+      run: async () => {
+        throw new StreamAuthError(401)
+      },
+    }),
+    (err: unknown) => err instanceof Error && err.message === 'please-relogin',
+  )
+  assert.equal(store.weknora_token, 'fresh')
+  assert.equal(store.weknora_refresh_token, 'rt-2')
+  assert.equal(location.href, 'http://localhost/chat')
+})
+
+test('a superseded send skips the replay', async () => {
+  let runs = 0
+  const result = await runStreamWithAuthRetry({
+    initialToken: 'expired',
+    isEmbed: false,
+    isCurrent: () => false,
+    refreshAccessToken: async () => 'fresh',
+    reloginMessage: 'please-relogin',
+    run: async () => {
+      runs += 1
+      throw new StreamAuthError(401)
+    },
+  })
+  assert.equal(runs, 1)
+  assert.equal(result, undefined)
+})
+
+test('forceReloginRedirect does not bounce embed visitors to /login', () => {
+  const { store, location } = installBrowser('/embed/ch-1')
+  store.weknora_token = 'jwt'
+  forceReloginRedirect()
+  assert.equal(store.weknora_token, undefined)
+  assert.equal(location.href, 'http://localhost/embed/ch-1')
+})
+
+test('chat stream wires the shared retry and keeps the abort controller', () => {
+  const source = readFileSync(new URL('../api/chat/streame.ts', import.meta.url), 'utf8')
+  assert.match(source, /runStreamWithAuthRetry/)
+  assert.match(source, /isStreamAuthError/)
+  assert.match(source, /streamAbort/)
+  assert.doesNotMatch(source, /forceReloginRedirect/)
+})

frontend/src/utils/authRefresh.ts

@@ -0,0 +1,196 @@
+/**
+ * Shared access-token refresh used by both axios and the chat SSE client.
+ *
+ * Refresh tokens rotate, so concurrent 401s must queue behind one in-flight
+ * /auth/refresh. The SSE handshake runs on raw fetch and cannot use the
+ * axios interceptor; both planes call {@link refreshAccessTokenShared}.
+ */
+
+const AUTH_STORAGE_KEYS = [
+  'weknora_token',
+  'weknora_refresh_token',
+  'weknora_user',
+  'weknora_tenant',
+  'weknora_knowledge_bases',
+  'weknora_current_kb',
+  'weknora_selected_tenant_id',
+  'weknora_selected_tenant_name',
+  'weknora_memberships',
+] as const
+
+let isRefreshing = false
+let failedQueue: Array<{
+  resolve: (token: string) => void
+  reject: (error: unknown) => void
+}> = []
+
+export type TokenRefreshResult = {
+  success: boolean
+  data?: { token: string; refreshToken: string }
+  message?: string
+}
+
+export type RefreshAccessTokenOptions = {
+  refresh?: (refreshToken: string) => Promise<TokenRefreshResult>
+  messages?: {
+    pleaseRelogin: string
+    tokenRefreshFailed: string
+  }
+}
+
+const defaultMessages = {
+  pleaseRelogin: 'Please log in again',
+  tokenRefreshFailed: 'Token refresh failed',
+}
+
+export class StreamAuthError extends Error {
+  constructor(readonly status: number) {
+    super(`HTTP ${status}`)
+    this.name = 'StreamAuthError'
+  }
+}
+
+export function isStreamAuthError(err: unknown): boolean {
+  if (!err || typeof err !== 'object') return false
+  if (err instanceof StreamAuthError) return true
+  return (err as { name?: string }).name === 'StreamAuthError'
+}
+
+export function isEmbedPage(): boolean {
+  if (typeof window === 'undefined') return false
+  return window.location.pathname.startsWith('/embed/')
+}
+
+export function redirectToLogin() {
+  if (typeof window === 'undefined') return
+  if (window.location.pathname === '/login') return
+  if (isEmbedPage()) return
+  window.location.href = '/login'
+}
+
+export function clearAuthStorage() {
+  for (const key of AUTH_STORAGE_KEYS) {
+    localStorage.removeItem(key)
+  }
+}
+
+export function forceReloginRedirect() {
+  clearAuthStorage()
+  redirectToLogin()
+}
+
+function processQueue(error: unknown, token: string | null = null) {
+  failedQueue.forEach(({ resolve, reject }) => {
+    if (error) {
+      reject(error)
+    } else {
+      resolve(token as string)
+    }
+  })
+  failedQueue = []
+}
+
+async function defaultRefresh(refreshToken: string): Promise<TokenRefreshResult> {
+  const { refreshToken: refreshTokenAPI } = await import('../api/auth/index')
+  return refreshTokenAPI(refreshToken)
+}
+
+/**
+ * Refresh the access token, de-duplicated across all callers.
+ *
+ * Resolves with the new access token. On failure it has already cleared
+ * credentials and redirected to /login.
+ */
+export async function refreshAccessTokenShared(
+  options: RefreshAccessTokenOptions = {},
+): Promise<string> {
+  const messages = { ...defaultMessages, ...options.messages }
+  const refresh = options.refresh ?? defaultRefresh
+
+  if (isRefreshing) {
+    return new Promise<string>((resolve, reject) => {
+      failedQueue.push({ resolve, reject })
+    })
+  }
+
+  isRefreshing = true
+  const storedRefreshToken = localStorage.getItem('weknora_refresh_token')
+
+  if (!storedRefreshToken) {
+    clearAuthStorage()
+    const noRefreshTokenError = new Error(messages.pleaseRelogin)
+    processQueue(noRefreshTokenError, null)
+    isRefreshing = false
+    redirectToLogin()
+    throw noRefreshTokenError
+  }
+
+  try {
+    const response = await refresh(storedRefreshToken)
+
+    if (!response.success || !response.data?.token) {
+      throw new Error(response.message || messages.tokenRefreshFailed)
+    }
+
+    const { token, refreshToken: newRefreshToken } = response.data
+    localStorage.setItem('weknora_token', token)
+    if (newRefreshToken) {
+      localStorage.setItem('weknora_refresh_token', newRefreshToken)
+    }
+    processQueue(null, token)
+    return token
+  } catch (refreshError) {
+    clearAuthStorage()
+    processQueue(refreshError, null)
+    redirectToLogin()
+    throw refreshError
+  } finally {
+    isRefreshing = false
+  }
+}
+
+/**
+ * Replay an SSE handshake once after a 401. Refresh failure already
+ * redirected; a second 401 after a successful refresh is surfaced to the
+ * caller without wiping the newly minted session (axios does the same).
+ *
+ * Returns undefined when a newer send or an abort superseded the replay.
+ */
+export async function runStreamWithAuthRetry<T>(options: {
+  run: (token: string) => Promise<T>
+  initialToken: string
+  isEmbed: boolean
+  isCurrent: () => boolean
+  refreshAccessToken: () => Promise<string>
+  reloginMessage: string
+}): Promise<T | undefined> {
+  try {
+    return await options.run(options.initialToken)
+  } catch (err) {
+    if (!isStreamAuthError(err) || options.isEmbed) throw err
+
+    let refreshedToken: string
+    try {
+      refreshedToken = await options.refreshAccessToken()
+    } catch {
+      throw new Error(options.reloginMessage)
+    }
+
+    if (!options.isCurrent()) return undefined
+
+    try {
+      return await options.run(refreshedToken)
+    } catch (retryErr) {
+      if (isStreamAuthError(retryErr)) {
+        throw new Error(options.reloginMessage)
+      }
+      throw retryErr
+    }
+  }
+}
+
+/** Reset module locks between unit tests. */
+export function resetAuthRefreshStateForTests() {
+  isRefreshing = false
+  failedQueue = []
+}

frontend/src/utils/request.ts

@@ -4,6 +4,13 @@ import { generateRandomString, MAX_FILE_SIZE_MB, MAX_SKILL_BUNDLE_SIZE_MB } from
 import i18n from '@/i18n'
 import { getApiBaseUrl } from './api-base';
 import { isSkillBundleUploadUrl } from './uploadLimit';
+import {
+  forceReloginRedirect,
+  isEmbedPage,
+  refreshAccessTokenShared,
+} from './authRefresh';
+
+export { forceReloginRedirect, refreshAccessTokenShared };
 
 const t = (key: string) => i18n.global.t(key)
 
@@ -102,10 +109,6 @@ instance.interceptors.request.use(
   }
 );
 
-// Token刷新标志，防止多个请求同时刷新token
-let isRefreshing = false;
-let failedQueue: Array<{ resolve: Function; reject: Function }> = [];
-
 // Share-link endpoints (/auth/invitations/lookup, /auth/register-by-invite)
 // are reachable by anonymous users opening an invite link. A 401 from these
 // must surface to the page (e.g. expired token), not trigger the
@@ -118,32 +121,6 @@ function isPublicAuthRequest(url?: string): boolean {
   return PUBLIC_AUTH_PATHS.some(p => url.includes(p));
 }
 
-// 处理队列中的请求
-const processQueue = (error: any, token: string | null = null) => {
-  failedQueue.forEach(({ resolve, reject }) => {
-    if (error) {
-      reject(error);
-    } else {
-      resolve(token);
-    }
-  });
-  
-  failedQueue = [];
-};
-
-function isEmbedPage(): boolean {
-  if (typeof window === 'undefined') return false;
-  return window.location.pathname.startsWith('/embed/');
-}
-
-function redirectToLogin() {
-  if (typeof window === 'undefined') return;
-  if (window.location.pathname === '/login') return;
-  // Embed 渠道用 Embed token 鉴权，匿名访问不应被踢到登录页
-  if (isEmbedPage()) return;
-  window.location.href = '/login';
-}
-
 instance.interceptors.response.use(
   (response) => {
     // 根据业务状态码处理逻辑
@@ -181,70 +158,19 @@ instance.interceptors.response.use(
 
     // 如果是401错误且不是刷新token的请求，尝试刷新token
     if (error.response.status === 401 && !originalRequest._retry && !originalRequest.url?.includes('/auth/refresh')) {
-      if (isRefreshing) {
-        // 如果正在刷新token，将请求加入队列
-        return new Promise((resolve, reject) => {
-          failedQueue.push({ resolve, reject });
-        }).then(token => {
-          originalRequest.headers['Authorization'] = 'Bearer ' + token;
-          return instance(originalRequest);
-        }).catch(err => {
-          return Promise.reject(err);
-        });
-      }
-      
       originalRequest._retry = true;
-      isRefreshing = true;
-      
-      const refreshToken = localStorage.getItem('weknora_refresh_token');
-      
-      if (refreshToken) {
-        try {
-          // 动态导入refresh token API
-          const { refreshToken: refreshTokenAPI } = await import('../api/auth/index');
-          const response = await refreshTokenAPI(refreshToken);
-          
-          if (response.success && response.data) {
-            const { token, refreshToken: newRefreshToken } = response.data;
-            
-            // 更新localStorage中的token
-            localStorage.setItem('weknora_token', token);
-            localStorage.setItem('weknora_refresh_token', newRefreshToken);
-            
-            // 更新请求头
-            originalRequest.headers['Authorization'] = 'Bearer ' + token;
-            
-            // 处理队列中的请求
-            processQueue(null, token);
-            
-            return instance(originalRequest);
-          } else {
-            throw new Error(response.message || t('error.tokenRefreshFailed'));
-          }
-        } catch (refreshError) {
-          // 刷新失败，清除所有token并跳转到登录页
-          localStorage.removeItem('weknora_token');
-          localStorage.removeItem('weknora_refresh_token');
-          localStorage.removeItem('weknora_user');
-          localStorage.removeItem('weknora_tenant');
-          
-          processQueue(refreshError, null);
-          
-          redirectToLogin();
-          
-          return Promise.reject(refreshError);
-        } finally {
-          isRefreshing = false;
-        }
-      } else {
-        // 没有refresh token，直接跳转到登录页
-        localStorage.removeItem('weknora_token');
-        localStorage.removeItem('weknora_user');
-        localStorage.removeItem('weknora_tenant');
-        
-        redirectToLogin();
-        
-        return Promise.reject({ message: t('error.pleaseRelogin') });
+      try {
+        const token = await refreshAccessTokenShared({
+          messages: {
+            pleaseRelogin: t('error.pleaseRelogin'),
+            tokenRefreshFailed: t('error.tokenRefreshFailed'),
+          },
+        });
+        originalRequest.headers['Authorization'] = 'Bearer ' + token;
+        return instance(originalRequest);
+      } catch (refreshError) {
+        // refreshAccessTokenShared already cleared credentials and redirected.
+        return Promise.reject(refreshError);
       }
     }