fix(chat): stop completed streams and expand audit queries

src/features/audit/engine.js

@@ -49,6 +49,12 @@ const REFINE_PATTERNS = [
   '最低分',
 ];
 
+// Remove request/action wording before extracting business themes. Without
+// this, "我要开展反洗钱专项审计" is treated as the single keyword
+// "我要开展反洗钱", which cannot match the knowledge-base term "反洗钱".
+const QUERY_ACTION_PATTERN =
+  /帮我开展|我要开展|我想开展|计划开展|准备开展|开展|实施|发起|启动|进行|做一个|帮我做|帮我|我要|我想/;
+
 function firstValue(...values) {
   return values.find(
     (value) => value !== undefined && value !== null && value !== '',
@@ -65,6 +71,52 @@ function toText(value) {
   return String(value).trim();
 }
 
+function normalizeExpansionTerms(values) {
+  const terms = Array.isArray(values) ? values : [values];
+  return terms
+    .flatMap((value) =>
+      toText(value)
+        .split(/[、，,；;|/\s\n]+/)
+        .map((term) => term.trim()),
+    )
+    .map((term) =>
+      term.replace(/^[\s:：()（）\u005b\u005d【】]+|[\s。！？!?，,；;]+$/g, ''),
+    )
+    .filter(
+      (term) =>
+        term.length >= 2 &&
+        term.length <= 24 &&
+        !AUDIT_STOP_WORDS.has(term) &&
+        !/[。！？!?]/.test(term),
+    );
+}
+
+export function applyAuditQueryExpansion(query = '', payload = {}) {
+  const originalQuery = toText(query);
+  const selectedTerms = new Set();
+  const addTerms = (values) => {
+    normalizeExpansionTerms(values).forEach((term) => {
+      if (!originalQuery.includes(term)) selectedTerms.add(term);
+    });
+  };
+
+  addTerms(payload.auditDomain);
+  addTerms(payload.auditDomains);
+  addTerms(payload.keywords);
+  const expandedKeywords = Array.from(selectedTerms).slice(0, 8);
+  const confidence = Math.max(0, Math.min(1, Number(payload.confidence) || 0));
+
+  return {
+    originalQuery,
+    optimizedQuery: originalQuery,
+    selectedTerms: expandedKeywords,
+    expandedKeywords,
+    confidence,
+    needsConfirmation: Boolean(payload.needsConfirmation),
+    applied: expandedKeywords.length > 0,
+  };
+}
+
 export function normalizeRecordList(res) {
   if (Array.isArray(res)) return res;
   if (Array.isArray(res?.records)) return res.records;
@@ -260,7 +312,11 @@ function extractPrimaryKeywords(query = '') {
     .match(/[\u4e00-\u9fffA-Za-z0-9]{2,}/g)
     ?.forEach((token) => {
       token
-        .split(/专项|审计|业务|风险|检查|项目|清单|生成|查询|检索|相关/)
+        .split(
+          new RegExp(
+            `${QUERY_ACTION_PATTERN.source}|专项|审计|业务|风险|检查|项目|清单|生成|查询|检索|相关`,
+          ),
+        )
         .map((item) => item.trim())
         .filter((item) => item.length >= 2 && !AUDIT_STOP_WORDS.has(item))
         .forEach((item) => results.add(item));
@@ -284,7 +340,7 @@ export function extractAuditKeywords(query = '', extraKeywords = []) {
   });
   extraKeywords
     .map((keyword) => toText(keyword))
-    .filter((keyword) => keyword.length >= 2 && keyword.length <= 12)
+    .filter((keyword) => keyword.length >= 2 && keyword.length <= 24)
     .forEach((keyword) => keywords.add(keyword));
   return {
     domains,

src/pages/Chat/index.jsx

@@ -28,12 +28,14 @@ import {
 } from '../../features/audit/components';
 import { AUDIT_KNOWLEDGE_FILE_SUBJECT } from '../../features/audit/constants';
 import {
+  applyAuditQueryExpansion,
   buildAuditKnowledgePrompt,
   buildAuditResult,
   buildGroundedAuditKnowledgePrompt,
   classifyAuditIntent,
   fetchCurrentAuditKnowledgeFile,
   findPreviousAuditResult,
+  isAuditRefineRequest,
 } from '../../features/audit/engine';
 import { parseAuditExcelFromRemote } from '../../features/audit/excel';
 import {
@@ -44,14 +46,12 @@ import {
   documentRerank,
   extractLlmChunkContent,
   extractLlmChunkReasoning,
-  getApplicationRuntimeConfig,
   imageAnalyze,
   llmChat,
   llmList,
   queryMcps,
   queryRecords,
   queryTopics,
-  resolveLlmModel,
   semanticSearch,
   updateConversation,
   updateRecord,
@@ -222,6 +222,66 @@ function getStoredLlmModel() {
   }
 }
 
+function parseJsonObject(value = '') {
+  const source = String(value).trim();
+  const fenced = source.match(/```(?:json)?\s*([\s\S]*?)```/i)?.[1];
+  const candidate = fenced || source;
+  const start = candidate.indexOf('{');
+  const end = candidate.lastIndexOf('}');
+  if (start < 0 || end <= start) return null;
+  try {
+    return JSON.parse(candidate.slice(start, end + 1));
+  } catch {
+    return null;
+  }
+}
+
+async function readLlmText(response) {
+  let content = '';
+  for await (const chunk of createSseStream(response.body)) {
+    const data = String(chunk?.data || '').trim();
+    if (!data || data === '[DONE]' || data === 'DONE') continue;
+    let parsed = data;
+    try {
+      parsed = JSON.parse(data);
+    } catch {}
+    content += extractLlmChunkContent(parsed);
+  }
+  return content;
+}
+
+async function expandAuditQuery(query, model, signal) {
+  const response = await llmChat(
+    {
+      model,
+      messages: [
+        {
+          role: 'system',
+          content: [
+            '你是审计知识库检索词扩展器，只负责扩展关键词，不回答用户问题。',
+            '根据原始问题识别审计领域，并补充同义词、别名、缩写或相关业务检索词。',
+            'keywords 只放短关键词，不要输出完整句子、风险事实、法规、建议或检查清单。',
+            '可以生成候选词表中没有的同义词，不需要本地词库维护。',
+            '最多输出 8 个扩展词；原始问题中已经出现的词不要重复输出。',
+            '只输出一个 JSON 对象，不要 Markdown 或解释。',
+            '格式：{"auditDomain":"","keywords":[],"confidence":0}',
+          ].join('\n'),
+        },
+        {
+          role: 'user',
+          content: `原始问题：${JSON.stringify(query)}`,
+        },
+      ],
+      enable_thinking: false,
+      temperature: 0,
+      max_tokens: 600,
+    },
+    signal,
+  );
+  const payload = parseJsonObject(await readLlmText(response));
+  return payload ? applyAuditQueryExpansion(query, payload) : null;
+}
+
 export default forwardRef((props, ref) => {
   const { userInfo, conversation, onAddConversation } = props;
   const conversationAgent = useMemo(
@@ -285,22 +345,7 @@ export default forwardRef((props, ref) => {
     queryKey: ['llm-models'],
     enabled: Boolean(userInfo),
     staleTime: 5 * 60 * 1000,
-    queryFn: async () => {
-      try {
-        return await llmList();
-      } catch (error) {
-        const fallbackModel = await resolveLlmModel({ thinking: true });
-        if (!fallbackModel) throw error;
-        return [
-          {
-            id: fallbackModel,
-            label: fallbackModel,
-            isDefault: true,
-            unavailable: true,
-          },
-        ];
-      }
-    },
+    queryFn: () => llmList(),
   });
 
   useEffect(() => {
@@ -541,19 +586,12 @@ export default forwardRef((props, ref) => {
     } catch (error) {
       console.error(error);
     }
-    console.log('toolCalls:', toolCalls);
     return {
       think: `${originThink}${currentThink}`,
       content: `${originContent}${currentContent}`,
       role: 'assistant',
-      auditResult:
-        originMessage?.auditResult ||
-        pendingAuditResultRef.current ||
-        undefined,
-      auditExport:
-        originMessage?.auditExport ||
-        pendingAuditExportRef.current ||
-        undefined,
+      auditResult: originMessage?.auditResult || undefined,
+      auditExport: originMessage?.auditExport || undefined,
       tool_calls: toolCalls,
       tools: toolCalls,
     };
@@ -760,11 +798,27 @@ export default forwardRef((props, ref) => {
           classifyAuditIntent(message.content) === 'checklist');
       if (usesChecklistMode) {
         const previousResult = findPreviousAuditResult(messages);
-        const result = buildAuditResult(
+        const query = message.content;
+        let queryExpansion = null;
+        if (!isAuditRefineRequest(message.content, previousResult)) {
+          try {
+            queryExpansion = await expandAuditQuery(
               message.content,
-          records,
-          previousResult,
+              selectedModel,
+              abortController.current?.signal,
             );
+          } catch (error) {
+            if (error?.name !== 'AbortError') {
+              console.warn('审计检索词 AI 扩展失败，回退本地规则：', error);
+            }
+          }
+        }
+        const result = buildAuditResult(query, records, previousResult, {
+          extraKeywords: queryExpansion?.applied
+            ? queryExpansion.expandedKeywords
+            : [],
+        });
+        if (queryExpansion) result.queryExpansion = queryExpansion;
         pendingAuditResultRef.current = result;
         auditKnowledgePrompt = buildAuditKnowledgePrompt(result);
       } else {
@@ -826,14 +880,13 @@ export default forwardRef((props, ref) => {
           }),
         ];
 
-        const content = JSON.stringify(requestMessages);
-        const long = content.length > 900000;
-        const model =
-          selectedModel ||
-          (await resolveLlmModel({
-            long,
-            thinking: true,
-          }));
+        const model = availableModels.some((item) => item.id === selectedModel)
+          ? selectedModel
+          : availableModels.find((item) => item.isDefault)?.id ||
+            availableModels[0]?.id;
+        if (!model) {
+          throw new Error('模型列表尚未加载完成，请稍后重试');
+        }
         const enable_search = conversationTopic?.metadata?.enable_search;
         const enable_thinking = true;
         const requestContext = {
@@ -1069,10 +1122,6 @@ export default forwardRef((props, ref) => {
 
   useEffect(() => {
     if (userInfo) {
-      getApplicationRuntimeConfig().catch((error) => {
-        console.error(error);
-        toast.danger(error.message || '应用模型配置读取失败');
-      });
       const usesAuditKnowledge =
         conversationAgent?.key === 'knowledge-copilot' ||
         conversationAgent?.key === 'audit-checklist' ||
@@ -1218,7 +1267,9 @@ export default forwardRef((props, ref) => {
       <InvoiceAssistant
         ref={ref}
         conversation={conversation}
+        models={availableModels}
         onAddConversation={onAddConversation}
+        selectedModel={selectedModel}
         userInfo={userInfo}
       />
     );
@@ -1273,7 +1324,7 @@ export default forwardRef((props, ref) => {
 
     return (
       <>
-        {item.message.auditResult ? (
+        {item.status === 'success' && item.message.auditResult ? (
           <AuditChecklistResult
             aiContent={item.message.content}
             allowWordExport={allowWordExport}

src/pages/Chat/invoice.jsx

@@ -23,7 +23,6 @@ import {
   getApplicationRuntimeConfig,
   llmChat,
   queryMcps,
-  resolveLlmModel,
   updateConversation,
 } from '../../services/DataService';
 
@@ -60,7 +59,13 @@ const SYSTEM_PROMPT = `你叫Mia，是一个有用的的发票智能助手。如
 `;
 
 const InvoiceAssistant = forwardRef((props, ref) => {
-  const { conversation, onAddConversation, userInfo } = props;
+  const {
+    conversation,
+    models = [],
+    onAddConversation,
+    selectedModel,
+    userInfo,
+  } = props;
   const messageApi = useMemo(
     () => ({
       error: toast.danger,
@@ -513,12 +518,12 @@ const InvoiceAssistant = forwardRef((props, ref) => {
             },
           ];
         }
-        const content = JSON.stringify(requestMessages);
-        const long = content.length > 900000;
-        const model = await resolveLlmModel({
-          long,
-          thinking: false,
-        });
+        const model = models.some((item) => item.id === selectedModel)
+          ? selectedModel
+          : models.find((item) => item.isDefault)?.id || models[0]?.id;
+        if (!model) {
+          throw new Error('模型列表尚未加载完成，请稍后重试');
+        }
         const enable_search = false;
         const enable_thinking = false;
 

src/services/DataService.js

@@ -11,16 +11,8 @@ const APPLICATION_NAME =
 
 let applicationConfigPromise = null;
 
-function joinUrl(baseUrl, path) {
-  return `${baseUrl.replace(/\/+$/, '')}/${path.replace(/^\/+/, '')}`;
-}
-
 const BACKEND_LLM_CHAT_PATH = '/api/llm/chat';
 
-function firstDefined(...values) {
-  return values.find((value) => value !== undefined && value !== null);
-}
-
 function firstNonEmptyString(...values) {
   return values.find((value) => typeof value === 'string' && value.trim());
 }
@@ -46,103 +38,15 @@ function normalizeObjectConfig(value) {
   return {};
 }
 
-const LLM_CONTROL_CONFIG_KEYS = new Set([
-  'url',
-  'endpoint',
-  'baseUrl',
-  'base_url',
-  'llmUrl',
-  'llm_url',
-  'llmBaseUrl',
-  'llm_base_url',
-  'path',
-  'chatPath',
-  'chat_path',
-  'llmPath',
-  'llm_path',
-  'llmChatPath',
-  'llm_chat_path',
-  'model',
-  'models',
-  'llmModel',
-  'llm_model',
-  'defaultModel',
-  'default_model',
-  'longModel',
-  'long_model',
-  'thinkingModel',
-  'thinking_model',
-  'thinkingLongModel',
-  'thinking_long_model',
-  'longThinkingModel',
-  'long_thinking_model',
-  'apiKey',
-  'api_key',
-  'apikey',
-  'llmApiKey',
-  'llm_api_key',
-  'authorization',
-  'headers',
-  'request',
-  'params',
-  'body',
-  'llm',
-  'invoiceTool',
-  'invoice_tool',
-  'invoiceToolApiKey',
-  'invoice_tool_api_key',
-  'invoiceToolBaseUrl',
-  'invoice_tool_base_url',
-]);
-
-function pickRequestParams(...configs) {
-  return configs.reduce((params, config) => {
-    const normalizedConfig = normalizeObjectConfig(config);
-    Object.entries(normalizedConfig).forEach(([key, value]) => {
-      if (
-        !LLM_CONTROL_CONFIG_KEYS.has(key) &&
-        value !== undefined &&
-        value !== null
-      ) {
-        params[key] = value;
-      }
-    });
-
-    return params;
-  }, {});
-}
-
 function firstNonEmptyObject(...values) {
   return values.find(
     (value) => Object.keys(normalizeObjectConfig(value)).length,
   );
 }
 
-function normalizeModelConfig(model, fallbackConfig = {}) {
-  const models = normalizeObjectConfig(fallbackConfig.models);
-
-  if (isPlainObject(model)) {
-    return {
-      ...models,
-      ...model,
-    };
-  }
-
-  if (typeof model === 'string' && model.trim()) {
-    return {
-      ...models,
-      default: model.trim(),
-    };
-  }
-
-  return models;
-}
-
 function normalizeApplicationConfig(application = {}) {
   const layoutsConfig = normalizeObjectConfig(application.layouts?.config);
   const settings = application.settings || {};
-  const settingsConfig = normalizeObjectConfig(settings.config);
-  const llmConfig = normalizeObjectConfig(settings.llm);
   const invoiceToolConfig = normalizeObjectConfig(
     firstNonEmptyObject(
       layoutsConfig.invoiceTool,
@@ -151,115 +55,8 @@ function normalizeApplicationConfig(application = {}) {
       settings.invoice_tool,
     ),
   );
-  const modelConfig = normalizeModelConfig(
-    firstDefined(layoutsConfig.model, llmConfig.model, settingsConfig.model),
-    {
-      models: firstDefined(
-        layoutsConfig.models,
-        llmConfig.models,
-        settingsConfig.models,
-      ),
-    },
-  );
-
   return {
     raw: layoutsConfig,
-    llm: {
-      url: firstNonEmptyString(
-        layoutsConfig.url,
-        layoutsConfig.endpoint,
-        layoutsConfig.baseUrl,
-        layoutsConfig.base_url,
-        layoutsConfig.llmUrl,
-        layoutsConfig.llm_url,
-        layoutsConfig.llmBaseUrl,
-        layoutsConfig.llm_base_url,
-        llmConfig.url,
-        llmConfig.endpoint,
-        llmConfig.baseUrl,
-        llmConfig.base_url,
-        settingsConfig.url,
-        settingsConfig.endpoint,
-      ),
-      path: firstNonEmptyString(
-        layoutsConfig.path,
-        layoutsConfig.chatPath,
-        layoutsConfig.chat_path,
-        layoutsConfig.llmPath,
-        layoutsConfig.llm_path,
-        layoutsConfig.llmChatPath,
-        layoutsConfig.llm_chat_path,
-        llmConfig.path,
-        llmConfig.chatPath,
-        llmConfig.chat_path,
-        settingsConfig.path,
-        settingsConfig.chatPath,
-        settingsConfig.chat_path,
-      ),
-      model: firstNonEmptyString(
-        modelConfig.default,
-        layoutsConfig.llmModel,
-        layoutsConfig.llm_model,
-        layoutsConfig.defaultModel,
-        layoutsConfig.default_model,
-        llmConfig.defaultModel,
-        llmConfig.default_model,
-        settingsConfig.defaultModel,
-        settingsConfig.default_model,
-      ),
-      models: {
-        default: firstNonEmptyString(
-          modelConfig.default,
-          layoutsConfig.llmModel,
-          layoutsConfig.llm_model,
-          layoutsConfig.defaultModel,
-          layoutsConfig.default_model,
-          llmConfig.defaultModel,
-          llmConfig.default_model,
-          settingsConfig.defaultModel,
-          settingsConfig.default_model,
-        ),
-        long: firstNonEmptyString(
-          modelConfig.long,
-          layoutsConfig.longModel,
-          layoutsConfig.long_model,
-          llmConfig.longModel,
-          llmConfig.long_model,
-          settingsConfig.longModel,
-          settingsConfig.long_model,
-        ),
-        thinking: firstNonEmptyString(
-          modelConfig.thinking,
-          layoutsConfig.thinkingModel,
-          layoutsConfig.thinking_model,
-          llmConfig.thinkingModel,
-          llmConfig.thinking_model,
-          settingsConfig.thinkingModel,
-          settingsConfig.thinking_model,
-        ),
-        thinkingLong: firstNonEmptyString(
-          modelConfig.thinkingLong,
-          modelConfig.longThinking,
-          layoutsConfig.thinkingLongModel,
-          layoutsConfig.thinking_long_model,
-          layoutsConfig.longThinkingModel,
-          layoutsConfig.long_thinking_model,
-          llmConfig.thinkingLongModel,
-          llmConfig.thinking_long_model,
-          llmConfig.longThinkingModel,
-          llmConfig.long_thinking_model,
-          settingsConfig.thinkingLongModel,
-          settingsConfig.thinking_long_model,
-          settingsConfig.longThinkingModel,
-          settingsConfig.long_thinking_model,
-        ),
-      },
-      requestParams: pickRequestParams(
-        settingsConfig,
-        llmConfig,
-        layoutsConfig,
-      ),
-    },
     invoiceTool: {
       apiKey: firstNonEmptyString(
         invoiceToolConfig.apiKey,
@@ -337,10 +134,6 @@ function normalizeLlmList(payload) {
   return [...new Map(models.map((model) => [model.id, model])).values()];
 }
 
-function resolveLlmUrl() {
-  return BACKEND_LLM_CHAT_PATH;
-}
-
 export function extractLlmChunkContent(value) {
   if (typeof value === 'string') {
     return value;
@@ -664,13 +457,8 @@ export async function callMcpTool(mcpName, toolName, data = {}) {
 }
 
 export async function llmChat(data, signal = undefined) {
-  const runtimeConfig = await getApplicationRuntimeConfig();
-  const llmConfig = runtimeConfig.llm || {};
-  const url = resolveLlmUrl(llmConfig);
-  const model = data.model || llmConfig.model;
-
-  if (!model) {
-    throw new Error('应用配置缺少模型名称：请在 layouts.config.model 中配置');
+  if (!data?.model || typeof data.model !== 'string') {
+    throw new Error('请求缺少模型名称，请先从模型列表中选择可用模型');
   }
 
   const headers = new Headers({
@@ -678,16 +466,12 @@ export async function llmChat(data, signal = undefined) {
     'Content-Type': 'application/json',
   });
 
-  const response = await fetch(url, {
+  const response = await fetch(BACKEND_LLM_CHAT_PATH, {
     method: 'POST',
     headers,
     signal,
     body: JSON.stringify({
-      ...(llmConfig.requestParams && typeof llmConfig.requestParams === 'object'
-        ? llmConfig.requestParams
-        : {}),
       ...data,
-      model,
       stream: true,
     }),
   });
@@ -710,32 +494,23 @@ export async function llmList(signal = undefined) {
     throw new Error('模型列表为空');
   }
 
-  const defaultModel = await resolveLlmModel().catch(() => null);
+  const defaultModel = firstNonEmptyString(
+    payload?.defaultModel,
+    payload?.default_model,
+    payload?.default,
+    payload?.data?.defaultModel,
+    payload?.data?.default_model,
+    payload?.data?.default,
+    payload?.result?.defaultModel,
+    payload?.result?.default_model,
+    payload?.result?.default,
+  );
   return models.map((model) => ({
     ...model,
-    isDefault: model.id === defaultModel,
+    isDefault: Boolean(model.isDefault) || model.id === defaultModel,
   }));
 }
 
-export async function resolveLlmModel(options = {}) {
-  const runtimeConfig = await getApplicationRuntimeConfig();
-  const models = runtimeConfig.llm?.models || {};
-
-  if (options.long && options.thinking && models.thinkingLong) {
-    return models.thinkingLong;
-  }
-
-  if (options.thinking && models.thinking) {
-    return models.thinking;
-  }
-
-  if (options.long && models.long) {
-    return models.long;
-  }
-
-  return models.default || runtimeConfig.llm?.model || null;
-}
-
 export async function queryConversations(query) {
   const userInfo = getUserInfo();
   if (!userInfo) {

src/utils/chatRuntime.js

@@ -182,36 +182,87 @@ export function createSseStream(readableStream) {
       const reader = readableStream.getReader();
       const decoder = new TextDecoder();
       let buffer = '';
+      let streamEnded = false;
 
-      while (true) {
-        const { done, value } = await reader.read();
-        buffer += decoder.decode(value || new Uint8Array(), { stream: !done });
-        const parts = buffer.split(/\r?\n\r?\n/);
-        buffer = parts.pop() || '';
-
-        for (const part of parts) {
+      const parseEvent = (part) => {
         const event = {};
         part.split(/\r?\n/).forEach((line) => {
           const separator = line.indexOf(':');
           if (separator <= 0) return;
           event[line.slice(0, separator)] = line.slice(separator + 1);
         });
-          if (Object.keys(event).length) yield event;
+        return event;
+      };
+
+      const isTerminalEvent = (event) => {
+        const data = String(event.data || '').trim();
+        const eventName = String(event.event || '')
+          .trim()
+          .toLowerCase();
+        if (
+          data === '[DONE]' ||
+          data === 'DONE' ||
+          eventName === 'done' ||
+          eventName === 'end'
+        ) {
+          return true;
+        }
+        if (!data.startsWith('{')) return false;
+
+        try {
+          const payload = JSON.parse(data);
+          if (payload.done === true || payload.finished === true) return true;
+          return Array.isArray(payload.choices)
+            ? payload.choices.some(
+                (choice) =>
+                  choice?.finish_reason !== null &&
+                  choice?.finish_reason !== undefined,
+              )
+            : false;
+        } catch {
+          return false;
+        }
+      };
+
+      try {
+        while (true) {
+          const { done, value } = await reader.read();
+          buffer += decoder.decode(value || new Uint8Array(), {
+            stream: !done,
+          });
+          const parts = buffer.split(/\r?\n\r?\n/);
+          buffer = parts.pop() || '';
+
+          for (const part of parts) {
+            const event = parseEvent(part);
+            if (Object.keys(event).length) {
+              yield event;
+              if (isTerminalEvent(event)) return;
+            }
           }
 
           if (done) {
+            streamEnded = true;
             if (buffer.trim()) {
-            const event = {};
-            buffer.split(/\r?\n/).forEach((line) => {
-              const separator = line.indexOf(':');
-              if (separator <= 0) return;
-              event[line.slice(0, separator)] = line.slice(separator + 1);
-            });
-            if (Object.keys(event).length) yield event;
+              const event = parseEvent(buffer);
+              if (Object.keys(event).length) {
+                yield event;
+                if (isTerminalEvent(event)) return;
+              }
             }
             break;
           }
         }
+      } finally {
+        if (!streamEnded) {
+          try {
+            await reader.cancel();
+          } catch {
+            // The stream may already have been closed by the server.
+          }
+        }
+        reader.releaseLock();
+      }
     },
   };
 }