From dbb150ff0e589ec5b34a261c201a853affe71a29 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Fri, 2 Oct 2026 13:02:47 +0800 Subject: [PATCH] fix(timeout): end an idle demand the transport abandons MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 8 +++-- packages/llm/llm-pi-ai/src/adapter.ts | 11 +++--- .../tests/transport-recovery.spec.ts | 34 +++++++++++++++++++ packages/util/timeout/src/index.ts | 22 +++++++++--- packages/util/timeout/tests/timeout.spec.ts | 19 +++++++++-- 5 files changed, 78 insertions(+), 16 deletions(-) diff --git a/packages/llm/llm-deepseek/src/adapter.ts b/packages/llm/llm-deepseek/src/adapter.ts index 20e899788e..47a42c27bc 100644 --- a/packages/llm/llm-deepseek/src/adapter.ts +++ b/packages/llm/llm-deepseek/src/adapter.ts @@ -66,9 +66,11 @@ export class DeepSeekAdapter 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(() => {}) } } diff --git a/packages/llm/llm-pi-ai/src/adapter.ts b/packages/llm/llm-pi-ai/src/adapter.ts index de9f3e0e42..747e1b9dbb 100644 --- a/packages/llm/llm-pi-ai/src/adapter.ts +++ b/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) { diff --git a/packages/llm/llm-retry/tests/transport-recovery.spec.ts b/packages/llm/llm-retry/tests/transport-recovery.spec.ts index 062f6ad300..2bf04974df 100644 --- a/packages/llm/llm-retry/tests/transport-recovery.spec.ts +++ b/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({ 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', diff --git a/packages/util/timeout/src/index.ts b/packages/util/timeout/src/index.ts index 3fc4d2387b..4082e8e6b5 100644 --- a/packages/util/timeout/src/index.ts +++ b/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(iterator: AsyncIterator): Promise> /** 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((_resolve, reject) => { + timeout.signal.addEventListener('abort', () => { reject(reason) }, { once: true }) + }) let timer: ReturnType | 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 diff --git a/packages/util/timeout/tests/timeout.spec.ts b/packages/util/timeout/tests/timeout.spec.ts index af0389d3d7..b67333ef62 100644 --- a/packages/util/timeout/tests/timeout.spec.ts +++ b/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>(() => {}) }) + 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/)