Merge pull request #1375 from deepseek-harness/codex/fix-deepseek-keepalive
fix(llm): honor DeepSeek SSE keep-alives
This commit is contained in:
fz
@@ -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 .agents/notes/implemented/architecture/2026-06-21-bounded-llm-request-recovery.md
|
||||
2026-06-21-bounded-llm-request-recovery.md: 587f2d26eee922e91c8797018964f983890eb8ec
|
||||
2026-06-21-bounded-llm-request-recovery.zh.md: dfc5e3ab67b62042c37f05199ed39bf47ded0c5d
|
||||
2026-06-21-bounded-llm-request-recovery.md: 013cf5d77b0cddc38af8bcf15f13142022cb3f75
|
||||
2026-06-21-bounded-llm-request-recovery.zh.md: 9b025eeed208d42b5ca14fdec33e45aade50f0d2
|
||||
|
||||
@@ -74,9 +74,9 @@ Adapters perform one provider request per `stream()` call. The pi-ai adapter rem
|
||||
|
||||
### Bound stalled streams where they can be stopped
|
||||
|
||||
Each adapter exposes a validated `streamIdleTimeoutMs` configuration field with the five-minute prior-art default cited above. The interval is capped at Node's maximum timer delay so it cannot be clamped to one millisecond. It covers each outstanding iterator `next()` from demand to the next valid `StreamChunk`; time a consumer spends between `next()` calls is not provider idle time.
|
||||
Each adapter exposes a validated `streamIdleTimeoutMs` configuration field with the five-minute prior-art default cited above. The interval is capped at Node's maximum timer delay so it cannot be clamped to one millisecond. It covers each outstanding iterator `next()` from demand to adapter-recognized provider activity; time a consumer spends between `next()` calls is not provider idle time. DeepSeek SSE comments count as transport activity but never become `StreamChunk` values or session-log events.
|
||||
|
||||
`@deepseek-ai/dsh-timeout` exposes a rearmable idle-watchdog primitive. One stable local `AbortController` is fused with the caller signal and passed to the transport for the whole adapter call; each outstanding `next()` arms the watchdog, resolution disarms it, and the next demand rearms it. Timeout aborts that stable controller with a capability-owned `TimeoutReason`, and `finally` clears the timer. The adapter classifies its watchdog as `TIMEOUT` and an earlier upstream abort as `ABORTED`. The existing one-shot `deadline()` is not presented as a sliding timer.
|
||||
`@deepseek-ai/dsh-timeout` exposes a rearmable idle-watchdog primitive. One stable local `AbortController` is fused with the caller signal and passed to the transport for the whole adapter call; each outstanding `next()` arms the watchdog, resolution disarms it, and the next demand rearms it. Out-of-band transport activity calls `pulse()` to rearm an outstanding demand without yielding a value. Timeout aborts that stable controller with a capability-owned `TimeoutReason`, and `finally` clears the timer. The adapter classifies its watchdog as `TIMEOUT` and an earlier upstream abort as `ABORTED`. The existing one-shot `deadline()` is not presented as a sliding timer.
|
||||
|
||||
Boundary tests prove termination at both actual transports. The hand-written adapter aborts its fetch/reader, and the pi-ai adapter maps the stable signal through the SDK and proves the SDK closes the response. A timer that merely rejects a consumer promise while leaving the request running does not satisfy the contract.
|
||||
|
||||
|
||||
@@ -74,9 +74,9 @@ agent-spine 演示组合包加载该插件，因此共享的 stdio/TUI、一次
|
||||
|
||||
### 在能够终止停滞流的位置施加边界
|
||||
|
||||
每个适配器都公开一个经过验证的 `streamIdleTimeoutMs` 配置字段，默认值采用上文引用的五分钟先例。该间隔不超过 Node 的最大定时器延迟，因此不会被钳制为 1 毫秒。它覆盖每个尚未完成的迭代器 `next()`：从消费方请求下一项开始，到下一条有效 `StreamChunk` 到达为止；消费方在两次 `next()` 调用之间花费的时间不属于提供方空闲时间。
|
||||
每个适配器都公开一个经过验证的 `streamIdleTimeoutMs` 配置字段，默认值采用上文引用的五分钟先例。该间隔不超过 Node 的最大定时器延迟，因此不会被钳制为 1 毫秒。它覆盖每个尚未完成的迭代器 `next()`：从消费方请求下一项开始，到适配器识别到提供方活动为止；消费方在两次 `next()` 调用之间花费的时间不属于提供方空闲时间。DeepSeek SSE（Server-Sent Events）注释计为传输活动，但绝不会成为 `StreamChunk` 值或会话日志事件。
|
||||
|
||||
`@deepseek-ai/dsh-timeout` 公开一个可重新布防的空闲看门狗原语。一个稳定的局部 `AbortController` 会与调用方信号融合，并在整个适配器调用期间传给传输层；每个尚未完成的 `next()` 都会布防看门狗，该调用完成时解除布防，下一次请求数据时再重新布防。超时会使用能力自身拥有的 `TimeoutReason` 中止这个稳定控制器，`finally` 则会清除定时器。适配器将自身看门狗归类为 `TIMEOUT`，将更早发生的上游中止归类为 `ABORTED`。现有的一次性 `deadline()` 不会被描述为滑动计时器。
|
||||
`@deepseek-ai/dsh-timeout` 公开一个可重新布防的空闲看门狗原语。一个稳定的局部 `AbortController` 会与调用方信号融合，并在整个适配器调用期间传给传输层；每个尚未完成的 `next()` 都会布防看门狗，该调用完成时解除布防，下一次请求数据时再重新布防。带外传输活动会调用 `pulse()`，在不产生值的情况下为尚未完成的需求重新布防。超时会使用能力自身拥有的 `TimeoutReason` 中止这个稳定控制器，`finally` 则会清除定时器。适配器将自身看门狗归类为 `TIMEOUT`，将更早发生的上游中止归类为 `ABORTED`。现有的一次性 `deadline()` 不会被描述为滑动计时器。
|
||||
|
||||
边界测试证明两个实际传输层都能终止。手写适配器会中止其 fetch／reader，pi-ai 适配器会把稳定信号映射到 SDK，并证明 SDK 会关闭响应。如果定时器只拒绝消费方 promise，却让请求继续运行，就不满足此约定。
|
||||
|
||||
|
||||
@@ -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 .agents/notes/implemented/architecture/2026-07-06-timeout-deadline-library.md
|
||||
2026-07-06-timeout-deadline-library.md: 63463a76a65743436d4e78479800c19e257a42de
|
||||
2026-07-06-timeout-deadline-library.zh.md: c47e6d4e6ed0bec368fca8857e5bd4d7229b6b0e
|
||||
2026-07-06-timeout-deadline-library.md: 1afb8e326356f012498eed778e9bab1b0be64ea5
|
||||
2026-07-06-timeout-deadline-library.zh.md: c6992ba5463122e4eb4ebd1d904d0c0906140353
|
||||
|
||||
@@ -57,10 +57,11 @@ export function deadline(
|
||||
export interface IdleWatchdog {
|
||||
readonly signal: AbortSignal
|
||||
next<T>(iterator: AsyncIterator<T>): Promise<IteratorResult<T>>
|
||||
pulse(): void
|
||||
[Symbol.dispose](): void
|
||||
}
|
||||
|
||||
/** Arm only while one iterator `next()` is outstanding, then rearm on later demand. */
|
||||
/** Arm only while one iterator `next()` is outstanding; rearm on later demand or out-of-band activity. */
|
||||
export function idleWatchdog(
|
||||
upstream: AbortSignal | undefined,
|
||||
timeoutMs: number,
|
||||
@@ -71,7 +72,7 @@ export function idleWatchdog(
|
||||
export function timeoutOf(x: AbortSignal | { reason?: unknown }, code?: string): TimeoutReason | undefined
|
||||
```
|
||||
|
||||
`deadline` fuses an upstream signal with a one-shot timer through `AbortSignal.any`, adds a typed `TimeoutReason`, and exposes disposable timer cleanup. Non-positive timeouts are an internal no-timeout sentinel for backend-owned background work; external hints pass through `clampTimeout` and must be positive and finite. Without a timer or upstream signal, the function returns a never-aborting signal with the same disposal shape. `idleWatchdog` instead requires a positive finite interval, keeps one stable fused signal for the entire stream, and arms its timer only while one iterator `next()` is outstanding; resolution disarms it, later demand rearms it, concurrent demand fails, and disposal clears the active arm. Providers translate timeout reasons into seam-specific results. `timeoutOf(signal, code)` scopes classification so an outer nested deadline is treated as upstream cancellation rather than the inner capability's timeout.
|
||||
`deadline` fuses an upstream signal with a one-shot timer through `AbortSignal.any`, adds a typed `TimeoutReason`, and exposes disposable timer cleanup. Non-positive timeouts are an internal no-timeout sentinel for backend-owned background work; external hints pass through `clampTimeout` and must be positive and finite. Without a timer or upstream signal, the function returns a never-aborting signal with the same disposal shape. `idleWatchdog` instead requires a positive finite interval, keeps one stable fused signal for the entire stream, and arms its timer only while one iterator `next()` is outstanding; resolution disarms it, later demand rearms it, and `pulse()` rearms that same outstanding demand after out-of-band transport activity. A pulse outside outstanding demand or after disposal is a no-op; concurrent demand fails, and disposal clears the active arm. Providers translate timeout reasons into seam-specific results. `timeoutOf(signal, code)` scopes classification so an outer nested deadline is treated as upstream cancellation rather than the inner capability's timeout.
|
||||
|
||||
### The division of labor
|
||||
|
||||
@@ -79,7 +80,7 @@ export function timeoutOf(x: AbortSignal | { reason?: unknown }, code?: string):
|
||||
|---|---|
|
||||
| Validate request hint and clamp default/max | `dsh-timeout` (`clampTimeout`) — pure arithmetic plus the shared positive-finite request contract |
|
||||
| Arm one-shot timer, abort on deadline, carry reason, fuse with upstream cancel | `dsh-timeout` (`deadline`) |
|
||||
| Arm and rearm only around outstanding iterator demand | `dsh-timeout` (`idleWatchdog`) |
|
||||
| Arm and rearm only around outstanding iterator demand, including out-of-band activity | `dsh-timeout` (`idleWatchdog`) |
|
||||
| Clear the timer | `dsh-timeout` (`[Symbol.dispose]` on either primitive) |
|
||||
| Classify the first abort reason after abort | `dsh-timeout` (`timeoutOf`) |
|
||||
| **Actually terminate the work** | the capability's implementation |
|
||||
@@ -92,7 +93,7 @@ The signal only *notifies*; termination is always the listener's job, and the li
|
||||
|
||||
- **web_fetch** — the tool stays validate-and-forward; the provider's hand-rolled controller + `setTimeout` + manual listener + `finally` + `signal.reason` recovery is replaced by provider-owned `deadline`/`timeoutOf`. A pre-aborted upstream signal still throws `WEB_ABORTED` up front; otherwise `fetch` runs against the fused `d.signal`, and `translateAbortOrNetwork` classifies a thrown error by the signal (`timeoutOf` → `WEB_FETCH_TIMEOUT`, else aborted → `WEB_ABORTED`, else network → `WEB_PROVIDER_ERROR`). The public error-code contract is unchanged, and `TimeoutReason` never crosses the web seam as the public error.
|
||||
- **bash** — `resolve()` clamps the request into an explicit spec. Foreground `run()` creates the deadline and passes its signal to process execution, whose existing abort listener performs the process-group kill. The executor classifies the first abort as timeout or cancellation. Background starts remain timeout-free and forward only upstream cancellation.
|
||||
- **LLM adapters** — `dsh-llm-deepseek` and `dsh-llm-pi-ai` wrap actual transport iteration with `idleWatchdog`. The five-minute configured interval covers only outstanding provider demand, not time the downstream consumer spends between chunks. The stable signal reaches `fetch` or the SDK for the whole call, so timeout closes the underlying request and maps to `TIMEOUT`, while an earlier caller abort maps to `ABORTED`.
|
||||
- **LLM adapters** — `dsh-llm-deepseek` and `dsh-llm-pi-ai` wrap actual transport iteration with `idleWatchdog`. The five-minute configured interval covers only outstanding provider demand, not time the downstream consumer spends between chunks. The direct DeepSeek adapter also pulses that outstanding demand when its SSE parser observes a comment, without yielding the comment as a `StreamChunk` or writing it to the session log. The pi-ai SDK does not expose comment activity to its adapter, so that path can rearm only when the SDK yields. The stable signal reaches `fetch` or the SDK for the whole call, so timeout closes the underlying request and maps to `TIMEOUT`, while an earlier caller abort maps to `ABORTED`.
|
||||
|
||||
## Consequences
|
||||
|
||||
@@ -100,7 +101,7 @@ The signal only *notifies*; termination is always the listener's job, and the li
|
||||
- `SpawnSpec.timeoutMs` and `SpawnOutcome.timedOut`/`aborted` were removed rather than kept as always-zero/always-false vestiges: with `runBash` owning no timer and the executor owning classification, they were read nowhere. This is the one deviation from the literal proposal shape (which passed `timeoutMs: 0` into `runBash`); an always-0 field read by nothing is dead weight under the per-file coverage gate.
|
||||
- web_fetch shed its bespoke controller/timer/listener/reason-recovery; the classifier now keys off the deadline signal (`timeoutOf` + `aborted`) rather than the thrown error's shape, which is robust across both the request-phase reject-with-reason and the read-phase bare-`AbortError`.
|
||||
- `AbortSignal.any` and `using`/`Symbol.dispose` enter the repo for the first time here (Node ≥ 24 baseline, already met).
|
||||
- Model streams now share one rearmable timer contract without turning a sliding idle interval into a total-call deadline or charging consumer think time. The primitive still only notifies; adapter tests prove their transports observe its stable signal and terminate.
|
||||
- Model streams now share one rearmable timer contract without turning a sliding idle interval into a total-call deadline or charging consumer think time. Adapters that can observe out-of-band transport activity may pulse an outstanding demand; suppressed activity remains invisible to the watchdog. The primitive still only notifies; adapter tests prove their transports observe its stable signal and terminate.
|
||||
|
||||
Out of scope, named to mark the boundary: `web_search` can gain an optional model-facing `timeout_ms` once its tool-schema/snapshot coverage is planned; future ripgrep-backed fs discovery tools can consume the same provider-owned deadline shape once they exist; a `tools/execute` waterfall middleware could arm a default deadline for every tool call by driving `exec.signal` — that would be a plugin that *consumes* this library and still only notifies, the hard kill remaining each capability's job.
|
||||
|
||||
|
||||
@@ -57,10 +57,11 @@ export function deadline(
|
||||
export interface IdleWatchdog {
|
||||
readonly signal: AbortSignal
|
||||
next<T>(iterator: AsyncIterator<T>): Promise<IteratorResult<T>>
|
||||
pulse(): void
|
||||
[Symbol.dispose](): void
|
||||
}
|
||||
|
||||
/** Arm only while one iterator `next()` is outstanding, then rearm on later demand. */
|
||||
/** Arm only while one iterator `next()` is outstanding; rearm on later demand or out-of-band activity. */
|
||||
export function idleWatchdog(
|
||||
upstream: AbortSignal | undefined,
|
||||
timeoutMs: number,
|
||||
@@ -71,7 +72,7 @@ export function idleWatchdog(
|
||||
export function timeoutOf(x: AbortSignal | { reason?: unknown }, code?: string): TimeoutReason | undefined
|
||||
```
|
||||
|
||||
`deadline` 通过 `AbortSignal.any` 将上游信号与一次性定时器融合，附加一个类型化的 `TimeoutReason`，并暴露可 dispose（资源释放）的定时器清理。非正数超时是内部的「无超时」哨兵，用于后端拥有的后台任务；外部提示经过 `clampTimeout`，必须为正有限值。既无定时器也无上游信号时，函数返回一个永不中止的信号，具有相同的 disposal 形状。`idleWatchdog` 则要求正有限的间隔，在整个流期间保持一个稳定的融合信号，并且只在一个迭代器 `next()` 尚未结算时启动定时器；结算会解除定时器，后续 demand 会重新启动，并发 demand 会失败，dispose 会清除当前 arm。提供方将超时原因转译为 seam 特定的结果。`timeoutOf(signal, code)` 限定分类范围，使外层嵌套的 deadline 被视为上游取消而非内层能力自身的超时。
|
||||
`deadline` 通过 `AbortSignal.any` 将上游信号与一次性定时器融合，附加一个类型化的 `TimeoutReason`，并暴露可 dispose（资源释放）的定时器清理。非正数超时是内部的「无超时」哨兵，用于后端拥有的后台任务；外部提示经过 `clampTimeout`，必须为正有限值。既无定时器也无上游信号时，函数返回一个永不中止的信号，具有相同的 disposal 形状。`idleWatchdog` 则要求正有限的间隔，在整个流期间保持一个稳定的融合信号，并且只在一个迭代器 `next()` 尚未结算时启动定时器；结算会解除定时器，后续 demand 会重新启动，带外传输活动发生后，`pulse()` 则会为同一个尚未结算的 demand 重新启动定时器。若没有尚未结算的 demand，或已经 dispose，pulse 不执行任何操作；并发 demand 会失败，dispose 会清除当前 arm。提供方将超时原因转译为 seam 特定的结果。`timeoutOf(signal, code)` 限定分类范围，使外层嵌套的 deadline 被视为上游取消而非内层能力自身的超时。
|
||||
|
||||
### 职责划分
|
||||
|
||||
@@ -79,7 +80,7 @@ export function timeoutOf(x: AbortSignal | { reason?: unknown }, code?: string):
|
||||
|---|---|
|
||||
| 校验请求提示并钳位默认值/最大值 | `dsh-timeout`（`clampTimeout`）：纯算术加共享的正有限请求约定 |
|
||||
| 启动一次性定时器、到期中止、携带 reason、与上游取消融合 | `dsh-timeout`（`deadline`） |
|
||||
| 仅围绕未结算的迭代器 demand 启动和重启 | `dsh-timeout`（`idleWatchdog`） |
|
||||
| 仅围绕未结算的迭代器 demand 启动和重启，带外活动也会触发重启 | `dsh-timeout`（`idleWatchdog`） |
|
||||
| 清除定时器 | `dsh-timeout`（任一原语的 `[Symbol.dispose]`） |
|
||||
| 中止后对首个 abort reason 进行分类 | `dsh-timeout`（`timeoutOf`） |
|
||||
| **实际终止工作** | 各能力的实现 |
|
||||
@@ -92,7 +93,7 @@ export function timeoutOf(x: AbortSignal | { reason?: unknown }, code?: string):
|
||||
|
||||
- **web_fetch**：工具层保持校验并转发；提供方手写的 controller + `setTimeout` + 手动监听器 + `finally` + `signal.reason` 恢复被替换为提供方自有的 `deadline`/`timeoutOf`。已预先中止的上游信号仍然立即抛出 `WEB_ABORTED`；否则 `fetch` 使用融合后的 `d.signal` 运行，`translateAbortOrNetwork` 根据信号分类抛出的错误（`timeoutOf` → `WEB_FETCH_TIMEOUT`，否则已中止 → `WEB_ABORTED`，否则网络错误 → `WEB_PROVIDER_ERROR`）。公开的错误码约定不变，`TimeoutReason` 永远不会作为公开错误跨越 web seam。
|
||||
- **bash**：`resolve()` 将请求钳位为显式规格。前台 `run()` 创建 deadline 并将其信号传给进程执行，后者既有的 abort 监听器执行进程组 kill。执行器将首个 abort 分类为超时或取消。后台启动保持无超时，仅转发上游取消。
|
||||
- **LLM（大语言模型）适配器**：`dsh-llm-deepseek` 和 `dsh-llm-pi-ai` 用 `idleWatchdog` 包装实际的传输迭代。配置的五分钟间隔只覆盖尚未结算的提供方 demand，不包括下游消费方在分片之间花费的时间。稳定信号在整个调用期间传给 `fetch` 或 SDK，因此超时会关闭底层请求并映射为 `TIMEOUT`，而更早的调用方中止映射为 `ABORTED`。
|
||||
- **LLM（大语言模型）适配器**：`dsh-llm-deepseek` 和 `dsh-llm-pi-ai` 用 `idleWatchdog` 包装实际的传输迭代。配置的五分钟间隔只覆盖尚未结算的提供方 demand，不包括下游消费方在分片之间花费的时间。DeepSeek 直连适配器还会在其 SSE（Server-Sent Events）解析器观察到注释时，对该项尚未结算的 demand 调用 `pulse()`；该注释既不会作为 `StreamChunk` 产出，也不会写入会话日志。pi-ai SDK 不会向其适配器暴露注释活动，因此该路径只能在 SDK 产出值时重新启动定时器。稳定信号在整个调用期间传给 `fetch` 或 SDK，因此超时会关闭底层请求并映射为 `TIMEOUT`，而更早的调用方中止映射为 `ABORTED`。
|
||||
|
||||
## 后果
|
||||
|
||||
@@ -100,7 +101,7 @@ export function timeoutOf(x: AbortSignal | { reason?: unknown }, code?: string):
|
||||
- `SpawnSpec.timeoutMs` 和 `SpawnOutcome.timedOut`/`aborted` 被移除，而非作为始终为零/始终为 false 的残余保留：由于 `runBash` 不再拥有定时器且执行器负责分类，这些字段无处被读取。这是与字面提案形状（向 `runBash` 传入 `timeoutMs: 0`）的唯一偏差；一个始终为 0 且无处读取的字段在逐文件覆盖率门禁下属于死代码。
|
||||
- web_fetch 去除了其定制的 controller/timer/listener/reason-recovery；分类器现在基于 deadline 信号（`timeoutOf` + `aborted`）而非抛出错误的形状来判断，这在请求阶段的 reject-with-reason 和读取阶段的裸 `AbortError` 两种情况下都是健壮的。
|
||||
- `AbortSignal.any` 和 `using`/`Symbol.dispose` 在此首次进入本仓库（Node ≥ 24 基线，已满足）。
|
||||
- 模型流现在共享一个可重启的定时器约定，不会把滑动的空闲间隔变成总调用截止时间，也不会计入消费方思考时间。该原语仍然只做通知；适配器测试证明其传输观察到稳定信号并终止。
|
||||
- 模型流现在共享一个可重启的定时器约定，不会把滑动的空闲间隔变成总调用截止时间，也不会计入消费方思考时间。能够观察到带外传输活动的适配器可以对尚未结算的 demand 调用 `pulse()`；被屏蔽的活动对 watchdog 仍不可见。该原语仍然只做通知；适配器测试证明其传输观察到稳定信号并终止。
|
||||
|
||||
以下内容不在本次范围内，列出以标明边界：`web_search` 可以在其工具 schema 和快照覆盖规划完成后获得可选的面向模型的 `timeout_ms`；未来基于 ripgrep 的文件系统发现工具可以在实现后消费同样的提供方自有 deadline 形状；`tools/execute` waterfall（瀑布式事件）中间件可以通过驱动 `exec.signal` 为每次工具调用设置默认 deadline——那将是一个*消费*本库的插件，仍然只做通知，硬终止仍是各能力自己的事。
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
config:
|
||||
baseURL: !!js process.env.DSH_SNAPSHOT_BASE_URL
|
||||
thinking: disabled
|
||||
streamIdleTimeoutMs: 150
|
||||
- id: agent-spine
|
||||
config:
|
||||
agents:
|
||||
|
||||
@@ -83,12 +83,21 @@ async function deepseekDefaultsServer(): Promise<DeepSeekDefaultsServer> {
|
||||
request.on('end', () => {
|
||||
requests.push(JSON.parse(body) as JsonObject)
|
||||
response.writeHead(200, { 'content-type': 'text/event-stream' })
|
||||
response.end([
|
||||
'data: {"choices":[{"delta":{"content":"DEFAULTS_OK"}}]}',
|
||||
'data: {"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}',
|
||||
'data: [DONE]',
|
||||
'',
|
||||
].join('\n\n'))
|
||||
let keepAlives = 3
|
||||
const write = (): void => {
|
||||
if (keepAlives-- > 0) {
|
||||
response.write(': keep-alive\n\n')
|
||||
setTimeout(write, 60)
|
||||
return
|
||||
}
|
||||
response.end([
|
||||
'data: {"choices":[{"delta":{"content":"DEFAULTS_OK"}}]}',
|
||||
'data: {"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}',
|
||||
'data: [DONE]',
|
||||
'',
|
||||
].join('\n\n'))
|
||||
}
|
||||
setTimeout(write, 60)
|
||||
})
|
||||
})
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
@@ -504,7 +513,7 @@ describe('headless stream-json snapshots', () => {
|
||||
`)
|
||||
}, LOADER_SMOKE_TEST_TIMEOUT_MS)
|
||||
|
||||
it('logs and sends the DeepSeek adapter maxTokens default through the one-shot app', async () => {
|
||||
it('keeps provider comments alive and sends DeepSeek defaults through the one-shot app', async () => {
|
||||
const server = await deepseekDefaultsServer()
|
||||
try {
|
||||
const result = await runLoaderSmoke({
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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）
|
||||
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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])
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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 请求。
|
||||
|
||||
## 哪些操作不设置超时
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user