fix(llm): honor DeepSeek SSE keep-alives

packages/llm/llm-deepseek/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/llm/llm-deepseek/README.md
-README.md: 020aa65073495526be3f32912b7cd06667c52a2e
-README.zh.md: 4c655e90ba00340c056f6ac16159621f7a8c1ddb
+README.md: 2ecdb4330e5871bbaf0da6fc583083a06c935486
+README.zh.md: 63eb7be330806b668e12867ed906d0d88acaff1f

packages/llm/llm-deepseek/README.md

@@ -46,7 +46,7 @@ The same exact-model result exposes ordered `off`, `high`, and `max` efforts und
 
 `thinking: disabled` is a deployment lock that publishes only `off` with `off` as its default. Omitting `reasoningEffort` or configuring it as `off` is valid; configuring `high` or `max` fails plugin loading, and a direct per-request attempt to enable thinking fails before network I/O. A request with `GenerateOptions.purpose: 'session-title'` also forces thinking disabled and omits the already-resolved effort, reserving its bounded output for visible title text without changing conversation or compaction defaults.
 
-`streamIdleTimeoutMs` bounds each outstanding provider read, including the initial `fetch`, without counting time the consumer spends between chunks. One stable abort signal reaches the request and body reader for the whole call; expiry stops the transport and throws `LlmError('TIMEOUT')`, while an earlier caller abort throws `LlmError('ABORTED')`. The adapter makes exactly one provider request per `stream()` call; it registers the configured policy as provider metadata, and `dsh-llm-retry` separately executes it at durable agent-step boundaries.
+`streamIdleTimeoutMs` bounds each outstanding provider read, including the initial `fetch`, without counting time the consumer spends between chunks. DeepSeek SSE comments rearm an outstanding read as transport activity but never become `StreamChunk` values or session-log events. One stable abort signal reaches the request and body reader for the whole call; expiry stops the transport and throws `LlmError('TIMEOUT')`, while an earlier caller abort throws `LlmError('ABORTED')`. The adapter makes exactly one provider request per `stream()` call; it registers the configured policy as provider metadata, and `dsh-llm-retry` separately executes it at durable agent-step boundaries.
 
 ## Dynamic configuration (settings + credentials)
 

packages/llm/llm-deepseek/README.zh.md

@@ -46,7 +46,7 @@ harness LLM（大语言模型）seam 的 DeepSeek chat-completions 适配器：
 
 `thinking: disabled` 是部署锁定：它只公布 `off`，并以 `off` 为默认值。省略 `reasoningEffort` 或将其配置为 `off` 均有效；配置 `high` 或 `max` 会使插件加载失败，直接按请求启用思考也会在网络 I/O 前失败。携带 `GenerateOptions.purpose: 'session-title'` 的请求也会强制禁用思考并省略已解析的推理强度，将有界输出保留给可见标题文本，不改变会话或压缩（compaction）默认值。
 
-`streamIdleTimeoutMs` 会限制每次未完成提供方读取，包括初始 `fetch`，但不计入消费方在分片间花费的时间。同一个稳定的 abort 信号会在整个调用期间传递给请求与 body reader；过期会停止传输并抛出 `LlmError('TIMEOUT')`，较早的调用方 abort 则抛出 `LlmError('ABORTED')`。适配器每次 `stream()` 调用恰好发起一次提供方请求；它把已配置策略注册为提供方元数据，再由 `dsh-llm-retry` 在持久化的 agent（智能体）步骤边界单独执行该策略。
+`streamIdleTimeoutMs` 会限制每次未完成提供方读取，包括初始 `fetch`，但不计入消费方在分片间花费的时间。DeepSeek SSE 注释会作为传输活动使尚未完成的读取重新布防，但绝不会成为 `StreamChunk` 值或会话日志事件。同一个稳定的 abort 信号会在整个调用期间传递给请求与 body reader；过期会停止传输并抛出 `LlmError('TIMEOUT')`，较早的调用方 abort 则抛出 `LlmError('ABORTED')`。适配器每次 `stream()` 调用恰好发起一次提供方请求；它把已配置策略注册为提供方元数据，再由 `dsh-llm-retry` 在持久化的 agent（智能体）步骤边界单独执行该策略。
 
 ## 动态配置（settings + credentials）
 

packages/llm/llm-deepseek/src/adapter.ts

@@ -215,7 +215,13 @@ export class DeepSeekAdapter extends LlmAdapter {
       ? consumer.signal
       : AbortSignal.any([options.signal, consumer.signal])
     using watchdog = idleWatchdog(upstream, connection.streamIdleTimeoutMs, STREAM_IDLE_TIMEOUT_CODE)
-    const iterator = this.request(options, watchdog.signal, connection, apiKey)[Symbol.asyncIterator]()
+    const iterator = this.request(
+      options,
+      watchdog.signal,
+      connection,
+      apiKey,
+      () => { watchdog.pulse() },
+    )[Symbol.asyncIterator]()
     let exhausted = false
     try {
       while (true) {
@@ -256,6 +262,7 @@ export class DeepSeekAdapter extends LlmAdapter {
     signal: AbortSignal,
     connection: DeepSeekConnectionOptions,
     apiKey: string,
+    onComment: () => void,
   ): AsyncIterable<StreamChunk> {
     const body = serializeRequest(options, connection.defaults)
     // Prepared outside the try so the TRANSPORT label below covers exactly the
@@ -321,6 +328,6 @@ export class DeepSeekAdapter extends LlmAdapter {
       throw new LlmError('DeepSeek API returned no response body', 'EMPTY_RESPONSE')
     }
 
-    yield* translate(parseSse(response.body))
+    yield* translate(parseSse(response.body, onComment))
   }
 }

packages/llm/llm-deepseek/src/sse.ts

@@ -1,11 +1,12 @@
 /**
  * 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.
+ * multi-`data:` joining — is `eventsource-parser`'s. Comments are reported
+ * only through an optional transport-activity callback. This module keeps 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.
  *
  * @module dsh-llm-deepseek/sse
  */
@@ -21,12 +22,16 @@ export const DONE = '[DONE]'
  * 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.
+ * @param onComment - optional transport-activity callback; comments never enter the yielded payload stream.
  * @returns each event's data payload in arrival order, the `[DONE]` sentinel last.
  */
-export async function* parseSse(stream: ReadableStream<BufferSource>): AsyncGenerator<string> {
+export async function* parseSse(
+  stream: ReadableStream<BufferSource>,
+  onComment?: (comment: string) => void,
+): AsyncGenerator<string> {
   const events = stream
     .pipeThrough(new TextDecoderStream())
-    .pipeThrough(new EventSourceParserStream())
+    .pipeThrough(new EventSourceParserStream({ onComment }))
   for await (const { data } of events) {
     yield data
     if (data === DONE) return

packages/llm/llm-deepseek/tests/adapter.spec.ts

@@ -545,6 +545,40 @@ describe('DeepSeekAdapter against a mock server', () => {
       fetchSpy.mockRestore()
     }
   })
+
+  it('keeps an idle provider read alive through SSE comments', async () => {
+    vi.useFakeTimers()
+    const encoder = new TextEncoder()
+    const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(() => {
+      const body = new ReadableStream<Uint8Array>({
+        start(controller) {
+          setTimeout(() => { controller.enqueue(encoder.encode(': keep-alive\n\n')) }, 75)
+          setTimeout(() => { controller.enqueue(encoder.encode(': keep-alive\n\n')) }, 150)
+          setTimeout(() => {
+            controller.enqueue(encoder.encode(textEvents.map(event => `data: ${event}\n\n`).join('')))
+            controller.close()
+          }, 225)
+        },
+      })
+      return Promise.resolve(new Response(body, { status: 200 }))
+    })
+    const adapter = adapterOf({ baseURL: 'https://example.invalid', streamIdleTimeoutMs: 100 })
+    try {
+      const chunks: string[] = []
+      const drain = (async () => {
+        for await (const chunk of adapter.stream({ provider: 'deepseek-official', model: 'm', messages: [] })) {
+          chunks.push(chunk.type)
+        }
+      })()
+      await vi.advanceTimersByTimeAsync(75)
+      await vi.advanceTimersByTimeAsync(75)
+      await vi.advanceTimersByTimeAsync(75)
+      await expect(drain).resolves.toBeUndefined()
+      expect(chunks).toEqual(['block-start', 'text-delta', 'block-end', 'usage', 'finish'])
+    } finally {
+      fetchSpy.mockRestore()
+    }
+  })
 })
 
 describe('plugin registration and config', () => {

packages/llm/llm-deepseek/tests/sse.spec.ts

@@ -31,6 +31,16 @@ describe('parseSse', () => {
     expect(events).toEqual(['{"a":1}', DONE])
   })
 
+  it('reports comments out of band without yielding them', async () => {
+    const comments: string[] = []
+    const events = await collect(parseSse(
+      bytes(': keep-alive\n\ndata: {"a":1}\n\ndata: [DONE]\n\n'),
+      (comment) => { comments.push(comment) },
+    ))
+    expect(comments).toEqual(['keep-alive'])
+    expect(events).toEqual(['{"a":1}', DONE])
+  })
+
   it('stops yielding after DONE even when more data follows', async () => {
     const events = await collect(parseSse(bytes('data: [DONE]\n\ndata: {"late":1}\n\n')))
     expect(events).toEqual([DONE])