Skip to content

Commit 84b72e6

Browse files
liuyobviyus
liuy
authored and
obviyus
committed
feat: add LLM idle timeout for streaming responses
Problem: When LLM stops responding, the agent hangs for ~5 minutes with no feedback. Users had to use /stop to recover. Solution: Add idle timeout detection for LLM streaming responses.
1 parent 47839d3 commit 84b72e6

6 files changed

Lines changed: 404 additions & 1 deletion

File tree

‎src/agents/pi-embedded-runner/run/attempt.ts‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -177,6 +177,7 @@ import {
177177
} from "./compaction-timeout.js";
178178
import { pruneProcessedHistoryImages } from "./history-image-prune.js";
179179
import { detectAndLoadPromptImages } from "./images.js";
180+
import { resolveLlmIdleTimeoutMs, streamWithIdleTimeout } from "./llm-idle-timeout.js";
180181
import type { EmbeddedRunAttemptParams, EmbeddedRunAttemptResult } from "./types.js";
181182

182183
export {
@@ -1056,14 +1057,25 @@ export async function runEmbeddedAttempt(
10561057
activeSession.agent.streamFn,
10571058
);
10581059
}
1059-
10601060
// Anthropic-compatible providers can add new stop reasons before pi-ai maps them.
10611061
// Recover the known "sensitive" stop reason here so a model refusal does not
10621062
// bubble out as an uncaught runner error and stall channel polling.
10631063
activeSession.agent.streamFn = wrapStreamFnHandleSensitiveStopReason(
10641064
activeSession.agent.streamFn,
10651065
);
10661066

1067+
let idleTimeoutTrigger: ((error: Error) => void) | undefined;
1068+
1069+
// Wrap stream with idle timeout detection
1070+
const idleTimeoutMs = resolveLlmIdleTimeoutMs(params.config);
1071+
if (idleTimeoutMs > 0) {
1072+
activeSession.agent.streamFn = streamWithIdleTimeout(
1073+
activeSession.agent.streamFn,
1074+
idleTimeoutMs,
1075+
(error) => idleTimeoutTrigger?.(error),
1076+
);
1077+
}
1078+
10671079
try {
10681080
const prior = await sanitizeSessionHistory({
10691081
messages: activeSession.messages,
@@ -1156,6 +1168,13 @@ export async function runEmbeddedAttempt(
11561168
};
11571169
const makeAbortError = (signal: AbortSignal): Error => {
11581170
const reason = getAbortReason(signal);
1171+
// If the reason is already an Error, preserve it to keep the original message
1172+
// (e.g., "LLM idle timeout (60s): no response from model" instead of "aborted")
1173+
if (reason instanceof Error) {
1174+
const err = new Error(reason.message, { cause: reason });
1175+
err.name = "AbortError";
1176+
return err;
1177+
}
11591178
const err = reason ? new Error("aborted", { cause: reason }) : new Error("aborted");
11601179
err.name = "AbortError";
11611180
return err;
@@ -1187,6 +1206,9 @@ export async function runEmbeddedAttempt(
11871206
abortCompaction();
11881207
void activeSession.abort();
11891208
};
1209+
idleTimeoutTrigger = (error) => {
1210+
abortRun(true, error);
1211+
};
11901212
const abortable = <T>(promise: Promise<T>): Promise<T> => {
11911213
const signal = runAbortController.signal;
11921214
if (signal.aborted) {

‎src/agents/pi-embedded-runner/run/llm-idle-timeout.test.ts‎

Lines changed: 219 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,219 @@
1+
import { describe, expect, it, vi } from "vitest";
2+
import type { OpenClawConfig } from "../../../config/config.js";
3+
import {
4+
DEFAULT_LLM_IDLE_TIMEOUT_MS,
5+
resolveLlmIdleTimeoutMs,
6+
streamWithIdleTimeout,
7+
} from "./llm-idle-timeout.js";
8+
9+
describe("resolveLlmIdleTimeoutMs", () => {
10+
it("returns default when config is undefined", () => {
11+
expect(resolveLlmIdleTimeoutMs(undefined)).toBe(DEFAULT_LLM_IDLE_TIMEOUT_MS);
12+
});
13+
14+
it("returns default when llm config is missing", () => {
15+
const cfg = { agents: {} } as OpenClawConfig;
16+
expect(resolveLlmIdleTimeoutMs(cfg)).toBe(DEFAULT_LLM_IDLE_TIMEOUT_MS);
17+
});
18+
19+
it("returns default when idleTimeoutSeconds is not set", () => {
20+
const cfg = { agents: { defaults: { llm: {} } } } as OpenClawConfig;
21+
expect(resolveLlmIdleTimeoutMs(cfg)).toBe(DEFAULT_LLM_IDLE_TIMEOUT_MS);
22+
});
23+
24+
it("returns 0 when idleTimeoutSeconds is 0 (disabled)", () => {
25+
const cfg = { agents: { defaults: { llm: { idleTimeoutSeconds: 0 } } } } as OpenClawConfig;
26+
expect(resolveLlmIdleTimeoutMs(cfg)).toBe(0);
27+
});
28+
29+
it("returns configured value in milliseconds", () => {
30+
const cfg = { agents: { defaults: { llm: { idleTimeoutSeconds: 30 } } } } as OpenClawConfig;
31+
expect(resolveLlmIdleTimeoutMs(cfg)).toBe(30_000);
32+
});
33+
34+
it("caps at max safe timeout", () => {
35+
const cfg = {
36+
agents: { defaults: { llm: { idleTimeoutSeconds: 10_000_000 } } },
37+
} as OpenClawConfig;
38+
expect(resolveLlmIdleTimeoutMs(cfg)).toBe(2_147_000_000);
39+
});
40+
41+
it("ignores negative values", () => {
42+
const cfg = { agents: { defaults: { llm: { idleTimeoutSeconds: -10 } } } } as OpenClawConfig;
43+
expect(resolveLlmIdleTimeoutMs(cfg)).toBe(DEFAULT_LLM_IDLE_TIMEOUT_MS);
44+
});
45+
46+
it("ignores non-finite values", () => {
47+
const cfg = {
48+
agents: { defaults: { llm: { idleTimeoutSeconds: Infinity } } },
49+
} as OpenClawConfig;
50+
expect(resolveLlmIdleTimeoutMs(cfg)).toBe(DEFAULT_LLM_IDLE_TIMEOUT_MS);
51+
});
52+
});
53+
54+
describe("streamWithIdleTimeout", () => {
55+
// Helper to create a mock async iterable
56+
function createMockAsyncIterable<T>(chunks: T[]): AsyncIterable<T> {
57+
return {
58+
[Symbol.asyncIterator]() {
59+
let index = 0;
60+
return {
61+
async next() {
62+
if (index < chunks.length) {
63+
return { done: false, value: chunks[index++] };
64+
}
65+
return { done: true, value: undefined };
66+
},
67+
async return() {
68+
return { done: true, value: undefined };
69+
},
70+
};
71+
},
72+
};
73+
}
74+
75+
it("wraps stream function", () => {
76+
const mockStream = createMockAsyncIterable([]);
77+
const baseFn = vi.fn().mockReturnValue(mockStream);
78+
const wrapped = streamWithIdleTimeout(baseFn, 1000);
79+
expect(typeof wrapped).toBe("function");
80+
});
81+
82+
it("passes through model, context, and options", async () => {
83+
const mockStream = createMockAsyncIterable([]);
84+
const baseFn = vi.fn().mockReturnValue(mockStream);
85+
const wrapped = streamWithIdleTimeout(baseFn, 1000);
86+
87+
const model = { api: "openai" } as Parameters<typeof baseFn>[0];
88+
const context = {} as Parameters<typeof baseFn>[1];
89+
const options = {} as Parameters<typeof baseFn>[2];
90+
91+
void wrapped(model, context, options);
92+
93+
expect(baseFn).toHaveBeenCalledWith(model, context, options);
94+
});
95+
96+
it("throws on idle timeout", async () => {
97+
// Create a stream that never yields
98+
const slowStream: AsyncIterable<unknown> = {
99+
[Symbol.asyncIterator]() {
100+
return {
101+
async next() {
102+
// Never resolves - simulates hung LLM
103+
return new Promise<IteratorResult<unknown>>(() => {});
104+
},
105+
};
106+
},
107+
};
108+
109+
const baseFn = vi.fn().mockReturnValue(slowStream);
110+
const wrapped = streamWithIdleTimeout(baseFn, 50); // 50ms timeout
111+
112+
const model = {} as Parameters<typeof baseFn>[0];
113+
const context = {} as Parameters<typeof baseFn>[1];
114+
const options = {} as Parameters<typeof baseFn>[2];
115+
116+
const stream = wrapped(model, context, options) as AsyncIterable<unknown>;
117+
const iterator = stream[Symbol.asyncIterator]();
118+
119+
await expect(iterator.next()).rejects.toThrow(/LLM idle timeout/);
120+
});
121+
122+
it("resets timer on each chunk", async () => {
123+
const chunks = [{ text: "a" }, { text: "b" }, { text: "c" }];
124+
const mockStream = createMockAsyncIterable(chunks);
125+
const baseFn = vi.fn().mockReturnValue(mockStream);
126+
const wrapped = streamWithIdleTimeout(baseFn, 1000);
127+
128+
const model = {} as Parameters<typeof baseFn>[0];
129+
const context = {} as Parameters<typeof baseFn>[1];
130+
const options = {} as Parameters<typeof baseFn>[2];
131+
132+
const stream = wrapped(model, context, options) as AsyncIterable<unknown>;
133+
const results: unknown[] = [];
134+
135+
for await (const chunk of stream) {
136+
results.push(chunk);
137+
}
138+
139+
expect(results).toHaveLength(3);
140+
expect(results).toEqual(chunks);
141+
});
142+
143+
it("handles stream with delays between chunks", async () => {
144+
// Create a stream with small delays
145+
const delayedStream: AsyncIterable<{ text: string }> = {
146+
[Symbol.asyncIterator]() {
147+
let count = 0;
148+
return {
149+
async next() {
150+
if (count < 3) {
151+
await new Promise((r) => setTimeout(r, 10)); // 10ms delay
152+
return { done: false, value: { text: String(count++) } };
153+
}
154+
return { done: true, value: undefined };
155+
},
156+
};
157+
},
158+
};
159+
160+
const baseFn = vi.fn().mockReturnValue(delayedStream);
161+
const wrapped = streamWithIdleTimeout(baseFn, 100); // 100ms timeout - should be enough
162+
163+
const model = {} as Parameters<typeof baseFn>[0];
164+
const context = {} as Parameters<typeof baseFn>[1];
165+
const options = {} as Parameters<typeof baseFn>[2];
166+
167+
const stream = wrapped(model, context, options) as AsyncIterable<{ text: string }>;
168+
const results: { text: string }[] = [];
169+
170+
for await (const chunk of stream) {
171+
results.push(chunk);
172+
}
173+
174+
expect(results).toHaveLength(3);
175+
});
176+
177+
it("aborts controller on idle timeout", async () => {
178+
// Create a stream that never yields
179+
const slowStream: AsyncIterable<unknown> = {
180+
[Symbol.asyncIterator]() {
181+
return {
182+
async next() {
183+
// Never resolves - simulates hung LLM
184+
return new Promise<IteratorResult<unknown>>(() => {});
185+
},
186+
};
187+
},
188+
};
189+
190+
const baseFn = vi.fn().mockReturnValue(slowStream);
191+
const controller = new AbortController();
192+
const wrapped = streamWithIdleTimeout(baseFn, 50, controller); // 50ms timeout
193+
194+
const model = {} as Parameters<typeof baseFn>[0];
195+
const context = {} as Parameters<typeof baseFn>[1];
196+
const options = {} as Parameters<typeof baseFn>[2];
197+
198+
const stream = wrapped(model, context, options) as AsyncIterable<unknown>;
199+
const iterator = stream[Symbol.asyncIterator]();
200+
201+
try {
202+
await iterator.next();
203+
// Should not reach here
204+
expect.fail("Expected timeout error");
205+
} catch (error) {
206+
// Verify the error message is preserved
207+
expect(error).toBeInstanceOf(Error);
208+
expect((error as Error).message).toMatch(/LLM idle timeout/);
209+
210+
// Verify the controller was aborted
211+
expect(controller.signal.aborted).toBe(true);
212+
213+
// Verify the abort reason is the same error
214+
const reason = controller.signal.reason;
215+
expect(reason).toBeInstanceOf(Error);
216+
expect((reason as Error).message).toMatch(/LLM idle timeout/);
217+
}
218+
});
219+
});

0 commit comments

Comments
 (0)