Add model stream idle-timeout handling

README.md

@@ -53,6 +53,12 @@ In headless mode, Aictrl automatically rejects all interactive permission reques
 ### CI/CD Integration
 Set `AICTRL_HEADLESS=true` in your environment to force headless behavior even in pseudo-TTYs.
 
+Model streams have a five-minute idle timeout by default. Every stream event resets
+the timer, so long-running responses that continue making progress are unaffected.
+Set `AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS` to a positive decimal integer of milliseconds to
+override the timeout, or `0` to disable it. Missing, empty, negative, fractional, and
+non-decimal, non-numeric, or unsafe integer values use the 300000 ms default.
+
 ## GitHub Integration
 
 Aictrl includes a specialized GitHub agent that can be installed into your repositories to automate PR reviews, issue triage, and code generation.

packages/cli/src/cli/cmd/run.errors.ts

@@ -16,6 +16,9 @@ export function classifySessionError(err: unknown): ClassifiedSessionError {
   if (status === 429) return { reason: "rate_limit", code: "429", message }
   if (status === 401 || status === 403) return { reason: "auth", code: String(status), message }
   if (name === "ProviderAuthError") return { reason: "auth", code: status ? String(status) : undefined, message }
+  if (name === "StreamIdleTimeoutError") {
+    return { reason: "timeout", code: "MODEL_STREAM_IDLE_TIMEOUT", message }
+  }
   if (name === "AbortError" || /timeout/i.test(message)) {
     return { reason: "timeout", code: status ? String(status) : undefined, message }
   }

packages/cli/src/flag/flag.ts

@@ -4,6 +4,7 @@ function truthy(key: string) {
 }
 
 export namespace Flag {
+  export const AICTRL_MODEL_STREAM_IDLE_TIMEOUT_DEFAULT = 5 * 60 * 1000
   export const AICTRL_GIT_BASH_PATH = process.env["AICTRL_GIT_BASH_PATH"]
   export const AICTRL_CONFIG = process.env["AICTRL_CONFIG"]
   export declare const AICTRL_CONFIG_DIR: string | undefined
@@ -20,6 +21,7 @@ export namespace Flag {
   export const AICTRL_FAKE_VCS = process.env["AICTRL_FAKE_VCS"]
   export declare const AICTRL_CLIENT: string
   export const AICTRL_ENABLE_QUESTION_TOOL = truthy("AICTRL_ENABLE_QUESTION_TOOL")
+  export declare const AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS: number
 
   // Experimental
   export const AICTRL_EXPERIMENTAL = truthy("AICTRL_EXPERIMENTAL")
@@ -43,6 +45,19 @@ export namespace Flag {
   }
 }
 
+// Dynamic getter for AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS.
+// Evaluated at access time so environment overrides take effect for each model stream.
+Object.defineProperty(Flag, "AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS", {
+  get() {
+    const value = process.env["AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS"]
+    if (value === undefined || value.trim() === "") return Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_DEFAULT
+    const parsed = /^\d+$/.test(value.trim()) ? Number(value) : Number.NaN
+    return Number.isSafeInteger(parsed) ? parsed : Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_DEFAULT
+  },
+  enumerable: true,
+  configurable: false,
+})
+
 // Dynamic getter for AICTRL_DISABLE_PROJECT_CONFIG
 // This must be evaluated at access time, not module load time,
 // because external tooling may set this env var at runtime

packages/cli/src/session/idle.ts

@@ -0,0 +1,44 @@
+import { MessageV2 } from "./message-v2"
+
+export namespace StreamIdle {
+  export function signal(input?: AbortSignal) {
+    const controller = new AbortController()
+    return {
+      controller,
+      signal: input ? AbortSignal.any([input, controller.signal]) : controller.signal,
+    }
+  }
+
+  export async function* timeout<T>(stream: AsyncIterable<T>, ms: number, abort: () => void) {
+    if (ms === 0) {
+      yield* stream
+      return
+    }
+
+    const iterator = stream[Symbol.asyncIterator]()
+    try {
+      while (true) {
+        const timer = Promise.withResolvers<never>()
+        const id = setTimeout(() => {
+          timer.reject(
+            new MessageV2.StreamIdleTimeoutError({
+              message: `Model stream produced no events for ${ms}ms`,
+              timeout: ms,
+            }),
+          )
+          abort()
+        }, ms)
+        const next = await Promise.race([iterator.next(), timer.promise]).finally(() => clearTimeout(id))
+        if (next.done) return
+        yield next.value
+      }
+    } finally {
+      // Do not await cleanup: an async generator queues return() behind an
+      // in-flight next(), which may never settle for the stalled stream we are
+      // escaping. The abort above gives cooperative providers a chance to close.
+      try {
+        iterator.return?.().catch(() => {})
+      } catch {}
+    }
+  }
+}

packages/cli/src/session/message-v2.ts

@@ -19,6 +19,13 @@ import type { Provider } from "@/provider/provider"
 export namespace MessageV2 {
   export const OutputLengthError = NamedError.create("MessageOutputLengthError", z.object({}))
   export const AbortedError = NamedError.create("MessageAbortedError", z.object({ message: z.string() }))
+  export const StreamIdleTimeoutError = NamedError.create(
+    "StreamIdleTimeoutError",
+    z.object({
+      message: z.string(),
+      timeout: z.number(),
+    }),
+  )
   export const StructuredOutputError = NamedError.create(
     "StructuredOutputError",
     z.object({
@@ -400,6 +407,7 @@ export namespace MessageV2 {
         NamedError.Unknown.Schema,
         OutputLengthError.Schema,
         AbortedError.Schema,
+        StreamIdleTimeoutError.Schema,
         StructuredOutputError.Schema,
         ContextOverflowError.Schema,
         APIError.Schema,
@@ -817,6 +825,8 @@ export namespace MessageV2 {
             cause: e,
           },
         ).toObject()
+      case MessageV2.StreamIdleTimeoutError.isInstance(e):
+        return e.toObject()
       case MessageV2.OutputLengthError.isInstance(e):
         return e
       case LoadAPIKeyError.isInstance(e):

packages/cli/src/session/processor.ts

@@ -13,8 +13,10 @@ import type { Provider } from "@/provider/provider"
 import { LLM } from "./llm"
 import { Config } from "@/config/config"
 import { SessionCompaction } from "./compaction"
+import { StreamIdle } from "./idle"
 import { PermissionNext } from "@/permission/next"
 import { Question } from "@/question"
+import { Flag } from "@/flag/flag"
 import { NamedError } from "@aictrl/util/error"
 
 export namespace SessionProcessor {
@@ -51,9 +53,17 @@ export namespace SessionProcessor {
           try {
             let currentText: MessageV2.TextPart | undefined
             let reasoningMap: Record<string, MessageV2.ReasoningPart> = {}
-            const stream = await LLM.stream(streamInput)
+            const idle = StreamIdle.signal(streamInput.abort)
+            const stream = await LLM.stream({
+              ...streamInput,
+              abort: idle.signal,
+            })
 
-            for await (const value of stream.fullStream) {
+            for await (const value of StreamIdle.timeout(
+              stream.fullStream,
+              Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS,
+              () => idle.controller.abort(),
+            )) {
               input.abort.throwIfAborted()
               switch (value.type) {
                 case "start":

packages/cli/test/cli/classify-session-error.test.ts

@@ -2,6 +2,19 @@ import { describe, expect, test } from "bun:test"
 import { classifySessionError } from "../../src/cli/cmd/run.errors"
 
 describe("classifySessionError (#63)", () => {
+  test("model stream idle timeout has a stable timeout code", () => {
+    expect(
+      classifySessionError({
+        name: "StreamIdleTimeoutError",
+        data: { message: "Model stream produced no events for 300000ms", timeout: 300000 },
+      }),
+    ).toEqual({
+      reason: "timeout",
+      code: "MODEL_STREAM_IDLE_TIMEOUT",
+      message: "Model stream produced no events for 300000ms",
+    })
+  })
+
   test("HTTP 429 → rate_limit", () => {
     const res = classifySessionError({ status: 429, message: "Rate limit exceeded" })
     expect(res.reason).toBe("rate_limit")

packages/cli/test/session/idle.test.ts

@@ -0,0 +1,147 @@
+import { describe, expect, test } from "bun:test"
+import { StreamIdle } from "../../src/session/idle"
+import { MessageV2 } from "../../src/session/message-v2"
+import { Flag } from "../../src/flag/flag"
+
+describe("model stream idle timeout", () => {
+  test("uses its own abort signal when the caller signal is undefined", () => {
+    const idle = StreamIdle.signal()
+
+    expect(idle.signal.aborted).toBe(false)
+    idle.controller.abort()
+    expect(idle.signal.aborted).toBe(true)
+  })
+
+  test("combines a supplied caller signal with its timeout controller", () => {
+    const caller = new AbortController()
+    const idle = StreamIdle.signal(caller.signal)
+
+    caller.abort()
+    expect(idle.signal.aborted).toBe(true)
+  })
+
+  test("fails and aborts a stream whose next event stalls", async () => {
+    const pending = Promise.withResolvers<IteratorResult<string>>()
+    let aborted = false
+    const stream = {
+      [Symbol.asyncIterator]() {
+        return {
+          next: () => pending.promise,
+        }
+      },
+    }
+
+    const result = StreamIdle.timeout(stream, 10, () => {
+      aborted = true
+    })
+    const error = await result.next().catch((error) => error)
+
+    expect(aborted).toBe(true)
+    expect(MessageV2.StreamIdleTimeoutError.isInstance(error)).toBe(true)
+    expect(error.data).toEqual({
+      message: "Model stream produced no events for 10ms",
+      timeout: 10,
+    })
+  })
+
+  test("calls return on the inner iterator after an idle timeout", async () => {
+    const pending = Promise.withResolvers<IteratorResult<string>>()
+    let called = false
+    const stream = {
+      [Symbol.asyncIterator]() {
+        return {
+          next: () => pending.promise,
+          async return() {
+            called = true
+            return { done: true as const, value: undefined }
+          },
+        }
+      },
+    }
+
+    await StreamIdle.timeout(stream, 10, () => {}).next().catch(() => {})
+    expect(called).toBe(true)
+  })
+
+  test("releases the inner iterator when the consumer stops early", async () => {
+    let released = false
+    const stream = {
+      [Symbol.asyncIterator]() {
+        return {
+          value: 0,
+          async next() {
+            return { done: false as const, value: ++this.value }
+          },
+          async return() {
+            released = true
+            return { done: true as const, value: undefined }
+          },
+        }
+      },
+    }
+
+    for await (const _ of StreamIdle.timeout(stream, 100, () => {})) break
+    expect(released).toBe(true)
+  })
+
+  test("resets after each event instead of limiting total stream duration", async () => {
+    async function* stream() {
+      yield 1
+      await Bun.sleep(8)
+      yield 2
+      await Bun.sleep(8)
+      yield 3
+    }
+
+    const values: number[] = []
+    for await (const value of StreamIdle.timeout(stream(), 20, () => {
+      throw new Error("active stream should not abort")
+    })) {
+      values.push(value)
+    }
+
+    expect(values).toEqual([1, 2, 3])
+  })
+
+  test("zero disables the timeout", async () => {
+    async function* stream() {
+      await Bun.sleep(15)
+      yield "done"
+    }
+
+    const values = []
+    for await (const value of StreamIdle.timeout(stream(), 0, () => {
+      throw new Error("disabled timeout should not abort")
+    })) {
+      values.push(value)
+    }
+
+    expect(values).toEqual(["done"])
+  })
+})
+
+describe("AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS", () => {
+  test("supports default, override, disable, and invalid fallback", () => {
+    const original = process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS
+
+    try {
+      delete process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(300_000)
+      process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = "1234"
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(1234)
+      process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = "0"
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(0)
+      process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = "invalid"
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(300_000)
+      process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = "1e3"
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(300_000)
+      process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = "0x10"
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(300_000)
+      process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = "9007199254740992"
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(300_000)
+    } finally {
+      if (original === undefined) delete process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS
+      else process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = original
+    }
+  })
+})