diff --git a/.changeset/chunk-idle-timeout-default.md b/.changeset/chunk-idle-timeout-default.md new file mode 100644 index 00000000000..e779f74d4d0 --- /dev/null +++ b/.changeset/chunk-idle-timeout-default.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": patch +--- + +Prevent agent and subagent sessions from stalling indefinitely with a 60-second model-stream idle watchdog that pauses while local tools run. Set `chunkTimeout: false` to disable it. diff --git a/packages/core/src/v1/config/provider.ts b/packages/core/src/v1/config/provider.ts index 03593cb644a..c906fe23282 100644 --- a/packages/core/src/v1/config/provider.ts +++ b/packages/core/src/v1/config/provider.ts @@ -115,9 +115,11 @@ export const Info = Schema.Struct({ description: "Timeout in milliseconds to wait for response headers. Provider integrations may set defaults. Set to false to disable timeout.", }), - chunkTimeout: Schema.optional(PositiveInt).annotate({ + // kilocode_change: accept `false` so internal callers can disable the + // watchdog. PositiveInt already excludes 0, so a public zero stays invalid. + chunkTimeout: Schema.optional(Schema.Union([PositiveInt, Schema.Literal(false)])).annotate({ description: - "Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted.", + "Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted. Set to false to disable the idle watchdog.", }), }), [Schema.Record(Schema.String, Schema.Any)], diff --git a/packages/opencode/src/kilocode/session/llm.ts b/packages/opencode/src/kilocode/session/llm.ts index c6627a28b9f..fc317270ef9 100644 --- a/packages/opencode/src/kilocode/session/llm.ts +++ b/packages/opencode/src/kilocode/session/llm.ts @@ -1,12 +1,16 @@ -import type { ModelMessage } from "ai" +import type { LanguageModelV2StreamPart } from "@ai-sdk/provider" import * as Stream from "effect/Stream" +import { ProviderError } from "@/provider/error" import type { LLMEvent } from "@opencode-ai/llm" -import type { Logger } from "@opencode-ai/core/util/log" +import type { ModelMessage } from "ai" import type { Provider } from "@/provider/provider" import { KiloSessionOverflow } from "./overflow" const SAFETY = 2048 const MIN_OUTPUT = 1024 +const DEFAULT_CHUNK_IDLE_MS = 60_000 + +type FullStreamPart = LanguageModelV2StreamPart export namespace KiloLLM { // Stream failures and interruptions propagate while text deltas are collected. @@ -17,20 +21,198 @@ export namespace KiloLLM { ) } - export function timeout(input: { + /** + * Resolves the configured chunk idle timeout in milliseconds, or `undefined` + * when the watchdog should be disabled. + * + * Precedence: + * 1. prepared `options.chunkTimeout` + * 2. provider `fallback.chunkTimeout` + * 3. DEFAULT_CHUNK_IDLE_MS + * + * Rules: + * - positive finite number wins. + * - public `false` or internal `0` disables (returns undefined). + * - invalid prepared values (non-number, negative, non-finite, strings, ...) + * fall through to the provider fallback. The same rules apply at every + * layer. + */ + export function resolveIdleMs(input: { options: Record fallback?: Record - log?: Pick - }): { timeout?: { chunkMs: number } } { - const value = - typeof input.options["chunkTimeout"] === "number" - ? input.options["chunkTimeout"] - : typeof input.fallback?.["chunkTimeout"] === "number" - ? input.fallback["chunkTimeout"] - : undefined - if (!value) return {} - input.log?.debug("chunk idle timeout configured", { chunkTimeout: value }) - return { timeout: { chunkMs: value } } + }): number | undefined { + const prepared = resolve(input.options["chunkTimeout"]) + if (prepared.disabled) return undefined + if (prepared.value !== undefined) return prepared.value + const fallback = resolve(input.fallback?.["chunkTimeout"]) + if (fallback.disabled) return undefined + if (fallback.value !== undefined) return fallback.value + return DEFAULT_CHUNK_IDLE_MS + } + + // Tri-state: `disabled` means "explicitly off"; `value` is a usable ms count. + // `null`/`undefined`/invalid numeric values are treated as not-configured. + function resolve(value: unknown): { value: number | undefined; disabled: boolean } { + if (value === false || value === 0) return { value: undefined, disabled: true } + if (value == null) return { value: undefined, disabled: false } + if (typeof value !== "number") return { value: undefined, disabled: false } + if (!Number.isFinite(value)) return { value: undefined, disabled: false } + if (value <= 0) return { value: undefined, disabled: false } + return { value, disabled: false } + } + + /** + * Wraps an AI SDK `fullStream` with a Kilo-owned per-event idle watchdog. + * + * Behavior: + * - `idleMs === undefined` returns the stream unchanged (disabled). + * - every raw AI SDK event resets the idle timer. + * - non-provider-executed `tool-call` adds an active tool id; matching + * `tool-result` / `tool-error` removes it. While any local tool id is + * active, the watchdog is suspended (long-running tool work is not a + * stall). + * - provider-executed `tool-call` does not suspend the watchdog and no id + * is tracked — those are settled server-side and a missing result is a + * real stall. + * - parallel local tool calls remain suspended until the last one settles. + * - the wrapper fails the stream with `ProviderError.ResponseStreamError` + * on stall. Existing `MessageV2` retry mapping handles that error. + * + * The wrapper is implemented against `AsyncIterable` so it composes with + * any stream the AI SDK exposes, including its native `fullStream`. The + * outer `Stream` is rebuilt from the wrapped iterable, which keeps the + * contract simple: one pull = one raw event. + */ + export function watchdogStream( + stream: Stream.Stream, + idleMs: number | undefined, + abort?: AbortController, + ): Stream.Stream { + if (idleMs === undefined) return stream + const source = Stream.toAsyncIterable(stream) + return Stream.fromAsyncIterable(watchdogAsyncIterable(source, idleMs, abort), (e) => + e instanceof Error ? e : new Error(String(e)), + ) + } + + /** + * Wraps an `AsyncIterable` of raw AI SDK `fullStream` parts with the same + * Kilo-owned per-event idle watchdog. Use this when the upstream is already + * an `AsyncIterable` (e.g. the AI SDK's `fullStream`) so we avoid a + * Stream → AsyncIterable → Stream round-trip. + */ + export function watchdogAsyncIterable( + source: AsyncIterable, + idleMs: number | undefined, + abort?: AbortController, + ): AsyncIterable { + if (idleMs === undefined) return source + return { [Symbol.asyncIterator]: () => watchIterator(source, idleMs, abort) } + } + + /** + * Implemented as a hand-rolled `AsyncIterator` rather than an `async + * function*` generator. An async generator's `.return()` cannot preempt an + * in-flight internal `await`: per spec, when the generator is suspended + * mid-`await` (as opposed to suspended at a `yield`), a `.return()` call + * only takes effect once that `await` settles on its own. When the source + * is genuinely stalled — the exact case this watchdog exists to catch — + * that `await` never settles, so a caller that wants to cancel promptly + * (e.g. Effect interrupting the consuming Stream) would hang forever + * waiting for cleanup instead. A plain iterator object's `return()` runs + * immediately and forwards to the underlying source's `return()` without + * waiting on any outstanding pull, matching how interruption already + * behaves for the unwrapped upstream iterator. + */ + function watchIterator( + source: AsyncIterable, + idleMs: number, + abort?: AbortController, + ): AsyncIterator { + const local = new Set() + const iter = source[Symbol.asyncIterator]() + let suspended = false + let closed = false + return { + async next(): Promise> { + if (closed) return { done: true, value: undefined } + try { + // Decide BEFORE pulling whether the next event is allowed to take as + // long as upstream needs. Local tool work in flight must not be timed + // out — the AI SDK only emits a tool-result / tool-error once the + // client-side tool has actually finished. + const pull = suspended ? iter.next() : raceWithTimeout(iter.next(), idleMs, abort) + const value = await pull + suspended = false + if (value.done) { + closed = true + await safeClose(iter) + return value + } + const part = value.value + trackPart(local, part) + suspended = local.size > 0 + return { done: false, value: part } + } catch (e) { + closed = true + await safeClose(iter) + throw e + } + }, + async return(value?: unknown): Promise> { + if (!closed) { + closed = true + await safeClose(iter) + } + return { done: true, value: value as FullStreamPart } + }, + } + } + + function trackPart(local: Set, part: FullStreamPart) { + if (!part || typeof part !== "object") return + const t = (part as { type?: unknown }).type + if (t === "tool-call") { + const call = part as unknown as { + toolCallId?: unknown + providerExecuted?: unknown + } + if (call.providerExecuted === true) return + if (typeof call.toolCallId !== "string") return + local.add(call.toolCallId) + return + } + if (t === "tool-result" || t === "tool-error") { + const call = part as unknown as { toolCallId?: unknown } + if (typeof call.toolCallId !== "string") return + local.delete(call.toolCallId) + } + } + + function raceWithTimeout(promise: Promise, ms: number, abort?: AbortController): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + const err = new ProviderError.ResponseStreamError(`AI SDK stream stalled: no event for ${ms}ms`) + if (abort && !abort.signal.aborted) { + abort.abort(err) + } + reject(err) + }, ms) + promise.then( + (v) => { + clearTimeout(timer) + resolve(v) + }, + (e) => { + clearTimeout(timer) + reject(e) + }, + ) + }) + } + + async function safeClose(iter: AsyncIterator) { + if (typeof iter.return === "function") await iter.return() } export function needsEstimate(input: { model: Provider.Model; configured: number | undefined }) { diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index ce38cc73867..5520e4d78ec 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -389,7 +389,9 @@ const live: Layer.Layer< toolChoice: input.toolChoice, maxOutputTokens: prepared.params.maxOutputTokens, abortSignal: input.abort, - ...KiloLLM.timeout({ options: prepared.params.options, fallback: item.options, log: l }), // kilocode_change + // kilocode_change: AI SDK's built-in chunk timeout is removed in favor + // of a Kilo-owned per-event watchdog applied to the raw fullStream + // before LLMAISDK.toLLMEvents normalization (see below). headers: prepared.headers, maxRetries: input.retries ?? 0, messages: prepared.messages, @@ -417,7 +419,14 @@ const live: Layer.Layer< }) // kilocode_change end // kilocode_change start - capture eligible session export request completion off the stream path - if (!exportable) return { type: "ai-sdk" as const, result } + // kilocode_change: resolve per-subscription idle watchdog so concurrent + // sessions each get their own timer. Computed here (not at the stream + // consumer) so the resolved value travels with the returned fullStream. + const idleMs = KiloLLM.resolveIdleMs({ + options: prepared.params.options, + fallback: item.options, + }) + if (!exportable) return { type: "ai-sdk" as const, result, idleMs } return { type: "ai-sdk" as const, result: { @@ -431,6 +440,7 @@ const live: Layer.Layer< retries: input.retries ?? 0, }), }, + idleMs, } // kilocode_change end }) @@ -451,10 +461,22 @@ const live: Layer.Layer< // Adapter seam: both runtimes expose the same LLMEvent stream. Native // already returns one; AI SDK streams are converted here. const state = LLMAISDK.adapterState() - return Stream.fromAsyncIterable(result.result.fullStream, (e) => - e instanceof Error ? e : new Error(String(e)), - ).pipe( - Stream.mapEffect((event) => LLMAISDK.toLLMEvents(state, event)), + // kilocode_change: wrap the raw AI SDK fullStream with the Kilo + // idle watchdog before normalization. Per-subscription timer was + // resolved inside `run` and travels with the result. Pass the + // scoped controller so the watchdog can abort a stalled source + // and avoid hanging cleanup. + const watched = KiloLLM.watchdogAsyncIterable( + result.result.fullStream as AsyncIterable, + result.idleMs, + ctrl, + ) + return Stream.fromAsyncIterable(watched, (e) => (e instanceof Error ? e : new Error(String(e)))).pipe( + // kilocode_change: the watchdog consumes raw LanguageModelV2 parts; + // cast back to the TextStreamPart shape LLMAISDK.toLLMEvents expects. + Stream.mapEffect((event) => + LLMAISDK.toLLMEvents(state, event as Parameters[1]), + ), Stream.flatMap((events) => Stream.fromIterable(events)), ) }), diff --git a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts new file mode 100644 index 00000000000..c222a1c71e7 --- /dev/null +++ b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts @@ -0,0 +1,558 @@ +import { NodeFileSystem } from "@effect/platform-node" +import { afterEach, describe, expect } from "bun:test" +import { Effect, Exit, Fiber, Layer } from "effect" +import { FetchHttpClient } from "effect/unstable/http" +import fs from "fs/promises" +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { Database } from "@opencode-ai/core/database/database" +import type { SessionID } from "../../src/session/schema" +import path from "path" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import * as Log from "@opencode-ai/core/util/log" +import { Agent as AgentSvc } from "../../src/agent/agent" +import { BackgroundJob } from "../../src/background/job" +import { Bus } from "../../src/bus" +import { Command } from "../../src/command" +import { Auth } from "../../src/auth" +import { Config } from "../../src/config/config" +import { RuntimeFlags } from "../../src/effect/runtime-flags" +import { EventV2Bridge } from "../../src/event-v2-bridge" +import { Env } from "../../src/env" +import { Format } from "../../src/format" +import { Git } from "../../src/git" +import { Image } from "../../src/image/image" +import { LSP } from "../../src/lsp/lsp" +import { MCP } from "../../src/mcp" +import { Permission } from "../../src/permission" +import { Plugin } from "../../src/plugin" +import { Provider as ProviderSvc } from "../../src/provider/provider" +import { Question } from "../../src/question" +import { Reference } from "../../src/reference/reference" +import { RepositoryCache } from "../../src/reference/repository-cache" +import { SessionCompaction } from "../../src/session/compaction" +import { Instruction } from "../../src/session/instruction" +import { LLM } from "../../src/session/llm" +import { SessionProcessor } from "../../src/session/processor" +import { SessionPrompt } from "../../src/session/prompt" +import { SessionRevert } from "../../src/session/revert" +import { SessionRunState } from "../../src/session/run-state" +import { Session } from "../../src/session/session" +import { SessionStatus } from "../../src/session/status" +import { SystemPrompt } from "../../src/session/system" +import { SessionSummary } from "../../src/session/summary" +import { Todo } from "../../src/session/todo" +import { Skill } from "../../src/skill" +import { Snapshot } from "../../src/snapshot" +import { Storage } from "../../src/storage/storage" +import { SyncEvent } from "../../src/sync" +import { Ripgrep } from "@opencode-ai/core/filesystem/ripgrep" +import { ToolRegistry } from "../../src/tool/registry" +import { Truncate } from "../../src/tool/truncate" +import { MemoryService } from "@kilocode/kilo-memory/effect/service" +import { provideTmpdirServer } from "../fixture/fixture" +import { awaitWithTimeout, pollWithTimeout, testEffect } from "../lib/effect" +import { reply, TestLLMServer } from "../lib/llm-server" + +void Log.init({ print: false }) + +afterEach(async () => { + // Dispose all test instances between integration scenarios. + const { disposeAllInstances } = await import("../fixture/fixture") + await disposeAllInstances() +}) + +const summary = Layer.succeed( + SessionSummary.Service, + SessionSummary.Service.of({ + summarize: () => Effect.void, + diff: () => Effect.succeed([]), + computeDiff: () => Effect.succeed([]), + }), +) + +const mcp = Layer.succeed( + MCP.Service, + MCP.Service.of({ + status: () => Effect.succeed({}), + clients: () => Effect.succeed({}), + tools: () => Effect.succeed({}), + prompts: () => Effect.succeed({}), + resources: () => Effect.succeed({}), + add: () => Effect.succeed({ status: { status: "disabled" as const } }), + connect: () => Effect.void, + disconnect: () => Effect.void, + getPrompt: () => Effect.succeed(undefined), + readResource: () => Effect.succeed(undefined), + startAuth: () => Effect.die("unexpected MCP auth in watchdog tests"), + authenticate: () => Effect.die("unexpected MCP auth in watchdog tests"), + finishAuth: () => Effect.die("unexpected MCP auth in watchdog tests"), + removeAuth: () => Effect.void, + supportsOAuth: () => Effect.succeed(false), + hasStoredTokens: () => Effect.succeed(false), + getAuthStatus: () => Effect.succeed("not_authenticated" as const), + }), +) + +const lsp = Layer.succeed( + LSP.Service, + LSP.Service.of({ + init: () => Effect.void, + status: () => Effect.succeed([]), + hasClients: () => Effect.succeed(false), + touchFile: () => Effect.void, + diagnostics: () => Effect.succeed({}), + hover: () => Effect.succeed(undefined), + definition: () => Effect.succeed([]), + references: () => Effect.succeed([]), + implementation: () => Effect.succeed([]), + documentSymbol: () => Effect.succeed([]), + workspaceSymbol: () => Effect.succeed([]), + prepareCallHierarchy: () => Effect.succeed([]), + incomingCalls: () => Effect.succeed([]), + outgoingCalls: () => Effect.succeed([]), + }), +) + +const status = Layer.mergeAll(SessionStatus.defaultLayer, Bus.layer) +const run = SessionRunState.layer.pipe(Layer.provide(status)) +const infra = Layer.mergeAll(NodeFileSystem.layer, CrossSpawnSpawner.defaultLayer) + +function makeHttp() { + const deps = Layer.mergeAll( + Session.defaultLayer, + BackgroundJob.defaultLayer, + Snapshot.defaultLayer, + LLM.defaultLayer, + Env.defaultLayer, + AgentSvc.defaultLayer, + Command.defaultLayer, + Permission.defaultLayer, + Plugin.defaultLayer, + Config.defaultLayer, + RuntimeFlags.layer(), + ProviderSvc.defaultLayer, + lsp, + mcp, + FSUtil.defaultLayer, + Reference.defaultLayer, + SyncEvent.defaultLayer, + EventV2Bridge.defaultLayer, + Database.defaultLayer, + status, + MemoryService.layer, + ).pipe(Layer.provideMerge(infra)) + const question = Question.layer.pipe(Layer.provideMerge(deps)) + const todo = Todo.layer.pipe(Layer.provideMerge(deps)) + const registry = ToolRegistry.layer.pipe( + Layer.provide(Skill.defaultLayer), + Layer.provide(FetchHttpClient.layer), + Layer.provide(CrossSpawnSpawner.defaultLayer), + Layer.provide(RepositoryCache.defaultLayer), + Layer.provide(Ripgrep.defaultLayer), + Layer.provide(Format.defaultLayer), + Layer.provide(Git.defaultLayer), + Layer.provide(Reference.defaultLayer), + Layer.provide(Command.defaultLayer), + Layer.provide(Auth.defaultLayer), + Layer.provideMerge(todo), + Layer.provideMerge(question), + Layer.provideMerge(deps), + ) + const trunc = Truncate.layer.pipe(Layer.provideMerge(deps)) + const proc = SessionProcessor.layer.pipe( + Layer.provide(summary), + Layer.provide(Image.defaultLayer), + Layer.provideMerge(deps), + ) + const compact = SessionCompaction.layer.pipe(Layer.provideMerge(proc), Layer.provideMerge(deps)) + return Layer.mergeAll( + TestLLMServer.layer, + SessionPrompt.layer.pipe( + Layer.provide(SessionRevert.defaultLayer), + Layer.provide(Image.defaultLayer), + Layer.provide(summary), + Layer.provideMerge(run), + Layer.provideMerge(compact), + Layer.provideMerge(proc), + Layer.provideMerge(registry), + Layer.provideMerge(trunc), + Layer.provideMerge(question), + Layer.provide(Instruction.defaultLayer), + Layer.provide(SystemPrompt.defaultLayer), + Layer.provideMerge(deps), + ), + ).pipe( + Layer.provide( + Layer.mergeAll( + summary, + deps, + Config.defaultLayer, + RuntimeFlags.layer(), + BackgroundJob.defaultLayer, + Bus.layer, + infra, + Storage.defaultLayer, + Reference.defaultLayer, + ), + ), + ) +} + +const it = testEffect(makeHttp()) + +const cfg = { + provider: { + test: { + name: "Test", + id: "test", + env: [], + npm: "@ai-sdk/openai-compatible", + models: { + "test-model": { + id: "test-model", + name: "Test Model", + attachment: false, + reasoning: false, + temperature: false, + tool_call: true, + release_date: "2025-01-01", + limit: { context: 100000, output: 10000 }, + cost: { input: 0, output: 0 }, + options: { chunkTimeout: 1_000 }, + }, + }, + options: { + apiKey: "test-key", + baseURL: "http://localhost:1/v1", + }, + }, + }, +} + +function providerCfg(url: string) { + return { + ...cfg, + provider: { + ...cfg.provider, + test: { + ...cfg.provider.test, + options: { + ...cfg.provider.test.options, + baseURL: url, + chunkTimeout: false as const, + }, + }, + }, + } +} + +const worktreeFile = (dir: string, name: string) => path.join(dir, name) + +const exists = (file: string) => + Effect.promise(() => + fs + .access(file) + .then(() => true) + .catch(() => false), + ) + +// The production bash tool runs every command through a *login* shell +// (`bash -l -c ...`, see src/shell/shell.ts) so `~/.bashrc` and shell +// aliases behave the same as an interactive terminal. Git for Windows' +// login-shell startup rescans the full Windows `PATH`, which is slower +// than the Unix shells used elsewhere in this file. Give the marker file +// these tests poll for a little extra headroom there, on top of the tests +// A/C `config.shell: "bash"` override that makes the bash tool actually +// use git-bash instead of cmd.exe on Windows (see those config comments). +const waitForFile = (file: string, label: string, duration = process.platform === "win32" ? 15_000 : 5_000) => + pollWithTimeout( + Effect.gen(function* () { + const ok = yield* exists(file) + return ok ? true : undefined + }), + label, + duration, + ) + +const touch = (file: string) => Effect.promise(() => fs.writeFile(file, "")) + +const waitForRunningTool = (sessionID: SessionID, sessions: Session.Interface, label: string, duration = 15_000) => + pollWithTimeout( + Effect.gen(function* () { + const msgs = yield* sessions.messages({ sessionID }) + const running = msgs + .flatMap((msg) => msg.parts) + .find((part) => part.type === "tool" && part.state.status === "running") + return running ? running : undefined + }), + label, + duration, + ) + +const waitForRequestHit = (llm: TestLLMServer["Service"], needle: string, label: string) => + pollWithTimeout( + Effect.gen(function* () { + const hits = yield* llm.hits + const matched = hits.filter((hit) => JSON.stringify(hit.body).includes(needle)) + return matched.length > 0 ? matched : undefined + }), + label, + 5_000, + ) + +const matchContains = (needle: string) => (hit: { body: Record }) => + JSON.stringify(hit.body).includes(needle) + +const assertNotInterrupted = (parts: SessionV1.WithParts["parts"]) => { + for (const part of parts) { + if (part.type === "tool") { + expect(part.state.status).toBe("completed") + if (part.state.status === "completed") { + expect(part.state.metadata?.interrupted).not.toBe(true) + expect(part.state.output).not.toContain("Tool execution aborted") + } + } + } +} + +// kilocode_change: normalize to forward slashes before embedding in the +// shell script. `ready`/`release` come from `path.join`, which yields +// backslash-separated paths on Windows; inside a double-quoted git-bash +// string a literal backslash is an escape character, so a Windows path can +// silently mangle into the wrong filename (or a path bash's `[ -f ... ]` +// test can't resolve) rather than throwing. Git-bash/MSYS accept +// forward-slash paths natively, so this is safe on every platform this +// suite runs on. +const posixPath = (p: string) => p.replaceAll("\\", "/") + +const bashGate = (dir: string, ready: string, release: string) => + `touch ${JSON.stringify(posixPath(ready))} && while [ ! -f ${JSON.stringify(posixPath(release))} ]; do sleep 0.05; done && echo done` + +describe("session stream watchdog integration", () => { + it.live( + "A: root session long-running Bash is not interrupted by the idle watchdog", + () => + provideTmpdirServer( + Effect.fnUntraced(function* ({ dir, llm }) { + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const chat = yield* sessions.create({ title: "Root long bash" }) + const ready = worktreeFile(dir, "bash-ready") + const release = worktreeFile(dir, "bash-release") + + yield* llm.tool("bash", { + command: bashGate(dir, ready, release), + description: "Long running bash command", + timeout: 60_000, + workdir: dir, + }) + yield* llm.text("bash complete") + + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "run a long bash command" }], + }) + + const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + + yield* llm.wait(1) + yield* waitForRunningTool(chat.id, sessions, "root bash tool never started") + yield* waitForFile(ready, "root bash readiness marker never appeared") + yield* Effect.sleep("1500 millis") + + yield* touch(release) + + const exit = yield* awaitWithTimeout(Fiber.await(fiber), "root bash loop did not finish", "15 seconds") + expect(Exit.isSuccess(exit)).toBe(true) + + // Check all messages in the session for interrupted tools + const allMessages = yield* sessions.messages({ sessionID: chat.id }) + for (const msg of allMessages) { + assertNotInterrupted(msg.parts) + } + }), + { + git: true, + // kilocode_change: without an explicit `shell`, the bash tool's + // `defaultShell()` falls back to cmd.exe on Windows (see + // packages/core/src/tool/bash.ts), which cannot run bashGate's + // POSIX syntax (`touch`, `[ -f ... ]`, `while ... done`). That + // made `touch` fail immediately and silently, so the readiness + // marker never appeared regardless of how long the test waited. + config: (url) => ({ ...providerCfg(url), shell: "bash", permission: { bash: "allow" } }), + }, + ), + { timeout: 30_000 }, + ) + + it.live( + "B: root foreground TaskTool child with held LLM response is not interrupted", + () => + provideTmpdirServer( + Effect.fnUntraced(function* ({ dir, llm }) { + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const chat = yield* sessions.create({ title: "Root foreground child" }) + const gate = Promise.withResolvers() + + yield* llm.tool("task", { + description: "Foreground child task", + prompt: "child task: say hello", + subagent_type: "child", + }) + yield* llm.pushMatch( + matchContains("child task: say hello"), + reply().wait(gate.promise).text("child done").stop(), + ) + yield* llm.text("parent done") + + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "run a foreground child" }], + }) + + const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + + yield* llm.wait(1) + yield* waitForRunningTool(chat.id, sessions, "root task tool never started") + yield* waitForRequestHit(llm, "child task: say hello", "child LLM request never hit server") + yield* Effect.sleep("1500 millis") + + gate.resolve(undefined) + + const exit = yield* awaitWithTimeout( + Fiber.await(fiber), + "root foreground child loop did not finish", + "15 seconds", + ) + expect(Exit.isSuccess(exit)).toBe(true) + + // Check all messages in root and child sessions for interrupted tools + const allMessages = yield* sessions.messages({ sessionID: chat.id }) + for (const msg of allMessages) { + assertNotInterrupted(msg.parts) + } + + const children = yield* sessions.children(chat.id) + expect(children).toHaveLength(1) + const childMessages = yield* sessions.messages({ sessionID: children[0]!.id }) + for (const msg of childMessages) { + assertNotInterrupted(msg.parts) + } + }), + { + git: true, + config: (url) => ({ + ...providerCfg(url), + permission: { bash: "allow", task: "allow" }, + agent: { + child: { + model: "test/test-model", + mode: "subagent", + options: { chunkTimeout: false }, + permission: { bash: "allow", task: "allow" }, + }, + }, + }), + }, + ), + { timeout: 30_000 }, + ) + + it.live( + "C: child session long-running Bash while root awaits it is not interrupted", + () => + provideTmpdirServer( + Effect.fnUntraced(function* ({ dir, llm }) { + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const chat = yield* sessions.create({ title: "Nested long bash" }) + const ready = worktreeFile(dir, "child-bash-ready") + const release = worktreeFile(dir, "child-bash-release") + + yield* llm.tool("task", { + description: "Nested child task", + prompt: "child task: run a long bash command", + subagent_type: "child", + }) + yield* llm.pushMatch( + matchContains("child task: run a long bash command"), + reply().tool("bash", { + command: bashGate(dir, ready, release), + description: "Long running child bash command", + timeout: 60_000, + workdir: dir, + }), + ) + yield* llm.text("child done") + yield* llm.text("parent done") + + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "run a nested child" }], + }) + + const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + + yield* llm.wait(1) + yield* waitForRunningTool(chat.id, sessions, "root task tool never started") + yield* waitForRequestHit(llm, "child task: run a long bash command", "child LLM request never hit server") + + const children = yield* sessions.children(chat.id) + expect(children).toHaveLength(1) + const childID = children[0]!.id + + yield* waitForRunningTool(childID, sessions, "child bash tool never started") + yield* waitForFile(ready, "child bash readiness marker never appeared") + yield* Effect.sleep("1500 millis") + + yield* touch(release) + + const exit = yield* awaitWithTimeout( + Fiber.await(fiber), + "nested child bash loop did not finish", + "15 seconds", + ) + expect(Exit.isSuccess(exit)).toBe(true) + + // Check all messages in root and child sessions for interrupted tools + const rootMessages = yield* sessions.messages({ sessionID: chat.id }) + for (const msg of rootMessages) { + assertNotInterrupted(msg.parts) + } + + // Reuse childID captured above + const childMessages = yield* sessions.messages({ sessionID: childID }) + for (const msg of childMessages) { + assertNotInterrupted(msg.parts) + } + }), + { + git: true, + // kilocode_change: see the matching comment on test A — without + // this, the nested child's bash tool falls back to cmd.exe on + // Windows and the readiness marker never appears. + config: (url) => ({ + ...providerCfg(url), + shell: "bash", + permission: { bash: "allow", task: "allow" }, + agent: { + child: { + model: "test/test-model", + mode: "subagent", + permission: { bash: "allow", task: "allow" }, + }, + }, + }), + }, + ), + { timeout: 30_000 }, + ) +}) diff --git a/packages/opencode/test/kilocode/session/llm.test.ts b/packages/opencode/test/kilocode/session/llm.test.ts index dc6c3dfa29a..c51c4ca1a2f 100644 --- a/packages/opencode/test/kilocode/session/llm.test.ts +++ b/packages/opencode/test/kilocode/session/llm.test.ts @@ -3,36 +3,84 @@ import { Effect, Stream } from "effect" import { LLMEvent } from "@opencode-ai/llm" import { KiloLLM } from "@/kilocode/session/llm" -describe("kilocode.session.llm.timeout", () => { +describe("kilocode.session.llm.resolveIdleMs", () => { test("uses prepared options before the provider fallback", () => { - const result = KiloLLM.timeout({ - options: { chunkTimeout: 15_000 }, - fallback: { chunkTimeout: 30_000 }, - }) - - expect(result).toEqual({ timeout: { chunkMs: 15_000 } }) + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: 15_000 }, + fallback: { chunkTimeout: 30_000 }, + }), + ).toBe(15_000) }) test("uses the provider fallback when prepared options omit the timeout", () => { - const result = KiloLLM.timeout({ - options: {}, - fallback: { chunkTimeout: 30_000 }, - }) - - expect(result).toEqual({ timeout: { chunkMs: 30_000 } }) + expect( + KiloLLM.resolveIdleMs({ + options: {}, + fallback: { chunkTimeout: 30_000 }, + }), + ).toBe(30_000) }) test("uses the provider fallback when the prepared value is not a number", () => { - const result = KiloLLM.timeout({ - options: { chunkTimeout: "15_000" }, - fallback: { chunkTimeout: 30_000 }, - }) + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: "15_000" }, + fallback: { chunkTimeout: 30_000 }, + }), + ).toBe(30_000) + }) + + test("defaults the chunk idle timeout to 60_000 ms when no override is configured", () => { + expect(KiloLLM.resolveIdleMs({ options: {} })).toBe(60_000) + }) + + test("returns undefined when prepared is false (disabled)", () => { + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: false }, + fallback: { chunkTimeout: 30_000 }, + }), + ).toBeUndefined() + }) + + test("returns undefined when prepared is 0 (internal disable)", () => { + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: 0 }, + fallback: { chunkTimeout: 30_000 }, + }), + ).toBeUndefined() + }) - expect(result).toEqual({ timeout: { chunkMs: 30_000 } }) + test("returns undefined when provider fallback is false", () => { + expect( + KiloLLM.resolveIdleMs({ + options: {}, + fallback: { chunkTimeout: false }, + }), + ).toBeUndefined() }) - test("omits the timeout when it is not configured", () => { - expect(KiloLLM.timeout({ options: {} })).toEqual({}) + test("falls through invalid prepared values to provider fallback", () => { + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: -1 }, + fallback: { chunkTimeout: 5_000 }, + }), + ).toBe(5_000) + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: Number.POSITIVE_INFINITY }, + fallback: { chunkTimeout: 5_000 }, + }), + ).toBe(5_000) + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: Number.NaN }, + fallback: { chunkTimeout: 5_000 }, + }), + ).toBe(5_000) }) }) diff --git a/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts b/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts new file mode 100644 index 00000000000..6c3efcb4cae --- /dev/null +++ b/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts @@ -0,0 +1,282 @@ +import { describe, expect, test } from "bun:test" +import { Effect, Stream } from "effect" +import type { LanguageModelV2CallWarning, LanguageModelV2StreamPart } from "@ai-sdk/provider" +import { KiloLLM } from "@/kilocode/session/llm" +import { ProviderError } from "@/provider/error" + +type FullStreamPart = LanguageModelV2StreamPart + +function part(type: string, extra: Record = {}): FullStreamPart { + return { type, ...extra } as unknown as FullStreamPart +} + +async function run(eff: Effect.Effect) { + return await Effect.runPromise(eff) +} + +function fromSchedule(events: Array<[number, FullStreamPart]>, end: number): Stream.Stream { + // Each `[at, value]` is an ABSOLUTE time in milliseconds from stream start. + // This matches how the tests are written: post-tool events (finish-step, + // finish) are scheduled within a short idle window of the tool-result so + // the watchdog, after the local active set drains, still receives the next + // event in time. + return Stream.fromAsyncIterable( + (async function* () { + const start = Date.now() + for (const [at, value] of events) { + const wait = at - (Date.now() - start) + if (wait > 0) await new Promise((r) => setTimeout(r, wait)) + yield value + } + await new Promise((r) => setTimeout(r, end)) + })(), + (e) => e as never, + ) +} + +describe("kilocode.session.llm.resolveIdleMs", () => { + test("returns prepared positive finite value as-is", () => { + const out = KiloLLM.resolveIdleMs({ options: { chunkTimeout: 15_000 }, fallback: { chunkTimeout: 30_000 } }) + expect(out).toBe(15_000) + }) + + test("falls back to provider value when prepared is missing", () => { + const out = KiloLLM.resolveIdleMs({ options: {}, fallback: { chunkTimeout: 30_000 } }) + expect(out).toBe(30_000) + }) + + test("falls back to provider value when prepared is a non-number string", () => { + const out = KiloLLM.resolveIdleMs({ + options: { chunkTimeout: "15_000" }, + fallback: { chunkTimeout: 30_000 }, + }) + expect(out).toBe(30_000) + }) + + test("falls back to provider value when prepared is negative", () => { + const out = KiloLLM.resolveIdleMs({ + options: { chunkTimeout: -1 }, + fallback: { chunkTimeout: 30_000 }, + }) + expect(out).toBe(30_000) + }) + + test("falls back to provider value when prepared is non-finite (Infinity, NaN)", () => { + expect( + KiloLLM.resolveIdleMs({ options: { chunkTimeout: Number.POSITIVE_INFINITY }, fallback: { chunkTimeout: 5_000 } }), + ).toBe(5_000) + expect(KiloLLM.resolveIdleMs({ options: { chunkTimeout: Number.NaN }, fallback: { chunkTimeout: 5_000 } })).toBe( + 5_000, + ) + }) + + test("treats boolean false as a request to disable the watchdog", () => { + expect( + KiloLLM.resolveIdleMs({ options: { chunkTimeout: false }, fallback: { chunkTimeout: 30_000 } }), + ).toBeUndefined() + }) + + test("treats internal 0 as a request to disable the watchdog", () => { + expect(KiloLLM.resolveIdleMs({ options: { chunkTimeout: 0 }, fallback: { chunkTimeout: 30_000 } })).toBeUndefined() + }) + + test("provider fallback false also disables", () => { + expect(KiloLLM.resolveIdleMs({ options: {}, fallback: { chunkTimeout: false } })).toBeUndefined() + }) + + test("uses 60_000 default when nothing valid is configured", () => { + expect(KiloLLM.resolveIdleMs({ options: {} })).toBe(60_000) + }) + + test("uses 60_000 default when both prepared and fallback are invalid", () => { + expect( + KiloLLM.resolveIdleMs({ + options: { chunkTimeout: "x" }, + fallback: { chunkTimeout: -5 }, + }), + ).toBe(60_000) + }) +}) + +describe("kilocode.session.llm.watchdogStream", () => { + test("returns the stream unchanged when idle is undefined (disabled)", async () => { + const events: FullStreamPart[] = [ + part("stream-start", { warnings: [] as LanguageModelV2CallWarning[] }), + part("text-delta", { id: "t1", delta: "ok" }), + ] + const out = await run(Stream.runCollect(KiloLLM.watchdogStream(Stream.fromIterable(events), undefined))) + expect(out.length).toBe(2) + }) + + test("emits events and completes when the stream delivers them within the idle window", async () => { + const events: FullStreamPart[] = [ + part("stream-start", { warnings: [] as LanguageModelV2CallWarning[] }), + part("text-delta", { id: "t1", delta: "hi" }), + part("text-delta", { id: "t1", delta: "!" }), + ] + const out = await run(Stream.runCollect(KiloLLM.watchdogStream(Stream.fromIterable(events), 1_000))) + expect(out.length).toBe(3) + }) + + test("fails with ProviderError.ResponseStreamError when the stream stalls", async () => { + const slow = Stream.fromEffect( + Effect.flatMap(Effect.sleep("5 seconds"), () => Effect.succeed(part("text-delta", { id: "t1", delta: "x" }))), + ) + const wrapped = KiloLLM.watchdogStream(slow, 100) + const err = await run(Effect.flip(Stream.runCollect(wrapped))) + expect(err).toBeInstanceOf(ProviderError.ResponseStreamError) + }) + + test("every raw AI SDK event resets the idle timer", async () => { + // idle 200ms; emit text-delta at 0 and 60ms (a single 260ms pull would time out without reset). + const stream = fromSchedule( + [ + [0, part("text-delta", { id: "t1", delta: "a" })], + [60, part("text-delta", { id: "t1", delta: "b" })], + ], + 10, + ) + const out = await run(Stream.runCollect(KiloLLM.watchdogStream(stream, 200))) + expect(out.length).toBe(2) + }) + + test("pending local tool calls suspend the idle timeout until they settle", async () => { + // tool-call at t=0 (local). 250ms quiet gap then tool-result. A healthy AI + // SDK run also emits finish-step + finish right after the tool-result, so + // the watchdog sees another event within idleMs and resets. + const stream = fromSchedule( + [ + [0, part("tool-call", { toolCallId: "c1", toolName: "bash" })], + [250, part("tool-result", { toolCallId: "c1", toolName: "bash", output: "ok" })], + [260, part("finish-step", { finishReason: "tool-calls" })], + [270, part("finish", { finishReason: "stop" })], + ], + 10, + ) + const out = await run(Stream.runCollect(KiloLLM.watchdogStream(stream, 200))) + expect(out.length).toBe(4) + }) + + test("provider-executed tool calls do not suspend the watchdog", async () => { + const stream = fromSchedule( + [[0, part("tool-call", { toolCallId: "c1", toolName: "web", providerExecuted: true })]], + 400, + ) + const err = await run(Effect.flip(Stream.runCollect(KiloLLM.watchdogStream(stream, 200)))) + expect(err).toBeInstanceOf(ProviderError.ResponseStreamError) + }) + + test("parallel local tool calls remain suspended until the last settles", async () => { + const stream = fromSchedule( + [ + [0, part("tool-call", { toolCallId: "a", toolName: "bash" })], + [10, part("tool-call", { toolCallId: "b", toolName: "bash" })], + [200, part("tool-result", { toolCallId: "a", toolName: "bash", output: "x" })], + [350, part("tool-result", { toolCallId: "b", toolName: "bash", output: "y" })], + [360, part("finish-step", { finishReason: "tool-calls" })], + [370, part("finish", { finishReason: "stop" })], + ], + 10, + ) + const out = await run(Stream.runCollect(KiloLLM.watchdogStream(stream, 200))) + expect(out.length).toBe(6) + }) + + test("tool-error for a local tool id also releases the suspension", async () => { + const stream = fromSchedule( + [ + [0, part("tool-call", { toolCallId: "c1", toolName: "bash" })], + [200, part("tool-error", { toolCallId: "c1", toolName: "bash", error: new Error("nope") })], + [210, part("finish-step", { finishReason: "tool-calls" })], + [220, part("finish", { finishReason: "stop" })], + ], + 10, + ) + const out = await run(Stream.runCollect(KiloLLM.watchdogStream(stream, 200))) + expect(out.length).toBe(4) + }) + + test("aborts the underlying source on timeout so cleanup does not hang", async () => { + const ctrl = new AbortController() + let abortReason: unknown + let nextResolved = false + const source: AsyncIterable = { + [Symbol.asyncIterator]() { + let nextPromise: Promise> | undefined + let resolveNext: ((value: IteratorResult) => void) | undefined + ctrl.signal.addEventListener("abort", () => { + abortReason = ctrl.signal.reason + if (resolveNext) { + resolveNext({ done: true, value: undefined }) + nextResolved = true + } + }) + return { + next() { + nextPromise = new Promise((resolve) => { + resolveNext = resolve + }) + return nextPromise + }, + async return() { + if (nextPromise) await nextPromise + return { done: true, value: undefined } + }, + } + }, + } + const wrapped = KiloLLM.watchdogAsyncIterable(source, 100, ctrl) + const err = await run(Effect.flip(Stream.runCollect(Stream.fromAsyncIterable(wrapped, (e) => e as never)))) + expect(err).toBeInstanceOf(ProviderError.ResponseStreamError) + expect(nextResolved).toBe(true) + expect(abortReason).toBeInstanceOf(ProviderError.ResponseStreamError) + }) + + test("propagates upstream stream errors without false timeout", async () => { + const stream = Stream.fail(new Error("upstream broken")) + const err = await run(Effect.flip(Stream.runCollect(KiloLLM.watchdogStream(stream, 1_000)))) + expect((err as Error).message).toBe("upstream broken") + }) + + test("return() closes the source immediately without waiting on a stalled pull", async () => { + // Regression test: a hand-rolled async generator's `.return()` cannot + // preempt an in-flight internal `await` — it only takes effect once that + // await settles on its own, which never happens for a genuinely stalled + // source. `watchdogAsyncIterable` must instead expose a `return()` that + // runs immediately and forwards to the source's `return()` without + // waiting for the outstanding `next()` to resolve. + let sourceReturnCalled = false + let neverResolvingNextCalled = false + const source: AsyncIterable = { + [Symbol.asyncIterator]() { + return { + next() { + neverResolvingNextCalled = true + return new Promise>(() => { + // Never resolves — simulates a fully stalled source (e.g. a + // hung fetch response) whose pending pull is abandoned once + // the consumer decides to stop. + }) + }, + async return() { + sourceReturnCalled = true + return { done: true, value: undefined } + }, + } + }, + } + const wrapped = KiloLLM.watchdogAsyncIterable(source, 60_000) + const it = wrapped[Symbol.asyncIterator]() + const pending = it.next() + expect(neverResolvingNextCalled).toBe(true) + + const returned = await Promise.race([ + it.return!(), + new Promise((_, reject) => setTimeout(() => reject(new Error("return() hung")), 500)), + ]) + expect(returned).toMatchObject({ done: true }) + expect(sourceReturnCalled).toBe(true) + // The abandoned pull is left unresolved; only return() is asserted here. + void pending + }) +}) diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 170a160b46b..65f15c85ff0 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -1384,8 +1384,11 @@ export type ProviderConfig = { * Timeout in milliseconds to wait for response headers. Provider integrations may set defaults. Set to false to disable timeout. */ headerTimeout?: number | false - chunkTimeout?: number - [key: string]: unknown | string | boolean | number | false | number | false | number | undefined + /** + * Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted. Set to false to disable the idle watchdog. + */ + chunkTimeout?: number | false + [key: string]: unknown | string | boolean | number | false | number | false | number | false | undefined } models?: { [key: string]: { diff --git a/packages/sdk/openapi.json b/packages/sdk/openapi.json index e77dbffeb42..bfd86726770 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -26115,8 +26115,17 @@ "description": "Timeout in milliseconds to wait for response headers. Provider integrations may set defaults. Set to false to disable timeout." }, "chunkTimeout": { - "type": "integer", - "exclusiveMinimum": 0 + "anyOf": [ + { + "type": "integer", + "exclusiveMinimum": 0 + }, + { + "type": "boolean", + "enum": [false] + } + ], + "description": "Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted. Set to false to disable the idle watchdog." } }, "additionalProperties": {}