diff --git a/.changeset/stalled-provider-first-byte.md b/.changeset/stalled-provider-first-byte.md new file mode 100644 index 00000000000..2d11921fd62 --- /dev/null +++ b/.changeset/stalled-provider-first-byte.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": patch +--- + +Bound the wait for a provider's first response byte by the request timeout. A provider that accepts a request and returns headers but never sends body data now fails and retries instead of leaving the turn hanging after a tool call completes. The same `timeout` value now covers both the connection phase and the wait for the first byte as a single deadline; streaming responses that have already produced data are unaffected. diff --git a/packages/kilo-docs/pages/code-with-ai/agents/custom-models.md b/packages/kilo-docs/pages/code-with-ai/agents/custom-models.md index d489dc63ebc..20ad7099085 100644 --- a/packages/kilo-docs/pages/code-with-ai/agents/custom-models.md +++ b/packages/kilo-docs/pages/code-with-ai/agents/custom-models.md @@ -387,7 +387,7 @@ You can also set options that apply to all models from a provider: |---|---|---| | `apiKey` | `string` | API key (supports `{env:VAR}` and `{file:...}` syntax in trusted config — see note below) | | `baseURL` | `string` | Override the provider's base API URL | -| `timeout` | `number \| false` | Request timeout in milliseconds. Defaults to `300000` (5 minutes); set to `false` to disable | +| `timeout` | `number \| false` | Request timeout in milliseconds, covering both the wait for response headers and the wait for the first byte of the response body. Defaults to `300000` (5 minutes); set to `false` to disable. Once data starts arriving the timeout no longer applies, so slow streaming responses are never cut short — use `chunkTimeout` for gaps inside a response | | `chunkTimeout` | `number` | Timeout in milliseconds between streamed response chunks. If no chunk arrives within this window, the request is aborted and retried. This catches silent provider dropouts where the TCP connection stays open but SSE streaming stops. Recommended: `15000`–`30000` (15–30 seconds) for providers with unreliable streaming. | {% callout type="warning" title="{env:} / {file:} only resolve in trusted config" %} diff --git a/packages/opencode/src/kilocode/provider/provider.ts b/packages/opencode/src/kilocode/provider/provider.ts index 91aef280731..cc8a452fc96 100644 --- a/packages/opencode/src/kilocode/provider/provider.ts +++ b/packages/opencode/src/kilocode/provider/provider.ts @@ -11,6 +11,7 @@ import { DEFAULT_HEADERS } from "@/kilocode/const" import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" import { optionalOmitUndefined } from "@opencode-ai/core/schema" +import { ProviderError } from "@/provider/error" import { Effect, Schema } from "effect" import type { LanguageModelV3 } from "@ai-sdk/provider" import { mapValues, omit, pickBy } from "remeda" @@ -245,24 +246,37 @@ export function kiloSmallModelPriority(providerID: string): string[] | undefined } // --------------------------------------------------------------------------- -// Fetch timeout wrapper +// Fetch timeout wrappers // Replaces AbortSignal.timeout() with a cancellable setTimeout+AbortController -// so the timer is cleared once response headers arrive. This prevents healthy -// streaming responses from being aborted mid-stream. +// so the timer is cleared once response headers arrive, then hands the remaining +// deadline to wrapFirstByte until the body produces data. One configured +// `timeout` value bounds both phases together, so providers that accept a +// request and go silent cannot hang the agent loop, while healthy streaming +// responses are never aborted mid-stream. // --------------------------------------------------------------------------- +/** + * Resolves the configured request timeout in milliseconds. `timeout: false` + * explicitly disables it (returns `undefined`); any other invalid, unset or + * non-positive value falls back to {@link REQUEST_TIMEOUT_MS} so the wait for a + * provider response is always bounded rather than left open-ended. + */ +export function requestTimeout(options: Record): number | undefined { + const ms = options["timeout"] ?? REQUEST_TIMEOUT_MS + if (ms === false) return undefined + if (typeof ms === "number" && Number.isFinite(ms) && ms > 0) return ms + return REQUEST_TIMEOUT_MS +} + export function buildTimeoutSignal(options: Record): { signal: AbortSignal | undefined clear: () => void } { - const ms = options["timeout"] ?? REQUEST_TIMEOUT_MS - if (ms === false || ms === undefined || ms === null) return { signal: undefined, clear() {} } + const ms = requestTimeout(options) + if (ms === undefined) return { signal: undefined, clear() {} } const controller = new AbortController() - const timer = setTimeout( - () => controller.abort(new DOMException("The operation timed out.", "TimeoutError")), - ms as number, - ) + const timer = setTimeout(() => controller.abort(new DOMException("The operation timed out.", "TimeoutError")), ms) return { signal: controller.signal, clear() { @@ -270,3 +284,67 @@ export function buildTimeoutSignal(options: Record): { }, } } + +/** + * Bounds the wait for the response body's first byte by `ms`. + * + * Response headers do not prove a live stream: a provider can answer 200 with + * SSE headers and then never send data. The connection-phase timeout is cleared + * as soon as headers arrive, so that state used to hang the agent loop forever + * (the turn sits between step-finish and the next step-start with no error). + * + * Only the first byte is guarded. Once any data arrives the wrapper becomes a + * passthrough, so idle gaps inside a streaming response (reasoning, buffering, + * slow token generation) are never touched here and remain opt-in through + * `chunkTimeout`. + */ +export function wrapFirstByte(res: Response, ms: number, ctl: AbortController) { + if (typeof ms !== "number" || ms <= 0) return res + if (!res.body) return res + + const reader = res.body.getReader() + let seen = false + const body = new ReadableStream({ + async pull(ctrl) { + if (seen) { + const part = await reader.read() + if (part.done) return ctrl.close() + return ctrl.enqueue(part.value) + } + + const part = await new Promise>>((resolve, reject) => { + const id = setTimeout(() => { + const err = new ProviderError.ResponseStreamError(`Provider sent no response data within ${ms}ms`) + ctl.abort(err) + void reader.cancel(err) + reject(err) + }, ms) + + reader.read().then( + (part) => { + clearTimeout(id) + resolve(part) + }, + (err) => { + clearTimeout(id) + reject(err) + }, + ) + }) + + seen = true + if (part.done) return ctrl.close() + ctrl.enqueue(part.value) + }, + async cancel(reason) { + ctl.abort(reason) + await reader.cancel(reason) + }, + }) + + return new Response(body, { + headers: new Headers(res.headers), + status: res.status, + statusText: res.statusText, + }) +} diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 8e78708465e..545fbba410a 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -41,6 +41,8 @@ import { patchKiloProviderPrivacy, kiloSmallModelPriority, buildTimeoutSignal, + requestTimeout, + wrapFirstByte, } from "@/kilocode/provider/provider" import * as ModelsRefresh from "@/kilocode/provider/models-refresh" // kilocode_change end @@ -1763,6 +1765,11 @@ export const layer = Layer.effect( const opts = init ?? {} const chunkAbortCtl = typeof chunkTimeout === "number" && chunkTimeout > 0 ? new AbortController() : undefined const timeout = buildTimeoutSignal(options) // kilocode_change - use cancellable timeout for connection phase + // kilocode_change start - extend the same deadline to the first response byte + const firstByteMs = requestTimeout(options) + const firstByteCtl = firstByteMs === undefined ? undefined : new AbortController() + const deadline = firstByteMs === undefined ? undefined : Date.now() + firstByteMs + // kilocode_change end const headerTimeoutMs = headerTimeout === false ? undefined : headerTimeout const headerTimeoutCtl = typeof headerTimeoutMs === "number" ? timeoutController(headerTimeoutMs) : undefined const signals: AbortSignal[] = [] @@ -1771,6 +1778,7 @@ export const layer = Layer.effect( if (chunkAbortCtl) signals.push(chunkAbortCtl.signal) if (headerTimeoutCtl) signals.push(headerTimeoutCtl.signal) if (timeout.signal) signals.push(timeout.signal) // kilocode_change + if (firstByteCtl) signals.push(firstByteCtl.signal) // kilocode_change const combined = signals.length === 0 ? null : signals.length === 1 ? signals[0] : AbortSignal.any(signals) if (combined) opts.signal = combined @@ -1783,8 +1791,12 @@ export const layer = Layer.effect( timeout: false, }).finally(() => headerTimeoutCtl?.clear()) timeout.clear() - if (!chunkAbortCtl) return res - return wrapSSE(res, chunkTimeout, chunkAbortCtl) + // kilocode_change start - hand the remaining deadline to the first-byte guard + const remaining = deadline !== undefined ? deadline - Date.now() : undefined + const live = remaining !== undefined && firstByteCtl ? wrapFirstByte(res, Math.max(remaining, 1), firstByteCtl) : res + if (!chunkAbortCtl) return live + return wrapSSE(live, chunkTimeout, chunkAbortCtl) + // kilocode_change end } catch (err) { timeout.clear() throw err diff --git a/packages/opencode/test/kilocode/fixture/stall-plugin.ts b/packages/opencode/test/kilocode/fixture/stall-plugin.ts new file mode 100644 index 00000000000..144a7b37ab4 --- /dev/null +++ b/packages/opencode/test/kilocode/fixture/stall-plugin.ts @@ -0,0 +1,30 @@ +// Plugin used by the issue #8656 regression tests. +// +// It attaches the simulated socket from stall-transport.ts to the `mock` +// provider through the plugin `config` hook. This is the supported injection +// point: src/provider/provider.ts loads plugins before reading `cfg.provider` +// exactly so hooks can add options such as `fetch`. Injecting here keeps the +// simulation scoped to the test's own instance instead of replacing +// `globalThis.fetch` for the whole test process. + +import { createStallTransport } from "./stall-transport" + +type Options = { state?: unknown; answer?: unknown; provider?: unknown } + +type Draft = { + provider?: Record } | undefined> +} + +export default async (_input: unknown, options?: Options) => ({ + config: async (cfg: Draft) => { + const id = typeof options?.provider === "string" ? options.provider : "mock" + const provider = cfg.provider?.[id] + const state = typeof options?.state === "string" ? options.state : undefined + if (!provider || !state) return + provider.options ??= {} + provider.options["fetch"] = createStallTransport({ + state, + answer: typeof options?.answer === "string" ? options.answer : undefined, + }) + }, +}) diff --git a/packages/opencode/test/kilocode/fixture/stall-transport.ts b/packages/opencode/test/kilocode/fixture/stall-transport.ts new file mode 100644 index 00000000000..8f9c145c169 --- /dev/null +++ b/packages/opencode/test/kilocode/fixture/stall-transport.ts @@ -0,0 +1,119 @@ +// Simulated provider socket for the issue #8656 regression tests. +// +// Only the socket is simulated. The transport is injected as the provider's +// `fetch` option, so Kilo's own fetch wrapper (connection timeout, first-byte +// guard, SSE chunk watchdog), the openai-compatible SDK, SSE parsing, the +// session processor and the agent loop are all the production ones. +// +// Request script: +// 1. title request -> short text answer +// 2. no tool result in the messages -> a bash tool call +// 3. first request carrying a tool result -> SSE headers, body never sends a +// byte (the stall reported in #8656) +// 4. later requests carrying a tool result -> final text answer, so a bounded +// stall can recover through the normal retry path +// +// Progress is mirrored to a JSON file so tests can assert what the provider saw +// without sharing module state with the plugin that loads this file. + +export type StallState = { calls: number; stalls: number; recovered: number } + +const HEAD = { id: "chatcmpl-stall", object: "chat.completion.chunk", created: 0, model: "mock-model" } + +const chunk = (obj: unknown) => `data: ${JSON.stringify(obj)}\n\n` + +const usage = () => + chunk({ ...HEAD, choices: [], usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 } }) + +function sse(body: BodyInit | null) { + return new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }) +} + +function answer(value: string) { + return sse( + [ + chunk({ ...HEAD, choices: [{ index: 0, delta: { role: "assistant", content: value }, finish_reason: null }] }), + chunk({ ...HEAD, choices: [{ index: 0, delta: {}, finish_reason: "stop" }] }), + usage(), + "data: [DONE]\n\n", + ].join(""), + ) +} + +function toolCall(command: string) { + return sse( + [ + chunk({ + ...HEAD, + choices: [ + { + index: 0, + delta: { + role: "assistant", + tool_calls: [ + { + index: 0, + id: "call_1", + type: "function", + function: { name: "bash", arguments: JSON.stringify({ command }) }, + }, + ], + }, + finish_reason: null, + }, + ], + }), + chunk({ ...HEAD, choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }] }), + usage(), + "data: [DONE]\n\n", + ].join(""), + ) +} + +/** Response headers arrive, then the body never produces a byte. */ +function stalling() { + return sse( + new ReadableStream({ + start() {}, + cancel() { + // the first-byte guard cancels this reader when it gives up + }, + }), + ) +} + +export function createStallTransport(input: { state: string; answer?: string; command?: string }) { + const state: StallState = { calls: 0, stalls: 0, recovered: 0 } + const save = () => Bun.write(input.state, JSON.stringify(state)) + + return async (_input: unknown, init?: { body?: unknown }) => { + const body = typeof init?.body === "string" ? init.body : "" + state.calls++ + + if (body.includes("Generate a title")) { + await save() + return answer("Stall repro") + } + + if (!body.includes('"role":"tool"')) { + await save() + return toolCall(input.command ?? "echo repro-8656") + } + + if (state.stalls === 0) { + state.stalls++ + await save() + return stalling() + } + + state.recovered++ + await save() + return answer(input.answer ?? "recovered after the stall") + } +} + +export async function readStallState(file: string): Promise { + const handle = Bun.file(file) + if (!(await handle.exists())) return { calls: 0, stalls: 0, recovered: 0 } + return JSON.parse(await handle.text()) as StallState +} diff --git a/packages/opencode/test/kilocode/issue-8656-stall.test.ts b/packages/opencode/test/kilocode/issue-8656-stall.test.ts new file mode 100644 index 00000000000..c6c939e1ec4 --- /dev/null +++ b/packages/opencode/test/kilocode/issue-8656-stall.test.ts @@ -0,0 +1,180 @@ +import { describe, expect, test } from "bun:test" +import path from "node:path" +import { pathToFileURL } from "node:url" +import { provideTestInstance, tmpdir } from "../fixture/fixture" +import { readStallState } from "./fixture/stall-transport" +import { Server } from "../../src/server/server" + +// Regression coverage for https://github.com/Kilo-Org/kilocode/issues/8656 +// +// Reported symptom: after a tool call finished, `step-finish:tool-calls` was +// recorded and the next `step-start` never arrived, leaving the session busy +// with no error while the HTTP server stayed responsive. +// +// One transport state produces exactly that symptom: the follow-up request that +// carries the tool result gets response headers and then never receives a byte +// of body. The connection-phase timeout is cleared as soon as headers arrive, so +// before the fix nothing bounded that wait. +// +// The simulated socket is injected as the provider's `fetch` through the plugin +// `config` hook (see ../fixture/stall-plugin.ts), so the SDK, Kilo's fetch +// wrapper, SSE parsing, the processor and the agent loop stay production code +// and nothing global is patched. + +const PLUGIN = pathToFileURL(path.join(import.meta.dir, "fixture", "stall-plugin.ts")).href +const ANSWER = "recovered after the stall" + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) + +function settings(state: string, timeout: number | false) { + return { + $schema: "https://app.kilo.ai/config.json", + model: "mock/mock-model", + plugin: [[PLUGIN, { state, answer: ANSWER }]], + provider: { + mock: { + npm: "@ai-sdk/openai-compatible", + name: "Mock", + options: { baseURL: "http://127.0.0.1:1/v1", apiKey: "test", timeout }, + models: { + "mock-model": { + name: "Mock Model", + tool_call: true, + limit: { context: 128000, output: 8192 }, + cost: { input: 0, output: 0 }, + }, + }, + }, + }, + permission: { bash: "allow" }, + } +} + +function project(timeout: number | false) { + return tmpdir<{ state: string }>({ + init: async (dir) => { + const state = path.join(dir, "stall-state.json") + await Bun.write(path.join(dir, "opencode.json"), JSON.stringify(settings(state, timeout), null, 2)) + return { state } + }, + }) +} + +type Part = Record +type Message = { info: Record; parts: Part[] } + +function session(dir: string) { + const app = Server.Default().app + const headers = { "Content-Type": "application/json", "x-kilo-directory": dir } + const query = `directory=${encodeURIComponent(dir)}` + + const json = async (route: string, init?: RequestInit) => { + const res = await app.request(route, { headers, ...init }) + return await res.json() + } + + return { + create: async () => ((await json("/session", { method: "POST", body: "{}" })) as { id: string }).id, + prompt: (id: string, text: string) => + app.request(`/session/${id}/prompt_async`, { + method: "POST", + headers, + body: JSON.stringify({ parts: [{ type: "text", text }] }), + }), + abort: (id: string) => app.request(`/session/${id}/abort`, { method: "POST", headers }), + messages: (id: string) => json(`/session/${id}/message?${query}`) as Promise, + status: async (id: string) => { + const all = (await json(`/session/status?${query}`)) as Record + return all[id]?.type ?? "idle" + }, + } +} + +const timeline = (messages: Message[]) => + messages + .flatMap((m) => m.parts) + .map((p) => + p.type === "tool" ? `tool:${p.tool}:${p.state?.status}` : `${p.type}${p.reason ? `:${p.reason}` : ""}`, + ) + .join(" | ") + +async function until(check: () => Promise, budget: number) { + const deadline = Date.now() + budget + while (Date.now() < deadline) { + if (await check()) return true + await sleep(200) + } + return false +} + +/** Runs a prompt and waits until the provider has stalled the follow-up request. */ +async function stalled(api: ReturnType, state: string) { + const id = await api.create() + await api.prompt(id, "run the echo command") + const ready = await until(async () => { + const stalls = (await readStallState(state)).stalls + const parts = timeline(await api.messages(id)) + return stalls > 0 && parts.includes("step-finish:tool-calls") + }, 30_000) + expect(ready).toBe(true) + return id +} + +describe("issue #8656: provider stalls after a tool call", () => { + test("recovers instead of freezing once the stall is bounded", async () => { + await using tmp = await project(2_000) + await provideTestInstance({ + directory: tmp.path, + fn: async () => { + const api = session(tmp.path) + const id = await stalled(api, tmp.extra.state) + + const done = await until(async () => (await api.status(id)) === "idle", 40_000) + const messages = await api.messages(id) + const assistant = messages.findLast((m) => m.info.role === "assistant") + const text = (assistant?.parts ?? []).find((p) => p.type === "text")?.text + const state = await readStallState(tmp.extra.state) + console.log("[repro] bounded ->", JSON.stringify({ timeline: timeline(messages), text, state })) + + // the turn finishes on its own: the stalled request was aborted, retried + // and answered, so the agent loop never sits frozen + expect(done).toBe(true) + expect(timeline(messages)).toContain("tool:bash:completed") + expect(state.stalls).toBe(1) + expect(state.recovered).toBeGreaterThan(0) + expect(text).toContain(ANSWER) + expect(assistant?.info.error).toBeUndefined() + }, + }) + }, 120_000) + + test("still hangs while the provider holds the connection open and timeout is disabled", async () => { + await using tmp = await project(false) + await provideTestInstance({ + directory: tmp.path, + fn: async () => { + const api = session(tmp.path) + const id = await stalled(api, tmp.extra.state) + try { + await sleep(5_000) + const messages = await api.messages(id) + const parts = timeline(messages) + const assistant = messages.findLast((m) => m.info.role === "assistant") + console.log("[repro] timeout:false ->", JSON.stringify({ timeline: parts, status: await api.status(id) })) + + // the reported freeze, kept reachable only through the documented opt-out + expect(parts.endsWith("step-finish:tool-calls")).toBe(true) + expect(await api.status(id)).toBe("busy") + expect(assistant?.info.error).toBeUndefined() + + // the server itself stays responsive during the freeze + expect((await Server.Default().app.request("/global/health")).status).toBe(200) + } finally { + // never leave a wedged turn behind for fixture teardown + expect((await api.abort(id)).status).toBe(200) + expect(await until(async () => (await api.status(id)) === "idle", 15_000)).toBe(true) + } + }, + }) + }, 120_000) +}) diff --git a/packages/opencode/test/kilocode/provider/first-byte.test.ts b/packages/opencode/test/kilocode/provider/first-byte.test.ts new file mode 100644 index 00000000000..5045032abc5 --- /dev/null +++ b/packages/opencode/test/kilocode/provider/first-byte.test.ts @@ -0,0 +1,84 @@ +import { describe, expect, test } from "bun:test" +import { requestTimeout, wrapFirstByte, REQUEST_TIMEOUT_MS } from "../../../src/kilocode/provider/provider" +import { ProviderError } from "../../../src/provider/error" + +const sse = (body: BodyInit | null) => + new Response(body, { headers: { "content-type": "text/event-stream" }, status: 200 }) + +// A body that delivers `head` immediately, then stays silent forever. +const stalling = (head?: string) => + new ReadableStream({ + start(ctrl) { + if (head) ctrl.enqueue(new TextEncoder().encode(head)) + }, + }) + +const drain = async (res: Response) => { + const reader = res.body!.getReader() + const out: string[] = [] + while (true) { + const part = await reader.read() + if (part.done) return out.join("") + out.push(new TextDecoder().decode(part.value)) + } +} + +describe("requestTimeout", () => { + test("defaults to the shared request timeout", () => { + expect(requestTimeout({})).toBe(REQUEST_TIMEOUT_MS) + }) + + test("honours an explicit value, disables only on false, and bounds invalid input", () => { + expect(requestTimeout({ timeout: 1234 })).toBe(1234) + expect(requestTimeout({ timeout: false })).toBeUndefined() + // invalid/unset values fall back to the default so the wait is always bounded + expect(requestTimeout({ timeout: 0 })).toBe(REQUEST_TIMEOUT_MS) + expect(requestTimeout({ timeout: -1 })).toBe(REQUEST_TIMEOUT_MS) + expect(requestTimeout({ timeout: "nope" })).toBe(REQUEST_TIMEOUT_MS) + expect(requestTimeout({ timeout: null })).toBe(REQUEST_TIMEOUT_MS) + }) +}) + +describe("wrapFirstByte", () => { + test("fails when the body never produces a byte", async () => { + const ctl = new AbortController() + const res = wrapFirstByte(sse(stalling()), 100, ctl) + + await expect(drain(res)).rejects.toBeInstanceOf(ProviderError.ResponseStreamError) + expect(ctl.signal.aborted).toBe(true) + expect((ctl.signal.reason as Error).message).toContain("no response data within 100ms") + }) + + test("stays a passthrough once the first byte arrived, even if the stream then stalls", async () => { + const ctl = new AbortController() + const res = wrapFirstByte(sse(stalling("data: hello\n\n")), 100, ctl) + const reader = res.body!.getReader() + + const first = await reader.read() + expect(new TextDecoder().decode(first.value)).toBe("data: hello\n\n") + + // the guard must not arm again: idle gaps mid-stream stay opt-in (chunkTimeout) + const next = await Promise.race([ + reader.read().then(() => "chunk"), + new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 300)), + ]) + expect(next).toBe("still-waiting") + expect(ctl.signal.aborted).toBe(false) + await reader.cancel("done") + }) + + test("passes a complete body through untouched", async () => { + const ctl = new AbortController() + const res = wrapFirstByte(sse("data: one\n\ndata: two\n\n"), 1_000, ctl) + expect(await drain(res)).toBe("data: one\n\ndata: two\n\n") + expect(ctl.signal.aborted).toBe(false) + }) + + test("is a no-op without a body or when disabled", () => { + const ctl = new AbortController() + const empty = new Response(null, { status: 204 }) + expect(wrapFirstByte(empty, 100, ctl)).toBe(empty) + const res = sse("data: x\n\n") + expect(wrapFirstByte(res, 0, ctl)).toBe(res) + }) +})