fix(cli): recover stalled response streams

.changeset/recover-stalled-streams.md

@@ -0,0 +1,5 @@
+---
+"@kilocode/cli": patch
+---
+
+Prevent subagents from hanging on stalled model response streams or interactive suggestions.

packages/opencode/src/kilocode/provider/provider.ts

@@ -17,6 +17,15 @@ import { mapValues, omit, pickBy } from "remeda"
 /** Default timeout (ms) for provider HTTP requests (connection phase). */
 export const REQUEST_TIMEOUT_MS = 300_000 // 5 minutes
 
+/** Default maximum idle time between streamed response chunks. */
+export const STREAM_IDLE_TIMEOUT_MS = 120_000 // 2 minutes
+
+export function streamTimeout(input: { options: Record<string, unknown>; defaultMs?: number }): number | undefined {
+  const value = typeof input.options["chunkTimeout"] === "number" ? input.options["chunkTimeout"] : input.defaultMs
+  if (value === undefined || value <= 0) return
+  return value
+}
+
 // ---------------------------------------------------------------------------
 // Bundled providers
 // ---------------------------------------------------------------------------

packages/opencode/src/kilocode/session/llm.ts

@@ -3,6 +3,7 @@ import * as Stream from "effect/Stream"
 import type { LLMEvent } from "@opencode-ai/llm"
 import type { Logger } from "@opencode-ai/core/util/log"
 import type { Provider } from "@/provider/provider"
+import { streamTimeout } from "@/kilocode/provider/provider"
 import { KiloSessionOverflow } from "./overflow"
 
 const SAFETY = 2048
@@ -17,17 +18,10 @@ export namespace KiloLLM {
     )
   }
 
-  export function timeout(input: {
-    options: Record<string, unknown>
-    fallback?: Record<string, unknown>
-    log?: Pick<Logger, "debug">
-  }): { timeout?: { chunkMs: number } } {
-    const value =
-      typeof input.options["chunkTimeout"] === "number"
-        ? input.options["chunkTimeout"]
-        : typeof input.fallback?.["chunkTimeout"] === "number"
-          ? input.fallback["chunkTimeout"]
-          : undefined
+  export function timeout(input: { options: Record<string, unknown>; log?: Pick<Logger, "debug"> }): {
+    timeout?: { chunkMs: number }
+  } {
+    const value = streamTimeout(input)
     if (!value) return {}
     input.log?.debug("chunk idle timeout configured", { chunkTimeout: value })
     return { timeout: { chunkMs: value } }

packages/opencode/src/plugin/openai/ws-pool.ts

@@ -121,6 +121,7 @@ export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
         onConnectionInvalid: (error) => {
           log.warn("websocket invalidated", { key, error: error.message })
           entry.busy = false
+          entry.lastUsedAt = Date.now() // kilocode_change - port upstream #30586
           if (!entry.fallback) recordStreamFailure(entry)
           invalidate(entry)
           resolveFirstEvent(false)
@@ -179,6 +180,7 @@ export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
     const now = Date.now()
     for (const [key, entry] of pool) {
       if (entry.busy) continue
+      if (entry.fallback) continue // kilocode_change - port upstream #30586
       if (now - entry.lastUsedAt < idleTimeout) continue
       log.debug("websocket idle prune", { key })
       invalidate(entry)

packages/opencode/src/provider/provider.ts

@@ -39,6 +39,8 @@ import {
   patchKiloProviderPrivacy,
   kiloSmallModelPriority,
   buildTimeoutSignal,
+  STREAM_IDLE_TIMEOUT_MS,
+  streamTimeout,
 } from "@/kilocode/provider/provider"
 import * as ModelsRefresh from "@/kilocode/provider/models-refresh"
 // kilocode_change end
@@ -1677,7 +1679,12 @@ export const layer = Layer.effect(
         if (existing) return existing
 
         const customFetch = options["fetch"]
-        const chunkTimeout = options["chunkTimeout"]
+        // kilocode_change start - prevent indefinitely stalled response streams
+        const chunkTimeout = streamTimeout({
+          options,
+          defaultMs: STREAM_IDLE_TIMEOUT_MS,
+        })
+        // kilocode_change end
         const headerTimeout = options["headerTimeout"]
         delete options["chunkTimeout"]
         delete options["headerTimeout"]
@@ -1725,7 +1732,7 @@ export const layer = Layer.effect(
               timeout: false,
             }).finally(() => headerTimeoutCtl?.clear())
             timeout.clear()
-            if (!chunkAbortCtl) return res
+            if (chunkTimeout === undefined || !chunkAbortCtl) return res
             return wrapSSE(res, chunkTimeout, chunkAbortCtl)
           } catch (err) {
             timeout.clear()

packages/opencode/src/session/llm.ts

@@ -373,7 +373,7 @@ 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
+        ...KiloLLM.timeout({ options: prepared.params.options, log: l }), // kilocode_change
         headers: prepared.headers,
         maxRetries: input.retries ?? 0,
         messages: prepared.messages,

packages/opencode/src/tool/task.ts

@@ -248,7 +248,10 @@ export const TaskTool = Tool.define(
           variant, // kilocode_change
           agent: next.name,
           tools: {
-            question: false, // kilocode_change - subagents cannot prompt the user directly
+            // kilocode_change start - subagents cannot prompt the user directly
+            question: false,
+            suggest: false,
+            // kilocode_change end
             ...(canTodo ? {} : { todowrite: false }),
             ...(canTask ? {} : { task: false }),
             ...Object.fromEntries((cfg.experimental?.primary_tools ?? []).map((item) => [item, false])),

packages/opencode/test/kilocode/session-prompt-permission-refresh.test.ts

@@ -28,10 +28,12 @@ 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 { MessageV2 } from "../../src/session/message-v2"
 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 { SessionID } from "../../src/session/schema"
 import { Session } from "../../src/session/session"
 import { SessionStatus } from "../../src/session/status"
 import { SystemPrompt } from "../../src/session/system"
@@ -45,7 +47,7 @@ import { Ripgrep } from "../../src/file/ripgrep"
 import { ToolRegistry } from "../../src/tool/registry"
 import { Truncate } from "../../src/tool/truncate"
 import { provideTmpdirServer } from "../fixture/fixture"
-import { testEffect } from "../lib/effect"
+import { awaitWithTimeout, pollWithTimeout, testEffect } from "../lib/effect"
 import { reply, TestLLMServer } from "../lib/llm-server"
 
 void Log.init({ print: false })
@@ -226,7 +228,7 @@ const cfg = {
   },
 }
 
-function providerCfg(url: string) {
+function providerCfg(url: string, options: Record<string, unknown> = {}) {
   return {
     ...cfg,
     provider: {
@@ -235,6 +237,7 @@ function providerCfg(url: string) {
         ...cfg.provider.test,
         options: {
           ...cfg.provider.test.options,
+          ...options,
           baseURL: url,
         },
       },
@@ -290,3 +293,93 @@ it.live("active tool calls use permissions changed after model streaming starts"
     },
   ),
 )
+
+it.live(
+  "recovers a foreground task when the child stream stalls after bash completes",
+  () =>
+    provideTmpdirServer(
+      Effect.fnUntraced(function* ({ llm }) {
+        const prompt = yield* SessionPrompt.Service
+        const sessions = yield* Session.Service
+        const chat = yield* sessions.create({
+          title: "Pinned",
+          permission: [{ permission: "*", pattern: "*", action: "allow" }],
+        })
+
+        yield* llm.tool("task", {
+          description: "run child bash",
+          prompt: "Run one shell command and report its output.",
+          subagent_type: "general",
+        })
+        yield* llm.tool("bash", {
+          command: "printf child-bash-done",
+          description: "Print child completion marker",
+        })
+        yield* llm.push(reply().reason("Reviewing the completed command output.").hang())
+        yield* llm.text("child recovered")
+        yield* llm.text("parent completed")
+
+        yield* prompt.prompt({
+          sessionID: chat.id,
+          agent: "build",
+          noReply: true,
+          parts: [{ type: "text", text: "delegate the child bash command" }],
+        })
+
+        const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkScoped)
+        yield* awaitWithTimeout(llm.wait(3), "child did not open its post-bash model request", "20 seconds")
+
+        const task = yield* pollWithTimeout(
+          Effect.gen(function* () {
+            const msgs = yield* MessageV2.filterCompactedEffect(chat.id)
+            const part = msgs.flatMap((msg) => msg.parts).find((part) => part.type === "tool" && part.tool === "task")
+            if (part?.type !== "tool" || part.state.status !== "running") return
+            const child = part.state.metadata?.sessionId
+            if (typeof child !== "string") return
+            return { part, child: SessionID.make(child) }
+          }),
+          "parent task never entered the running state",
+        )
+
+        const output = yield* pollWithTimeout(
+          Effect.gen(function* () {
+            const msgs = yield* MessageV2.filterCompactedEffect(task.child)
+            const part = msgs.flatMap((msg) => msg.parts).find((part) => part.type === "tool" && part.tool === "bash")
+            if (part?.type !== "tool" || part.state.status !== "completed") return
+            return part.state.output
+          }),
+          "child bash tool never completed",
+        )
+
+        expect(output).toContain("child-bash-done")
+
+        const exit = yield* awaitWithTimeout(
+          Fiber.await(fiber),
+          "parent prompt remained stuck after the child stream stalled",
+          "10 seconds",
+        )
+        expect(Exit.isSuccess(exit)).toBe(true)
+        expect(yield* llm.calls).toBe(5)
+
+        const status = yield* SessionStatus.Service
+        expect((yield* status.get(chat.id)).type).toBe("idle")
+        expect((yield* status.get(task.child)).type).toBe("idle")
+
+        const parent = (yield* MessageV2.filterCompactedEffect(chat.id))
+          .flatMap((msg) => msg.parts)
+          .find((part) => part.type === "tool" && part.tool === "task")
+        expect(parent?.type === "tool" ? parent.state.status : undefined).toBe("completed")
+        expect(parent?.type === "tool" && parent.state.status === "completed" ? parent.state.output : "").toContain(
+          "child recovered",
+        )
+      }),
+      {
+        git: true,
+        config: (url) => ({
+          ...providerCfg(url, { chunkTimeout: 500 }),
+          permission: { task: "allow", bash: "allow" },
+        }),
+      },
+    ),
+  30_000,
+)

packages/opencode/test/kilocode/session/llm.test.ts

@@ -1,38 +1,46 @@
 import { describe, expect, test } from "bun:test"
 import { Effect, Stream } from "effect"
 import { LLMEvent } from "@opencode-ai/llm"
+import { STREAM_IDLE_TIMEOUT_MS, streamTimeout } from "@/kilocode/provider/provider"
 import { KiloLLM } from "@/kilocode/session/llm"
 
 describe("kilocode.session.llm.timeout", () => {
-  test("uses prepared options before the provider fallback", () => {
+  test("uses the prepared request timeout", () => {
     const result = KiloLLM.timeout({
       options: { chunkTimeout: 15_000 },
-      fallback: { chunkTimeout: 30_000 },
     })
 
     expect(result).toEqual({ timeout: { chunkMs: 15_000 } })
   })
 
-  test("uses the provider fallback when prepared options omit the timeout", () => {
+  test("ignores invalid prepared values", () => {
     const result = KiloLLM.timeout({
-      options: {},
-      fallback: { chunkTimeout: 30_000 },
+      options: { chunkTimeout: "15_000" },
     })
 
-    expect(result).toEqual({ timeout: { chunkMs: 30_000 } })
+    expect(result).toEqual({})
   })
 
-  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 },
-    })
+  test("omits the AI SDK timeout when it is not configured", () => {
+    expect(KiloLLM.timeout({ options: {} })).toEqual({})
+  })
 
-    expect(result).toEqual({ timeout: { chunkMs: 30_000 } })
+  test("allows the AI SDK timeout to be disabled explicitly", () => {
+    expect(KiloLLM.timeout({ options: { chunkTimeout: 0 } })).toEqual({})
   })
+})
 
-  test("omits the timeout when it is not configured", () => {
-    expect(KiloLLM.timeout({ options: {} })).toEqual({})
+describe("kilocode.provider.streamTimeout", () => {
+  test("uses the default stream inactivity timeout", () => {
+    expect(streamTimeout({ options: {}, defaultMs: STREAM_IDLE_TIMEOUT_MS })).toBe(120_000)
+  })
+
+  test("prefers an explicit timeout over the default", () => {
+    expect(streamTimeout({ options: { chunkTimeout: 30_000 }, defaultMs: STREAM_IDLE_TIMEOUT_MS })).toBe(30_000)
+  })
+
+  test("allows the default timeout to be disabled", () => {
+    expect(streamTimeout({ options: { chunkTimeout: 0 }, defaultMs: STREAM_IDLE_TIMEOUT_MS })).toBeUndefined()
   })
 })
 

packages/opencode/test/plugin/openai-ws.test.ts

@@ -210,7 +210,7 @@ describe("plugin.openai.ws-pool", () => {
     fetch.close()
   })
 
-  test("prunes HTTP fallback after its idle timeout", async () => {
+  test("keeps HTTP fallback active after its idle timeout", async () => { // kilocode_change - port upstream #30586
     let websocketAttempts = 0
     await using server = await createRejectingWebSocketServer(() => websocketAttempts++)
     const fetch = OpenAIWebSocketPool.createWebSocketFetch({
@@ -226,7 +226,7 @@ describe("plugin.openai.ws-pool", () => {
     const second = await fetch(server.url, streamRequest())
 
     expect(await second.text()).toBe("http")
-    expect(websocketAttempts).toBe(2)
+    expect(websocketAttempts).toBe(1) // kilocode_change - port upstream #30586
     expect(server.httpRequests).toHaveLength(2)
     fetch.close()
   })
@@ -392,28 +392,37 @@ describe("plugin.openai.ws-pool", () => {
     fetch.close()
   })
 
+  // kilocode_change start - port upstream #33387
   test("retries failed websocket streams before using HTTP fallback", async () => {
+    const attempts: Array<(socket: WebSocket) => void> = []
     await using server = await createWebSocketServer((socket) => {
       socket.once("message", () => {
         socket.send(JSON.stringify({ type: "response.output_text.delta", delta: "started" }))
+        attempts.shift()?.(socket)
       })
     })
     const fetch = OpenAIWebSocketPool.createWebSocketFetch({
       url: server.url,
-      idleTimeout: 20,
       streamRetries: 1,
     })
 
+    const firstAttempt = new Promise<WebSocket>((resolve) => attempts.push(resolve))
     const first = await fetch(server.url, streamRequest())
-    expect((await readTextError(first.text())).message).toContain("idle timeout waiting for websocket")
+    const firstSocket = await firstAttempt
+    firstSocket.terminate()
+    expect((await readTextError(first.text())).message).toContain("WebSocket closed before response.completed")
+    const secondAttempt = new Promise<WebSocket>((resolve) => attempts.push(resolve))
     const second = await fetch(server.url, streamRequest())
-    expect((await readTextError(second.text())).message).toContain("idle timeout waiting for websocket")
+    const secondSocket = await secondAttempt
+    secondSocket.terminate()
+    expect((await readTextError(second.text())).message).toContain("WebSocket closed before response.completed")
     const third = await fetch(server.url, streamRequest())
 
     expect(await third.text()).toBe("http")
     expect(server.httpRequests).toHaveLength(1)
     fetch.close()
   })
+  // kilocode_change end
 
   test("resets websocket stream failures after a completed response", async () => {
     let connections = 0

packages/opencode/test/tool/task.test.ts

@@ -482,7 +482,10 @@ describe("tool.task", () => {
         )
         // kilocode_change end
         expect(seen?.tools).toEqual({
-          question: false, // kilocode_change - subagents cannot prompt the user directly
+          // kilocode_change start - subagents cannot prompt the user directly
+          question: false,
+          suggest: false,
+          // kilocode_change end
           todowrite: false,
           task: false, // kilocode_change - Kilo disallows nested subagents
           bash: false,