Merge remote-tracking branch 'origin/master' into codex/concrete-provenance-prose

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: 6ad674ebdf8da4fd927a9499e80e06462d3c0dfb
-README.zh.md: ca51a6f56ab2fa3bf47ca4f400a1474440bb39a4
+README.md: a21f9f0464e9d43d2091bd446eb123d4d0990c3d
+README.zh.md: 2c45f2144694785590b339642cb62b62a1aa4198

packages/llm/llm-deepseek/README.md

@@ -45,7 +45,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

@@ -45,7 +45,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

@@ -216,7 +216,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) {
@@ -257,6 +263,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
@@ -322,6 +329,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

@@ -548,6 +548,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])

packages/util/timeout/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/util/timeout/README.md
-README.md: 0ff5550ef7ea6b8315a6a529a4b8b503162a8f12
-README.zh.md: 79d7ee674209b0324ae9428c1a04fda8a2547db5
+README.md: e0d3595e7771fd19c808e89d440e69bca40da286
+README.zh.md: 99b7f281c2af7746fc2a105ee08df870fd3a9583

packages/util/timeout/README.md

@@ -18,7 +18,7 @@ import { clampTimeout, deadline, idleWatchdog, MAX_TIMER_DELAY_MS, timeoutOf, Ti
 |---|---|
 | `clampTimeout(requested, def, max, name?)` | Validate the caller's optional positive-finite hint, fill from `def`, cap at `max`. Throws (with `name`) on a non-positive/non-finite hint. |
 | `deadline(upstream, timeoutMs, code)` | Fuse `upstream` cancellation with a timeout into one `AbortSignal` (`AbortSignal.any`); the timeout carries a `TimeoutReason`. `[Symbol.dispose]` clears the timer. |
-| `idleWatchdog(upstream, timeoutMs, code)` | Keep one stable fused signal and arm only while its guarded async-iterator `next()` is outstanding. Resolution disarms; later demand rearms; disposal clears; concurrent demand rejects. |
+| `idleWatchdog(upstream, timeoutMs, code)` | Keep one stable fused signal and arm only while its guarded async-iterator `next()` is outstanding. Resolution disarms; later demand or `pulse()` activity rearms; disposal clears; concurrent demand rejects. |
 | `MAX_TIMER_DELAY_MS` | Largest delay Node schedules without clamping it to one millisecond (`2_147_483_647`). Timer-owning config must not exceed it. |
 | `timeoutOf(signal \| { reason }, code?)` | Recover the `TimeoutReason` from an aborted signal/error, else `undefined` — the timeout-vs-cancel classifier. Pass `code` to match only THIS deadline's timer (see nesting below). |
 | `TimeoutReason` | The internal reason (`code` + `timeoutMs`) stamped on a timeout abort. Not a public error — providers translate it into their own error/field. |
@@ -48,7 +48,7 @@ The signal only *notifies* — the caller MUST attach its own termination (`d.si
 
 Pass your own `code` to `timeoutOf` so classification composes under nesting. When `upstream` is itself a deadline signal, `AbortSignal.any` preserves its `TimeoutReason` if that timer fires first. Scoping to your code makes a foreign timeout read as an ordinary upstream cancel instead of claiming that the local timer expired.
 
-For a streamed transport, create one `idleWatchdog`, pass its stable `signal` into the transport, and call `watchdog.next(iterator)` for each provider read. The interval must be positive, finite, and no greater than `MAX_TIMER_DELAY_MS`; Node otherwise clamps it to one millisecond. It measures only outstanding demand, so no timer runs while downstream code renders or otherwise waits before asking for the next chunk. The primitive still only notifies, so the transport must observe the stable signal; the DeepSeek and pi-ai adapters prove that timeout closes their real response body or SDK request.
+For a streamed transport, create one `idleWatchdog`, pass its stable `signal` into the transport, and call `watchdog.next(iterator)` for each provider read. Call `watchdog.pulse()` when transport activity does not yield an iterator value. The interval must be positive, finite, and no greater than `MAX_TIMER_DELAY_MS`; Node otherwise clamps it to one millisecond. It measures only outstanding demand, so no timer runs while downstream code renders or otherwise waits before asking for the next chunk. The primitive still only notifies, so the transport must observe the stable signal; the DeepSeek and pi-ai adapters prove that timeout closes their real response body or SDK request.
 
 ## What does NOT get a timeout
 

packages/util/timeout/README.zh.md

@@ -18,7 +18,7 @@ import { clampTimeout, deadline, idleWatchdog, MAX_TIMER_DELAY_MS, timeoutOf, Ti
 |---|---|
 | `clampTimeout(requested, def, max, name?)` | 验证调用方可选的、值为正且有限的提示，从 `def` 填充，并限制在 `max` 以内。如果提示为非正数或非有限数，则抛出错误（包含 `name`）。 |
 | `deadline(upstream, timeoutMs, code)` | 将 `upstream` 取消与超时融合为一个 `AbortSignal`（`AbortSignal.any`）；超时携带 `TimeoutReason`。`[Symbol.dispose]` 清除 timer。 |
-| `idleWatchdog(upstream, timeoutMs, code)` | 保持一个稳定的融合信号，并且只在受保护的异步迭代器 `next()` 尚未完成时启动 timer。完成后停止 timer；后续需求重新启动 timer；dispose（资源释放）时清除；并发需求被拒绝。 |
+| `idleWatchdog(upstream, timeoutMs, code)` | 保持一个稳定的融合信号，并且只在受保护的异步迭代器 `next()` 尚未完成时启动 timer。完成后停止 timer；后续需求或 `pulse()` 活动会重新启动 timer；dispose（资源释放）时清除；并发需求被拒绝。 |
 | `MAX_TIMER_DELAY_MS` | Node 在不将延迟限制为 1 毫秒时可调度的最大延迟（`2_147_483_647`）。负责 timer 的配置不得超过该值。 |
 | `timeoutOf(signal \| { reason }, code?)` | 从已中止的信号/错误中恢复 `TimeoutReason`，否则返回 `undefined`，即超时与取消的分类器。传入 `code` 可仅匹配这个 deadline 的 timer（见下文的嵌套）。 |
 | `TimeoutReason` | 标记在超时中止上的内部原因（`code` + `timeoutMs`）。它不是公开错误；提供方将其转换为自己的错误/字段。 |
@@ -48,7 +48,7 @@ export async function runWithDeadline(upstream: AbortSignal | undefined, timeout
 
 将你自己的 `code` 传给 `timeoutOf`，使分类可在嵌套场景中正确组合。当 `upstream` 本身是 deadline 信号时，如果该 timer 先触发，`AbortSignal.any` 会保留它的 `TimeoutReason`。将匹配范围限定为你的 code，会把外部超时视为普通的 upstream 取消，而不会声称本地 timer 已到期。
 
-对于流式传输，创建一个 `idleWatchdog`，将其稳定的 `signal` 传给传输层，并为提供方的每次读取调用 `watchdog.next(iterator)`。间隔必须为正有限数，且不得超过 `MAX_TIMER_DELAY_MS`；否则 Node 会将其限制为 1 毫秒。它只对尚未完成的读取请求计时，因此当下游代码进行渲染或在请求下一个分片前以其他方式等待时，timer 不会运行。该原语仍然只会通知，因此传输层必须观察稳定信号；DeepSeek 和 pi-ai 适配器证明，超时会关闭它们的真实响应正文或 SDK 请求。
+对于流式传输，创建一个 `idleWatchdog`，将其稳定的 `signal` 传给传输层，并为提供方的每次读取调用 `watchdog.next(iterator)`。当传输活动不产生迭代器值时，调用 `watchdog.pulse()`。间隔必须为正有限数，且不得超过 `MAX_TIMER_DELAY_MS`；否则 Node 会将其限制为 1 毫秒。它只对尚未完成的读取请求计时，因此当下游代码进行渲染或在请求下一个分片前以其他方式等待时，timer 不会运行。该原语仍然只会通知，因此传输层必须观察稳定信号；DeepSeek 和 pi-ai 适配器证明，超时会关闭它们的真实响应正文或 SDK 请求。
 
 ## 哪些操作不设置超时
 

packages/util/timeout/src/index.ts

@@ -72,6 +72,8 @@ export interface IdleWatchdog {
    * @returns the iterator's next result.
    */
   next<T>(iterator: AsyncIterator<T>): Promise<IteratorResult<T>>
+  /** Rearm an outstanding demand after transport activity that yields no iterator value; otherwise a no-op. */
+  pulse(): void
   /** Clear an armed timer; safe to call once at the owning stream's exit. */
   [Symbol.dispose](): void
 }
@@ -135,15 +137,20 @@ export function idleWatchdog(
   let outstanding = false
   let disposed = false
 
+  const arm = (): void => {
+    if (timer !== undefined) clearTimeout(timer)
+    timer = setTimeout(() => {
+      timeout.abort(new TimeoutReason(code, timeoutMs))
+    }, timeoutMs)
+  }
+
   return {
     signal,
     async next<T>(iterator: AsyncIterator<T>): Promise<IteratorResult<T>> {
       if (disposed) throw new Error('idleWatchdog is disposed')
       if (outstanding) throw new Error('idleWatchdog next is already outstanding')
       outstanding = true
-      timer = setTimeout(() => {
-        timeout.abort(new TimeoutReason(code, timeoutMs))
-      }, timeoutMs)
+      arm()
       try {
         return await iterator.next()
       } finally {
@@ -152,6 +159,10 @@ export function idleWatchdog(
         outstanding = false
       }
     },
+    pulse(): void {
+      if (disposed || !outstanding) return
+      arm()
+    },
     [Symbol.dispose](): void {
       if (disposed) return
       disposed = true

packages/util/timeout/tests/timeout.spec.ts

@@ -229,6 +229,28 @@ describe('idleWatchdog', () => {
     await expect(secondNext).rejects.toBe(stableSignal.reason)
   })
 
+  it('rearms outstanding demand on an out-of-band activity pulse', async () => {
+    vi.useFakeTimers()
+    const pending = Promise.withResolvers<IteratorResult<number>>()
+    const watchdog = idleWatchdog(undefined, 100, 'LLM_STREAM_IDLE_TIMEOUT')
+    watchdog.pulse()
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(watchdog.signal.aborted).toBe(false)
+
+    const next = watchdog.next({ next: () => pending.promise })
+    await vi.advanceTimersByTimeAsync(99)
+    watchdog.pulse()
+    await vi.advanceTimersByTimeAsync(99)
+    expect(watchdog.signal.aborted).toBe(false)
+    await vi.advanceTimersByTimeAsync(1)
+    expect(timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT')).toMatchObject({ timeoutMs: 100 })
+    pending.reject(watchdog.signal.reason)
+    await expect(next).rejects.toBe(watchdog.signal.reason)
+
+    watchdog[Symbol.dispose]()
+    watchdog.pulse()
+  })
+
   it('keeps an earlier upstream abort distinct from its own timeout', async () => {
     vi.useFakeTimers()
     const upstream = new AbortController()