Merge branch 'main' of https://git.nianxx.cn/wangxuming/makelore

electron/services/learning-agent-client.ts

@@ -50,6 +50,72 @@ function isLearningAccountChanged(error: unknown): boolean {
   return record(error).code === 'LEARNING_ACCOUNT_CHANGED';
 }
 
+function createSseDataParser() {
+  let line = '';
+  let pendingCarriageReturn = false;
+  let dataLines: string[] = [];
+
+  const assertBufferSize = () => {
+    const size = line.length + dataLines.reduce((total, data) => total + data.length, 0);
+    if (size > MAX_SSE_BUFFER_LENGTH) throw new Error('助教服务暂时不可用');
+  };
+
+  const commitLine = (events: string[]) => {
+    if (line === '') {
+      if (dataLines.length > 0) events.push(dataLines.join('\n'));
+      dataLines = [];
+      return;
+    }
+    if (!line.startsWith(':')) {
+      const separator = line.indexOf(':');
+      const field = separator === -1 ? line : line.slice(0, separator);
+      let value = separator === -1 ? '' : line.slice(separator + 1);
+      if (value.startsWith(' ')) value = value.slice(1);
+      if (field === 'data') {
+        dataLines.push(value);
+        assertBufferSize();
+      }
+    }
+    line = '';
+  };
+
+  const push = (chunk: string): string[] => {
+    const events: string[] = [];
+    for (const character of chunk) {
+      if (pendingCarriageReturn) {
+        pendingCarriageReturn = false;
+        commitLine(events);
+        if (character === '\n') continue;
+      }
+      if (character === '\r') {
+        pendingCarriageReturn = true;
+      } else if (character === '\n') {
+        commitLine(events);
+      } else {
+        line += character;
+        assertBufferSize();
+      }
+    }
+    return events;
+  };
+
+  const finish = (): string[] => {
+    const events: string[] = [];
+    if (pendingCarriageReturn) {
+      pendingCarriageReturn = false;
+      commitLine(events);
+    }
+    if (line !== '') commitLine(events);
+    if (dataLines.length > 0) {
+      events.push(dataLines.join('\n'));
+      dataLines = [];
+    }
+    return events;
+  };
+
+  return { push, finish };
+}
+
 export function createLearningAgentClient(dependencies: Dependencies = {}) {
   const fetchImpl = dependencies.fetchImpl ?? proxyAwareFetch;
   const getAccessToken = dependencies.getAccessToken ?? getValidWorksSquareAccessToken;
@@ -217,40 +283,47 @@ export function createLearningAgentClient(dependencies: Dependencies = {}) {
     }
     const reader = response.body.getReader();
     const decoder = new TextDecoder();
-    let buffer = '';
+    const parser = createSseDataParser();
     let text = '';
+    const consume = (data: string): { text?: string; completed?: true } => {
+      if (!data) return {};
+      let envelope: Record<string, unknown>;
+      try {
+        envelope = record(JSON.parse(data));
+      } catch {
+        throw new Error('助教服务暂时不可用');
+      }
+      if (envelope.run_id !== runId) return {};
+      const payload = record(envelope.payload);
+      if (envelope.type === 'learning.assistant.delta' && typeof payload.delta === 'string') {
+        if (text.length + payload.delta.length > MAX_ANSWER_LENGTH) throw new Error('助教服务暂时不可用');
+        text += payload.delta;
+      }
+      if (envelope.type === 'learning.assistant.failed') {
+        throw new Error('助教回答失败');
+      }
+      if (envelope.type === 'learning.assistant.completed') {
+        const answer = text.trim();
+        if (!answer) throw new Error('助教回答失败');
+        return { text: answer, completed: true };
+      }
+      return {};
+    };
     try {
       while (true) {
         const chunk = await reader.read();
         assertCurrentAccount();
-        if (chunk.done) break;
-        buffer += decoder.decode(chunk.value, { stream: true });
-        if (buffer.length > MAX_SSE_BUFFER_LENGTH) throw new Error('助教服务暂时不可用');
-        const frames = buffer.split('\n\n');
-        buffer = frames.pop() || '';
-        for (const frame of frames) {
-          const data = frame.split('\n').find((line) => line.startsWith('data:'))?.slice(5).trim();
-          if (!data) continue;
-          let envelope: Record<string, unknown>;
-          try {
-            envelope = record(JSON.parse(data));
-          } catch {
-            throw new Error('助教服务暂时不可用');
-          }
-          if (envelope.run_id !== runId) continue;
-          const payload = record(envelope.payload);
-          if (envelope.type === 'learning.assistant.delta' && typeof payload.delta === 'string') {
-            if (text.length + payload.delta.length > MAX_ANSWER_LENGTH) throw new Error('助教服务暂时不可用');
-            text += payload.delta;
-          }
-          if (envelope.type === 'learning.assistant.failed') throw new Error('助教回答失败');
-          if (envelope.type === 'learning.assistant.completed') {
+        const events = chunk.done
+          ? [...parser.push(decoder.decode()), ...parser.finish()]
+          : parser.push(decoder.decode(chunk.value, { stream: true }));
+        for (const data of events) {
+          const result = consume(data);
+          if (result.completed) {
             controller.abort();
-            const answer = text.trim();
-            if (!answer) throw new Error('助教回答失败');
-            return { text: answer };
+            return { text: result.text ?? '' };
           }
         }
+        if (chunk.done) break;
       }
     } finally {
       clearTimeout(timeout);