fix(timeout): end an idle demand the transport abandons

An aborted read can stay pending forever: undici drops the error when the
abort lands on an in-flight response-body read (observed on the coverage
lane's Node v24.13.0), and the stalled body of
`llm-retry/tests/transport-recovery.spec.ts` then never settles. The idle
watchdog delegated termination to the iterator observing its signal, so the
request outlived its own deadline and the agent stayed `running` until
Vitest's 90s budget; awaiting the abandoned request generator's `return()`
queued behind that demand and would have re-hung the settled outcome.

`idleWatchdog.next` now races its demand against its own deadline and
rejects with the watchdog's TimeoutReason, so the deadline — not the
transport — decides when an outstanding demand ends. Both adapters that
consume the watchdog stop awaiting the abandoned generator's drain; the
abort they already issue owns socket and SDK teardown.

Evidence: instrumented stress harness at ~1s idle on Node v24.13.0 hung 4/160
iterations before this change (each with the abort delivered at ~1002ms and
the underlying read never settling) and 0/160 after; Node v24.19.0 never
hung. The new case in transport-recovery.spec.ts pins the behavior
deterministically: it fails (times out) with the watchdog change alone and
with no product change at all.

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

@@ -66,9 +66,11 @@ export class DeepSeekAdapter<C extends Connection = Connection> extends LlmAdapt
       throw new LlmError('DeepSeek Messages transport failed', 'TRANSPORT', { cause: error })
     } finally {
       consumer.abort()
-      try { await iterator.return(undefined) } catch (_abortedRequestCleanup) {
-        // The request already settled; aborting its reader cannot replace that outcome.
-      }
+      // The abort above owns the socket teardown. A demand the deadline abandoned
+      // can stay pending forever, and this generator queues its own return behind
+      // that demand, so awaiting the drain would re-hang the settled outcome.
+      /* v8 ignore next -- the detached drain only rejects while unwinding an abandoned request. */
+      void iterator.return(undefined).catch(() => {})
     }
   }
 

packages/llm/llm-pi-ai/src/adapter.ts

@@ -392,8 +392,6 @@ export class PiAiAdapter extends LlmAdapter {
       try {
         while (true) {
           const result = await watchdog.next(iterator)
-          const timeout = timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT')
-          if (timeout !== undefined) throw timeout
           if (result.done) {
             exhausted = true
             return
@@ -403,11 +401,10 @@ export class PiAiAdapter extends LlmAdapter {
       } finally {
         if (!exhausted) {
           consumer.abort('pi-ai stream consumer stopped')
-          try {
-            await iterator.return(undefined)
-          } catch (_abortedSdkTeardown) {
-            // The stable signal already owns SDK termination; return-time abort cannot add an outcome.
-          }
+          // A demand the deadline abandoned can stay pending forever, and this
+          // generator queues its return behind it; the abort above owns SDK termination.
+          /* v8 ignore next -- the detached drain only rejects while unwinding an abandoned stream. */
+          void iterator.return(undefined).catch(() => {})
         }
       }
     } catch (error: unknown) {

packages/llm/llm-retry/tests/transport-recovery.spec.ts

@@ -32,6 +32,7 @@ afterEach(async (test) => {
     ...servers.splice(0).map(async server => (await server).close()),
   ])
   vi.unstubAllEnvs()
+  vi.unstubAllGlobals()
   const failure = results.find(result => result.status === 'rejected')
   if (failure?.status === 'rejected') throw failure.reason
 })
@@ -259,6 +260,39 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
     expect(finalAssistantText(agent)).toBe('recovered after timeout')
   })
 
+  it('turns a body that never observes the abort into TIMEOUT and succeeds on the next request', async () => {
+    const server = await start(['success'], {
+      apiKey: 'mock-key',
+      successText: 'recovered from an ignored abort',
+    })
+    const realFetch = globalThis.fetch
+    let attempts = 0
+    // A transport that ignores the abort on an in-flight read, as undici does when
+    // the abort lands on a pending body read: the abandoned read never settles, so
+    // only the idle deadline can end this attempt.
+    vi.stubGlobal('fetch', (input: RequestInfo | URL, init?: RequestInit) => {
+      attempts += 1
+      if (attempts > 1) return realFetch(input, init)
+      return Promise.resolve(new Response(
+        new ReadableStream<Uint8Array>({ start() {} }),
+        { status: 200, headers: { 'content-type': 'text/event-stream' } },
+      ))
+    })
+    context = await harness(server.baseURL, { streamIdleTimeoutMs: 50 })
+    const agent = await context.agentLoop.create(SessionId('wire-ignored-abort'), {
+      provider: 'deepseek-official',
+      model: 'mock-model',
+    })
+
+    await sendAndWait(context, agent)
+
+    expect(attempts).toBe(2)
+    expect(server.requests).toHaveLength(1)
+    expect(agent.session.snapshotEvents().filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
+      .toEqual(['TIMEOUT'])
+    expect(finalAssistantText(agent)).toBe('recovered from an ignored abort')
+  })
+
   it('stops after the configured transport retry budget is exhausted', async () => {
     const server = await start(['connection_reset', 'connection_reset', 'connection_reset'], {
       apiKey: 'mock-key',

packages/util/timeout/src/index.ts

@@ -67,9 +67,14 @@ 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.
+   * Await one iterator demand while the idle timer is armed. The deadline
+   * settles the demand even when the awaited work ignores its abort signal.
    * @param iterator - iterator whose next value represents provider progress.
    * @returns the iterator's next result.
+   * @throws The {@link TimeoutReason} of this watchdog when the idle interval
+   *   elapses first. The abandoned demand stays the caller's to release through
+   *   {@link IdleWatchdog.signal}; an async generator queues its own return
+   *   behind that demand, so callers must not await it.
    */
   next<T>(iterator: AsyncIterator<T>): Promise<IteratorResult<T>>
   /** Rearm an outstanding demand after transport activity that yields no iterator value; otherwise a no-op. */
@@ -116,7 +121,8 @@ 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.
+ * call and only notifies; the iterator must observe it to release the resources
+ * of a demand the deadline abandoned.
  *
  * @param upstream - caller cancellation fused into the stable signal.
  * @param timeoutMs - positive finite idle interval in milliseconds.
@@ -133,6 +139,14 @@ export function idleWatchdog(
   const signal = upstream === undefined
     ? timeout.signal
     : AbortSignal.any([upstream, timeout.signal])
+  // The deadline owns the outstanding demand. A transport can leave a read it
+  // aborted pending forever (undici drops the error when the abort lands on an
+  // in-flight body read), so waiting for the iterator to observe the signal
+  // would leave the caller waiting past its own deadline.
+  const reason = new TimeoutReason(code, timeoutMs)
+  const expired = new Promise<never>((_resolve, reject) => {
+    timeout.signal.addEventListener('abort', () => { reject(reason) }, { once: true })
+  })
   let timer: ReturnType<typeof setTimeout> | undefined
   let outstanding = false
   let disposed = false
@@ -140,7 +154,7 @@ export function idleWatchdog(
   const arm = (): void => {
     if (timer !== undefined) clearTimeout(timer)
     timer = setTimeout(() => {
-      timeout.abort(new TimeoutReason(code, timeoutMs))
+      timeout.abort(reason)
     }, timeoutMs)
   }
 
@@ -152,7 +166,7 @@ export function idleWatchdog(
       outstanding = true
       arm()
       try {
-        return await iterator.next()
+        return await Promise.race([iterator.next(), expired])
       } finally {
         clearTimeout(timer)
         timer = undefined

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

@@ -223,10 +223,12 @@ describe('idleWatchdog', () => {
     expect(watchdog.signal).toBe(stableSignal)
 
     const secondNext = watchdog.next(iterator)
+    // Observe before advancing: the deadline rejects an iterator that never settles.
+    const secondOutcome = secondNext.catch((error: unknown) => error)
     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)
+    expect(await secondOutcome).toBe(stableSignal.reason)
   })
 
   it('rearms outstanding demand on an out-of-band activity pulse', async () => {
@@ -238,6 +240,7 @@ describe('idleWatchdog', () => {
     expect(watchdog.signal.aborted).toBe(false)
 
     const next = watchdog.next({ next: () => pending.promise })
+    const outcome = next.catch((error: unknown) => error)
     await vi.advanceTimersByTimeAsync(99)
     watchdog.pulse()
     await vi.advanceTimersByTimeAsync(99)
@@ -245,7 +248,7 @@ describe('idleWatchdog', () => {
     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)
+    expect(await outcome).toBe(watchdog.signal.reason)
 
     watchdog[Symbol.dispose]()
     watchdog.pulse()
@@ -276,6 +279,18 @@ describe('idleWatchdog', () => {
     watchdog[Symbol.dispose]()
   })
 
+  it('rejects an outstanding demand the iterator never settles', async () => {
+    vi.useFakeTimers()
+    using watchdog = idleWatchdog(undefined, 100, 'LLM_STREAM_IDLE_TIMEOUT')
+    // A transport can leave an aborted read pending forever; the deadline, not the
+    // iterator, decides when this demand ends.
+    const next = watchdog.next({ next: () => new Promise<IteratorResult<number>>(() => {}) })
+    const outcome = next.catch((error: unknown) => error)
+    await vi.advanceTimersByTimeAsync(100)
+    expect(await outcome).toBe(watchdog.signal.reason)
+    expect(timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT')).toMatchObject({ timeoutMs: 100 })
+  })
+
   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/)