From c6215488e3ecd5ebbfb1afe1db788d8c55ae5a94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 15 Jul 2026 13:41:48 +0200 Subject: [PATCH 1/9] fix(cli): prevent stalled agent streams --- .changeset/chunk-idle-timeout-default.md | 5 +++++ packages/opencode/src/kilocode/session/llm.ts | 3 ++- packages/opencode/test/kilocode/session/llm.test.ts | 4 ++-- 3 files changed, 9 insertions(+), 3 deletions(-) create mode 100644 .changeset/chunk-idle-timeout-default.md diff --git a/.changeset/chunk-idle-timeout-default.md b/.changeset/chunk-idle-timeout-default.md new file mode 100644 index 00000000000..9ba420853b5 --- /dev/null +++ b/.changeset/chunk-idle-timeout-default.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": patch +--- + +Prevent stalled model streams from leaving agent and subagent sessions stuck indefinitely. Apply a 60,000 ms chunk idle timeout by default for all AI SDK model streams, with prepared request/model/agent overrides and provider-level fallbacks still winning as before. diff --git a/packages/opencode/src/kilocode/session/llm.ts b/packages/opencode/src/kilocode/session/llm.ts index c6627a28b9f..8def0213e59 100644 --- a/packages/opencode/src/kilocode/session/llm.ts +++ b/packages/opencode/src/kilocode/session/llm.ts @@ -7,6 +7,7 @@ import { KiloSessionOverflow } from "./overflow" const SAFETY = 2048 const MIN_OUTPUT = 1024 +const DEFAULT_CHUNK_IDLE_MS = 60_000 export namespace KiloLLM { // Stream failures and interruptions propagate while text deltas are collected. @@ -27,7 +28,7 @@ export namespace KiloLLM { ? input.options["chunkTimeout"] : typeof input.fallback?.["chunkTimeout"] === "number" ? input.fallback["chunkTimeout"] - : undefined + : DEFAULT_CHUNK_IDLE_MS if (!value) return {} input.log?.debug("chunk idle timeout configured", { chunkTimeout: value }) return { timeout: { chunkMs: value } } diff --git a/packages/opencode/test/kilocode/session/llm.test.ts b/packages/opencode/test/kilocode/session/llm.test.ts index dc6c3dfa29a..94d82233c87 100644 --- a/packages/opencode/test/kilocode/session/llm.test.ts +++ b/packages/opencode/test/kilocode/session/llm.test.ts @@ -31,8 +31,8 @@ describe("kilocode.session.llm.timeout", () => { expect(result).toEqual({ timeout: { chunkMs: 30_000 } }) }) - test("omits the timeout when it is not configured", () => { - expect(KiloLLM.timeout({ options: {} })).toEqual({}) + test("defaults the chunk idle timeout to 60_000 ms when no override is configured", () => { + expect(KiloLLM.timeout({ options: {} })).toEqual({ timeout: { chunkMs: 60_000 } }) }) }) From ad61520a29d80c3a37a8b4d55a62d7f76704b6d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 15 Jul 2026 19:56:24 +0200 Subject: [PATCH 2/9] docs(cli): clarify stream timeout scope --- .changeset/chunk-idle-timeout-default.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/chunk-idle-timeout-default.md b/.changeset/chunk-idle-timeout-default.md index 9ba420853b5..0aa0533303a 100644 --- a/.changeset/chunk-idle-timeout-default.md +++ b/.changeset/chunk-idle-timeout-default.md @@ -2,4 +2,4 @@ "@kilocode/cli": patch --- -Prevent stalled model streams from leaving agent and subagent sessions stuck indefinitely. Apply a 60,000 ms chunk idle timeout by default for all AI SDK model streams, with prepared request/model/agent overrides and provider-level fallbacks still winning as before. +Prevent stalled model streams from leaving agent and subagent sessions stuck indefinitely. Apply a 60,000 ms chunk idle timeout by default for normal agent model streams, with prepared request, model, and agent overrides and provider-level fallbacks still winning as before. From 4f89293a5da5b8352a503913099c54e258450447 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 15 Jul 2026 22:01:39 +0200 Subject: [PATCH 3/9] test(cli): stabilize global skill permission timing --- .../session-prompt-permission-refresh.test.ts | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/packages/opencode/test/kilocode/session-prompt-permission-refresh.test.ts b/packages/opencode/test/kilocode/session-prompt-permission-refresh.test.ts index 73bf22c0f45..3447d6b5f7b 100644 --- a/packages/opencode/test/kilocode/session-prompt-permission-refresh.test.ts +++ b/packages/opencode/test/kilocode/session-prompt-permission-refresh.test.ts @@ -1012,6 +1012,9 @@ it.live( const call = { command: "pwd", workdir: skill, description: "Run global skill resource" } yield* Effect.promise(() => fs.mkdir(skill, { recursive: true })) + yield* Effect.addFinalizer(() => + Effect.promise(() => rm(skill, { recursive: true, force: true }).catch(() => {})), + ) yield* llm.push(reply().tool("bash", call), reply().text("first complete").stop()) yield* prompt.prompt({ @@ -1028,6 +1031,7 @@ it.live( return list.find((item) => item.sessionID === chat.id) }), "global skill permission was never surfaced", + "15 seconds", ) expect(pending?.permission).toBe("external_directory") const always = (pending?.always ?? []) as string[] @@ -1040,7 +1044,9 @@ it.live( yield* permission.reply({ requestID: pending.id, reply: "always" }) expect( - Exit.isSuccess(yield* awaitWithTimeout(Fiber.await(first), "first global skill run did not finish")), + Exit.isSuccess( + yield* awaitWithTimeout(Fiber.await(first), "first global skill run did not finish", "15 seconds"), + ), ).toBe(true) yield* llm.push(reply().tool("bash", call), reply().text("second complete").stop()) @@ -1052,7 +1058,9 @@ it.live( }) const second = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkScoped) expect( - Exit.isSuccess(yield* awaitWithTimeout(Fiber.await(second), "trusted global skill prompted a second time")), + Exit.isSuccess( + yield* awaitWithTimeout(Fiber.await(second), "trusted global skill prompted a second time", "15 seconds"), + ), ).toBe(true) expect(yield* permission.list()).toEqual([]) }), @@ -1064,7 +1072,7 @@ it.live( }), }, ), - { timeout: 15_000 }, + { timeout: 30_000 }, ) it.live("active tool calls use permissions changed after model streaming starts", () => From 92999c3f1eca27543e5a7e2787546659317c198f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 16 Jul 2026 17:31:23 +0200 Subject: [PATCH 4/9] fix(cli): make stream timeout tool-aware --- .changeset/chunk-idle-timeout-default.md | 2 +- packages/core/src/v1/config/provider.ts | 6 +- packages/opencode/src/kilocode/session/llm.ts | 178 +++++- packages/opencode/src/session/llm.ts | 34 +- .../kilocode/session-stream-watchdog.test.ts | 519 ++++++++++++++++++ .../test/kilocode/session/llm.test.ts | 88 ++- .../session/session-stream-watchdog.test.ts | 240 ++++++++ packages/sdk/js/src/v2/gen/types.gen.ts | 7 +- packages/sdk/openapi.json | 13 +- 9 files changed, 1040 insertions(+), 47 deletions(-) create mode 100644 packages/opencode/test/kilocode/session-stream-watchdog.test.ts create mode 100644 packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts diff --git a/.changeset/chunk-idle-timeout-default.md b/.changeset/chunk-idle-timeout-default.md index 0aa0533303a..e779f74d4d0 100644 --- a/.changeset/chunk-idle-timeout-default.md +++ b/.changeset/chunk-idle-timeout-default.md @@ -2,4 +2,4 @@ "@kilocode/cli": patch --- -Prevent stalled model streams from leaving agent and subagent sessions stuck indefinitely. Apply a 60,000 ms chunk idle timeout by default for normal agent model streams, with prepared request, model, and agent overrides and provider-level fallbacks still winning as before. +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 8def0213e59..8a2f2ff87fd 100644 --- a/packages/opencode/src/kilocode/session/llm.ts +++ b/packages/opencode/src/kilocode/session/llm.ts @@ -1,7 +1,8 @@ -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" @@ -9,6 +10,8 @@ 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. export function text(stream: Stream.Stream) { @@ -18,20 +21,167 @@ 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"] - : DEFAULT_CHUNK_IDLE_MS - 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 watchIterable(source, idleMs, abort) + } + + async function* watchIterable( + source: AsyncIterable, + idleMs: number, + abort?: AbortController, + ): AsyncGenerator { + const local = new Set() + const iter = source[Symbol.asyncIterator]() + let suspended = false + try { + while (true) { + // 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) return + const part = value.value + trackPart(local, part) + yield part + suspended = local.size > 0 + } + } finally { + await safeClose(iter) + } + } + + 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..5ad60742f5d --- /dev/null +++ b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts @@ -0,0 +1,519 @@ +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), + ) + +const waitForFile = (file: string, label: string, duration = 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") + } + } + } +} + +const bashGate = (dir: string, ready: string, release: string) => + `touch ${JSON.stringify(ready)} && while [ ! -f ${JSON.stringify(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, config: (url) => ({ ...providerCfg(url), 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, + config: (url) => ({ + ...providerCfg(url), + 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 94d82233c87..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(result).toEqual({ timeout: { chunkMs: 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.timeout({ options: {} })).toEqual({ timeout: { chunkMs: 60_000 } }) + 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() + }) + + test("returns undefined when provider fallback is false", () => { + expect( + KiloLLM.resolveIdleMs({ + options: {}, + fallback: { chunkTimeout: false }, + }), + ).toBeUndefined() + }) + + 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..972c5f7de2c --- /dev/null +++ b/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts @@ -0,0 +1,240 @@ +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") + }) +}) diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 3b7f759f439..5d1eb2d741d 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 cf78c6653c9..4aa5736b01f 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -26088,8 +26088,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": {} From ca8950a26c9bb970513d1c80bb0289a3a965a715 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 16 Jul 2026 22:14:53 +0200 Subject: [PATCH 5/9] fix(cli): let the idle watchdog cancel a stalled pull immediately An async generator's return() cannot preempt an in-flight internal await; when suspended mid-await it only applies once that await settles on its own. For a genuinely stalled stream that await never settles, so interrupting a session mid-stream (e.g. aborting while a local tool call is pending) hung instead of cancelling. Replace the generator with a hand-rolled AsyncIterator whose return() runs immediately and forwards to the source's return() without waiting on any outstanding pull, matching how interruption already behaves for the unwrapped upstream iterator. Fixes CI failures in test/session/processor-effect.test.ts and test/session/prompt.test.ts that hung/timed out on this branch. --- packages/opencode/src/kilocode/session/llm.ts | 71 +++++++++++++------ .../session/session-stream-watchdog.test.ts | 42 +++++++++++ 2 files changed, 93 insertions(+), 20 deletions(-) diff --git a/packages/opencode/src/kilocode/session/llm.ts b/packages/opencode/src/kilocode/session/llm.ts index 8a2f2ff87fd..fc317270ef9 100644 --- a/packages/opencode/src/kilocode/session/llm.ts +++ b/packages/opencode/src/kilocode/session/llm.ts @@ -107,34 +107,65 @@ export namespace KiloLLM { abort?: AbortController, ): AsyncIterable { if (idleMs === undefined) return source - return watchIterable(source, idleMs, abort) + return { [Symbol.asyncIterator]: () => watchIterator(source, idleMs, abort) } } - async function* watchIterable( + /** + * 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, - ): AsyncGenerator { + ): AsyncIterator { const local = new Set() const iter = source[Symbol.asyncIterator]() let suspended = false - try { - while (true) { - // 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) return - const part = value.value - trackPart(local, part) - yield part - suspended = local.size > 0 - } - } finally { - await safeClose(iter) + 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 } + }, } } diff --git a/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts b/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts index 972c5f7de2c..6c3efcb4cae 100644 --- a/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts +++ b/packages/opencode/test/kilocode/session/session-stream-watchdog.test.ts @@ -237,4 +237,46 @@ describe("kilocode.session.llm.watchdogStream", () => { 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 + }) }) From 7e7a3d583c4c6b8d6452a8defe6d38b847c25590 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 16 Jul 2026 22:31:42 +0200 Subject: [PATCH 6/9] test(cli): give Windows more time for the watchdog integration bash gate git-bash on Windows CI runners spawns and writes the readiness marker file noticeably slower than the Unix shells this suite otherwise runs under, so tests A and C's 5s file-poll and 30s scenario timeout were too tight there and failed with 'readiness marker never appeared' even though the tool was already running. Double both on win32, matching the existing platform-aware timeout doubling in test/kilocode/background-process.test.ts. --- .../kilocode/session-stream-watchdog.test.ts | 33 ++++++++++++++----- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts index 5ad60742f5d..a3bc9fa2f24 100644 --- a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts +++ b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts @@ -257,7 +257,12 @@ const exists = (file: string) => .catch(() => false), ) -const waitForFile = (file: string, label: string, duration = 5_000) => +// Windows CI runners spawn git-bash noticeably slower than the Unix shells +// used elsewhere in this file, so the marker file these tests poll for can +// take longer than 5s to appear even when the tool is already reported as +// running. Matches the existing platform-aware timeout doubling in +// test/kilocode/background-process.test.ts. +const waitForFile = (file: string, label: string, duration = process.platform === "win32" ? 15_000 : 5_000) => pollWithTimeout( Effect.gen(function* () { const ok = yield* exists(file) @@ -349,7 +354,7 @@ describe("session stream watchdog integration", () => { 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) { @@ -358,7 +363,10 @@ describe("session stream watchdog integration", () => { }), { git: true, config: (url) => ({ ...providerCfg(url), permission: { bash: "allow" } }) }, ), - { timeout: 30_000 }, + // kilocode_change: doubled on Windows — git-bash spawns and writes the + // readiness marker noticeably slower there than the Unix shells used + // elsewhere in this file (see waitForFile above). + { timeout: process.platform === "win32" ? 60_000 : 30_000 }, ) it.live( @@ -398,9 +406,13 @@ describe("session stream watchdog integration", () => { gate.resolve(undefined) - const exit = yield* awaitWithTimeout(Fiber.await(fiber), "root foreground child loop did not finish", "15 seconds") + 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) { @@ -484,9 +496,13 @@ describe("session stream watchdog integration", () => { yield* touch(release) - const exit = yield* awaitWithTimeout(Fiber.await(fiber), "nested child bash loop did not finish", "15 seconds") + 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) { @@ -514,6 +530,7 @@ describe("session stream watchdog integration", () => { }), }, ), - { timeout: 30_000 }, + // kilocode_change: doubled on Windows, see the matching comment on test A. + { timeout: process.platform === "win32" ? 60_000 : 30_000 }, ) }) From 50a311cc3e54b32dda851d004efad74048b650a4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 16 Jul 2026 22:46:15 +0200 Subject: [PATCH 7/9] test(cli): use POSIX-style paths in the watchdog bash gate script path.join() yields backslash-separated paths on Windows. Embedded inside a double-quoted git-bash string, a literal backslash is an escape character, so the ready/release marker paths could resolve to the wrong file (or nothing) instead of erroring, making 'touch' and the '[ -f ... ]' poll silently miss each other. Normalize to forward slashes before interpolating into the script; git-bash/MSYS accept them natively on every platform this suite runs on. This is the actual root cause of the 'readiness marker never appeared' failures on Windows shards; the previous commit's timeout doubling was only masking symptoms. --- .../kilocode/session-stream-watchdog.test.ts | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts index a3bc9fa2f24..69ddf683ca2 100644 --- a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts +++ b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts @@ -258,10 +258,10 @@ const exists = (file: string) => ) // Windows CI runners spawn git-bash noticeably slower than the Unix shells -// used elsewhere in this file, so the marker file these tests poll for can -// take longer than 5s to appear even when the tool is already reported as -// running. Matches the existing platform-aware timeout doubling in -// test/kilocode/background-process.test.ts. +// used elsewhere in this file, so give the marker file these tests poll for +// extra headroom there even with a correctly resolved path (see `posixPath` +// below for the actual bug this was masking). Matches the existing +// platform-aware timeout doubling in test/kilocode/background-process.test.ts. const waitForFile = (file: string, label: string, duration = process.platform === "win32" ? 15_000 : 5_000) => pollWithTimeout( Effect.gen(function* () { @@ -313,8 +313,18 @@ const assertNotInterrupted = (parts: SessionV1.WithParts["parts"]) => { } } +// 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(ready)} && while [ ! -f ${JSON.stringify(release)} ]; do sleep 0.05; done && echo done` + `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( From c42e3c578daba3b2291a20b039f563f227770caf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 16 Jul 2026 23:02:44 +0200 Subject: [PATCH 8/9] test(cli): extend Windows margins further for the watchdog bash gate The production bash tool runs every command through a login shell (bash -l -c ..., src/shell/shell.ts) so ~/.bashrc/aliases behave like an interactive terminal. Git for Windows' login-shell startup rescans the full Windows PATH and is known to take several seconds on CI hardware, well past the previous 15s/60s Windows margins, before the script's own touch ever runs. Extend waitForFile to 30s and the two affected scenario timeouts to 90s on win32. --- .../kilocode/session-stream-watchdog.test.ts | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts index 69ddf683ca2..5ef236c67cb 100644 --- a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts +++ b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts @@ -257,12 +257,15 @@ const exists = (file: string) => .catch(() => false), ) -// Windows CI runners spawn git-bash noticeably slower than the Unix shells -// used elsewhere in this file, so give the marker file these tests poll for -// extra headroom there even with a correctly resolved path (see `posixPath` -// below for the actual bug this was masking). Matches the existing +// 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` and is well known to +// take several seconds on CI hardware — well beyond what a Unix login shell +// costs — before it even reaches the `touch` in `bashGate`. Give the marker +// file these tests poll for a generous margin there. Matches the existing // platform-aware timeout doubling in test/kilocode/background-process.test.ts. -const waitForFile = (file: string, label: string, duration = process.platform === "win32" ? 15_000 : 5_000) => +const waitForFile = (file: string, label: string, duration = process.platform === "win32" ? 30_000 : 5_000) => pollWithTimeout( Effect.gen(function* () { const ok = yield* exists(file) @@ -373,10 +376,9 @@ describe("session stream watchdog integration", () => { }), { git: true, config: (url) => ({ ...providerCfg(url), permission: { bash: "allow" } }) }, ), - // kilocode_change: doubled on Windows — git-bash spawns and writes the - // readiness marker noticeably slower there than the Unix shells used - // elsewhere in this file (see waitForFile above). - { timeout: process.platform === "win32" ? 60_000 : 30_000 }, + // kilocode_change: extended on Windows to cover the slow login-shell + // startup described on waitForFile above. + { timeout: process.platform === "win32" ? 90_000 : 30_000 }, ) it.live( @@ -540,7 +542,7 @@ describe("session stream watchdog integration", () => { }), }, ), - // kilocode_change: doubled on Windows, see the matching comment on test A. - { timeout: process.platform === "win32" ? 60_000 : 30_000 }, + // kilocode_change: extended on Windows, see the matching comment on test A. + { timeout: process.platform === "win32" ? 90_000 : 30_000 }, ) }) From c128eefdf835bb14387f23738fe65f6b500432ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 16 Jul 2026 23:28:08 +0200 Subject: [PATCH 9/9] test(cli): give the watchdog bash gate an explicit shell so it runs on Windows Root cause, finally isolated: without a config-level shell field, the bash tool defaultShell() falls back to cmd.exe on Windows (see packages/core/src/tool/bash.ts). cmd.exe cannot run bashGate POSIX syntax (touch, test -f, while/done), so touch failed instantly and silently and the readiness marker never appeared - no timeout was ever going to fix that, which is why the previous two commits margin increases did not help. Set shell to bash in tests A and C config so the bash tool resolves real git-bash via src/shell/shell.ts on Windows, and drop the speculative timeout inflation back to the original values plus a small, now-accurate margin for git-bash slower login-shell startup. --- .../kilocode/session-stream-watchdog.test.ts | 34 ++++++++++++------- 1 file changed, 22 insertions(+), 12 deletions(-) diff --git a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts index 5ef236c67cb..c222a1c71e7 100644 --- a/packages/opencode/test/kilocode/session-stream-watchdog.test.ts +++ b/packages/opencode/test/kilocode/session-stream-watchdog.test.ts @@ -260,12 +260,12 @@ const exists = (file: string) => // 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` and is well known to -// take several seconds on CI hardware — well beyond what a Unix login shell -// costs — before it even reaches the `touch` in `bashGate`. Give the marker -// file these tests poll for a generous margin there. Matches the existing -// platform-aware timeout doubling in test/kilocode/background-process.test.ts. -const waitForFile = (file: string, label: string, duration = process.platform === "win32" ? 30_000 : 5_000) => +// 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) @@ -374,11 +374,18 @@ describe("session stream watchdog integration", () => { assertNotInterrupted(msg.parts) } }), - { git: true, config: (url) => ({ ...providerCfg(url), permission: { bash: "allow" } }) }, + { + 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" } }), + }, ), - // kilocode_change: extended on Windows to cover the slow login-shell - // startup described on waitForFile above. - { timeout: process.platform === "win32" ? 90_000 : 30_000 }, + { timeout: 30_000 }, ) it.live( @@ -529,8 +536,12 @@ describe("session stream watchdog integration", () => { }), { 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: { @@ -542,7 +553,6 @@ describe("session stream watchdog integration", () => { }), }, ), - // kilocode_change: extended on Windows, see the matching comment on test A. - { timeout: process.platform === "win32" ? 90_000 : 30_000 }, + { timeout: 30_000 }, ) })