refactor(llm-deepseek): replace hand-rolled SSE parser with eventsource-parser

Implements the approved simplification Agent Note: sse.ts now pipes the
response body through TextDecoderStream and EventSourceParserStream
(eventsource-parser/stream) and keeps only the DeepSeek protocol shim —
yield each event's data, terminate on [DONE], throw
LlmError('STREAM_CLOSED') on EOF without the sentinel. The SSE
spec-conformance tests are deleted; sse.spec.ts pins only the
[DONE]/STREAM_CLOSED/EOF contract, including the new spec-strict verdict
that an unterminated trailing event is truncation (the old parser
flushed it — a robustness nicety no real provider shape needs).

eventsource-parser@^3.1.0 becomes llm-deepseek's second runtime
dependency (already in the lockfile transitively via the MCP SDK).

Docs: the Agent Note moves proposed/ → implemented/ and is rewritten per
the lifecycle contract; the rejected NIH roll-up note's inbound links
follow. The twin-adapters note, dsh-llm LlmAdapter JSDoc (and its
type-equiv fences), cookbook, group/package READMEs, root AGENTS.md
layout line, sdk-helper comments, and the regenerated config catalog
drop the "hand-rolled fetch + SSE" claim in both languages; all eight
touched pairs re-recorded.
This commit is contained in:
Tianyi Cui
2026-07-26 22:45:33 +08:00
parent c9dc097749
commit c50b899015
36 changed files with 143 additions and 231 deletions

View File

@@ -20,7 +20,7 @@ import { parseSse } from './sse.ts'
import { translate } from './translate.ts'
import type { WireError } from './types.ts'
/** One optional model entry advertised by the hand-written adapter. */
/** One optional model entry advertised by the direct-fetch adapter. */
export interface DeepSeekCatalogModel {
/** Wire model id accepted by the configured endpoint. */
id: string

View File

@@ -1,65 +1,33 @@
/**
* Decode an SSE byte stream into event `data` payloads. Network reads may split UTF-8 or lines;
* CRLF, comments, non-data fields, and multi-data events are handled per SSE rules. The literal
* `[DONE]` is yielded so the caller owns final flushing, and EOF before it raises {@link LlmError}.
* Decode an SSE byte stream into event `data` payloads. Framing — chunk
* reassembly, UTF-8/CRLF/BOM handling, comment and non-data field skipping,
* multi-`data:` joining — is `eventsource-parser`'s; this module keeps only
* the DeepSeek protocol: the literal `[DONE]` is yielded so the caller owns
* final flushing, and EOF before it raises {@link LlmError}. Framing is
* spec-strict: an event dispatches only on its blank-line terminator, so an
* unterminated tail at EOF is truncation, not a flushable payload.
*
* Minimal SSE (text/event-stream) parser for the chat-completions stream.
* @module dsh-llm-deepseek/sse
*/
import { EventSourceParserStream } from 'eventsource-parser/stream'
import { LlmError } from '@deepseek-ai/dsh-llm'
/** The terminal payload DeepSeek (and OpenAI) send after the last chunk. */
export const DONE = '[DONE]'
/** Extract the joined data payload from one raw SSE event block. */
function eventData(block: string): string | undefined {
const data: string[] = []
for (const rawLine of block.split('\n')) {
const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine
if (line.startsWith('data:')) {
// The spec strips ONE leading space after the colon.
data.push(line.startsWith('data: ') ? line.slice(6) : line.slice(5))
}
// Comments (':…') and other fields (event:, id:, retry:) are ignored.
}
if (data.length === 0) return undefined
return data.join('\n')
}
/**
* Parse a byte stream into SSE data payloads. Yields `[DONE]` as the final
* Parse an SSE byte stream into data payloads. Yields `[DONE]` as the final
* value and returns; throws `LlmError('STREAM_CLOSED')` when the stream ends
* without it (truncated response — the model call cannot be trusted).
* @param stream - raw SSE bytes; reads may split anywhere, including mid-UTF-8 sequence.
* @returns each event's data payload in arrival order, the `[DONE]` sentinel last.
*/
export async function* parseSse(stream: AsyncIterable<Uint8Array>): AsyncGenerator<string> {
const decoder = new TextDecoder()
let buffer = ''
for await (const bytes of stream) {
buffer += decoder.decode(bytes, { stream: true })
// Events are separated by a blank line (\n\n; tolerate \r\n\r\n via the
// per-line \r strip in eventData and a normalized split here).
let boundary: number
while ((boundary = buffer.search(/\r?\n\r?\n/)) !== -1) {
const matched = /\r?\n\r?\n/.exec(buffer.slice(boundary))
const block = buffer.slice(0, boundary)
// matched cannot be null: search() just found the same pattern at 0.
buffer = buffer.slice(boundary + (matched as RegExpExecArray)[0].length)
const data = eventData(block)
if (data === undefined) continue
yield data
if (data === DONE) return
}
}
// Flush any final un-terminated event (servers usually end with \n\n, but
// a trailing block without one is still parseable).
buffer += decoder.decode()
const data = eventData(buffer)
if (data !== undefined) {
export async function* parseSse(stream: ReadableStream<BufferSource>): AsyncGenerator<string> {
const events = stream
.pipeThrough(new TextDecoderStream())
.pipeThrough(new EventSourceParserStream())
for await (const { data } of events) {
yield data
if (data === DONE) return
}