Merge master into worktree-windows-runtime

packages/util/timeout/README.md

@@ -9,13 +9,15 @@ It is a **library, not a service or plugin**: no `ctx`, registers nothing, holds
 ## Surface
 
 ```ts
-import { clampTimeout, deadline, timeoutOf, TimeoutReason } from '@deepseek-ai/dsh-timeout'
+import { clampTimeout, deadline, idleWatchdog, MAX_TIMER_DELAY_MS, timeoutOf, TimeoutReason } from '@deepseek-ai/dsh-timeout'
 ```
 
 | Export | Role |
 |---|---|
 | `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. |
+| `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. |
 
@@ -44,6 +46,8 @@ 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 the `upstream` you were handed is *itself* a deadline signal (a future `tools/execute` middleware arming a per-call deadline), `AbortSignal.any` preserves the outer `TimeoutReason` if the outer timer fires first. Scoping to your `code` makes a foreign timeout read as an ordinary upstream cancel — the correct classification from your capability's view — instead of your own timeout firing when your local timer never 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.
+
 ## What does NOT get a timeout
 
 Local file `read`/`write`/`edit` take no `timeoutMs`: a syscall is best-effort-abortable at most, a timeout could not force `fsync`/`rename` to stop, and adding one would be an implicit default that violates explicit-over-implicit. See [`fs/`](../../fs/README.md).
@@ -61,3 +65,4 @@ No direct invalidation; the named consumer owns any request-prefix changes.
 - **Notification only** — a deadline cannot stop work that ignores its signal; every capability still needs its own socket/process/task termination path.
 - **`timeoutMs <= 0` is internal vocabulary** — it disables the local timer only after an owning backend has resolved policy, never as a public model/plugin knob.
 - **The first abort reason wins classification** — when an upstream cancellation beats the local timer, this layer cannot later report that its own timeout would also have elapsed.
+- **An idle watchdog is not a total deadline** — it rearms per outstanding iterator demand and deliberately excludes consumer think time.

packages/util/timeout/src/index.ts

@@ -21,6 +21,15 @@ export class TimeoutReason extends Error {
   }
 }
 
+/** Largest delay Node schedules without clamping it to one millisecond. */
+export const MAX_TIMER_DELAY_MS = 2_147_483_647
+
+function assertTimerDelay(timeoutMs: number, name: string): void {
+  if (!Number.isFinite(timeoutMs) || timeoutMs <= 0 || timeoutMs > MAX_TIMER_DELAY_MS) {
+    throw new Error(`${name} must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`)
+  }
+}
+
 /**
  * Validate a caller's optional timeout hint, use the backend default, then cap
  * it. Supplied values must be positive and finite; zero is not a public
@@ -53,6 +62,20 @@ export interface Deadline {
   [Symbol.dispose](): void
 }
 
+/** Rearmable timeout around one outstanding async-iterator demand. */
+export interface IdleWatchdog {
+  /** Stable signal aborted by upstream cancellation or this watchdog's timeout. */
+  readonly signal: AbortSignal
+  /**
+   * Await one iterator demand while the idle timer is armed.
+   * @param iterator - iterator whose next value represents provider progress.
+   * @returns the iterator's next result.
+   */
+  next<T>(iterator: AsyncIterator<T>): Promise<IteratorResult<T>>
+  /** Clear an armed timer; safe to call once at the owning stream's exit. */
+  [Symbol.dispose](): void
+}
+
 /**
  * Fuse upstream cancellation with an identifiable timeout. `timeoutMs <= 0` is
  * the internal no-timer sentinel; the returned disposer clears an armed timer.
@@ -74,6 +97,8 @@ export function deadline(
     return { signal: upstream ?? new AbortController().signal, [Symbol.dispose]() {} }
   }
 
+  assertTimerDelay(timeoutMs, 'deadline timeoutMs')
+
   const timer = new AbortController()
   const id = setTimeout(() => { timer.abort(new TimeoutReason(code, timeoutMs)) }, timeoutMs)
   return {
@@ -85,6 +110,57 @@ export function deadline(
   }
 }
 
+/**
+ * Create a rearmable idle watchdog for an async iterator. The timer exists only
+ * while {@link IdleWatchdog.next} is outstanding, so consumer think time does
+ * not count as provider idle time. The returned signal is stable for the whole
+ * call and only notifies; the iterator must observe it to terminate its work.
+ *
+ * @param upstream - caller cancellation fused into the stable signal.
+ * @param timeoutMs - positive finite idle interval in milliseconds.
+ * @param code - capability-owned code carried by the timeout reason.
+ * @returns a stable signal, guarded next operation, and timer disposer.
+ */
+export function idleWatchdog(
+  upstream: AbortSignal | undefined,
+  timeoutMs: number,
+  code: string,
+): IdleWatchdog {
+  assertTimerDelay(timeoutMs, 'idleWatchdog timeoutMs')
+  const timeout = new AbortController()
+  const signal = upstream === undefined
+    ? timeout.signal
+    : AbortSignal.any([upstream, timeout.signal])
+  let timer: ReturnType<typeof setTimeout> | undefined
+  let outstanding = false
+  let disposed = false
+
+  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)
+      try {
+        return await iterator.next()
+      } finally {
+        clearTimeout(timer)
+        timer = undefined
+        outstanding = false
+      }
+    },
+    [Symbol.dispose](): void {
+      if (disposed) return
+      disposed = true
+      if (timer !== undefined) clearTimeout(timer)
+      timer = undefined
+    },
+  }
+}
+
 /**
  * Recover a timeout reason from a reason-bearing object. Supplying `code`
  * distinguishes this deadline from a nested upstream deadline; a foreign code

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

@@ -1,5 +1,12 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { clampTimeout, deadline, timeoutOf, TimeoutReason } from '@deepseek-ai/dsh-timeout'
+import {
+  clampTimeout,
+  deadline,
+  idleWatchdog,
+  MAX_TIMER_DELAY_MS,
+  timeoutOf,
+  TimeoutReason,
+} from '@deepseek-ai/dsh-timeout'
 
 describe('TimeoutReason', () => {
   it('is an Error carrying the code and elapsed ms', () => {
@@ -67,6 +74,13 @@ describe('deadline — timeout arm', () => {
     expect(d.signal.aborted).toBe(false)
     expect(timeoutOf(d.signal)).toBeUndefined()
   })
+
+  it('rejects delays that Node would clamp to one millisecond', () => {
+    expect(() => deadline(undefined, MAX_TIMER_DELAY_MS + 1, 'BASH_TIMEOUT'))
+      .toThrow(`no greater than ${MAX_TIMER_DELAY_MS}`)
+    expect(() => deadline(undefined, Number.POSITIVE_INFINITY, 'BASH_TIMEOUT'))
+      .toThrow(`no greater than ${MAX_TIMER_DELAY_MS}`)
+  })
 })
 
 describe('deadline — fuse with upstream', () => {
@@ -182,3 +196,74 @@ describe('deadline — nested deadlines', () => {
     expect(timeoutOf(inner.signal)?.code).toBe('OUTER_TIMEOUT') // but IS a timeout, unscoped
   })
 })
+
+describe('idleWatchdog', () => {
+  afterEach(() => { vi.useRealTimers() })
+
+  it('arms only while next is outstanding and rearms the same signal for later demand', async () => {
+    vi.useFakeTimers()
+    const first = Promise.withResolvers<IteratorResult<number>>()
+    const second = Promise.withResolvers<IteratorResult<number>>()
+    const iterator: AsyncIterator<number> = {
+      next: vi.fn()
+        .mockImplementationOnce(() => first.promise)
+        .mockImplementationOnce(() => second.promise),
+    }
+    using watchdog = idleWatchdog(undefined, 100, 'LLM_STREAM_IDLE_TIMEOUT')
+    const stableSignal = watchdog.signal
+
+    const firstNext = watchdog.next(iterator)
+    await vi.advanceTimersByTimeAsync(99)
+    expect(stableSignal.aborted).toBe(false)
+    first.resolve({ done: false, value: 1 })
+    await expect(firstNext).resolves.toEqual({ done: false, value: 1 })
+
+    await vi.advanceTimersByTimeAsync(10_000)
+    expect(stableSignal.aborted).toBe(false)
+    expect(watchdog.signal).toBe(stableSignal)
+
+    const secondNext = watchdog.next(iterator)
+    await vi.advanceTimersByTimeAsync(100)
+    expect(timeoutOf(stableSignal, 'LLM_STREAM_IDLE_TIMEOUT')).toMatchObject({ timeoutMs: 100 })
+    second.reject(stableSignal.reason)
+    await expect(secondNext).rejects.toBe(stableSignal.reason)
+  })
+
+  it('keeps an earlier upstream abort distinct from its own timeout', async () => {
+    vi.useFakeTimers()
+    const upstream = new AbortController()
+    using watchdog = idleWatchdog(upstream.signal, 100, 'LLM_STREAM_IDLE_TIMEOUT')
+    upstream.abort('caller cancelled')
+    expect(watchdog.signal.aborted).toBe(true)
+    expect(timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT')).toBeUndefined()
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(watchdog.signal.reason).toBe('caller cancelled')
+  })
+
+  it('clears an outstanding arm on disposal', async () => {
+    vi.useFakeTimers()
+    const pending = Promise.withResolvers<IteratorResult<number>>()
+    const watchdog = idleWatchdog(undefined, 100, 'LLM_STREAM_IDLE_TIMEOUT')
+    void watchdog.next({ next: () => pending.promise })
+    watchdog[Symbol.dispose]()
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(watchdog.signal.aborted).toBe(false)
+    pending.resolve({ done: true, value: undefined })
+    await expect(watchdog.next({ next: () => Promise.resolve({ done: true, value: undefined }) }))
+      .rejects.toThrow(/disposed/)
+    watchdog[Symbol.dispose]()
+  })
+
+  it('rejects invalid bounds and concurrent iterator demand', async () => {
+    expect(() => idleWatchdog(undefined, 0, 'IDLE')).toThrow(/positive finite/)
+    expect(() => idleWatchdog(undefined, Number.NaN, 'IDLE')).toThrow(/positive finite/)
+    expect(() => idleWatchdog(undefined, MAX_TIMER_DELAY_MS + 1, 'IDLE'))
+      .toThrow(`no greater than ${MAX_TIMER_DELAY_MS}`)
+    const pending = Promise.withResolvers<IteratorResult<number>>()
+    using watchdog = idleWatchdog(undefined, 100, 'IDLE')
+    const iterator = { next: () => pending.promise }
+    void watchdog.next(iterator)
+    await expect(watchdog.next(iterator)).rejects.toThrow(/already outstanding/)
+    pending.resolve({ done: true, value: undefined })
+  })
+})