fix: add model stream idle timeout

README.md

@@ -53,6 +53,15 @@ 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 Stream Idle Timeout
+
+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. Local
+tool execution uses a ceiling twelve times the configured model timeout (one hour by default).
+Set `AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS` to a decimal integer of milliseconds through
+2147483647 to override the timeout, or `0` to disable it. Missing, empty, negative,
+fractional, non-decimal, non-numeric, or unsupported 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

@@ -24,6 +24,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,8 @@ function truthy(key: string) {
 }
 
 export namespace Flag {
+  export const AICTRL_MODEL_STREAM_IDLE_TIMEOUT_DEFAULT = 5 * 60 * 1000
+  export const AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MAX = 2_147_483_647
   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 +22,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 +46,21 @@ 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_MAX
+      ? 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,62 @@
+import { MessageV2 } from "./message-v2"
+
+export namespace StreamIdle {
+  function error(ms: number, message: string) {
+    return new MessageV2.StreamIdleTimeoutError({
+      message,
+      timeout: ms,
+    })
+  }
+
+  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,
+    updateSuspended: (value: T) => boolean = () => false,
+    suspendedTimeout = ms,
+  ) {
+    if (ms === 0) {
+      yield* stream
+      return
+    }
+
+    const iterator = stream[Symbol.asyncIterator]()
+    let suspended = false
+    try {
+      while (true) {
+        const timer = Promise.withResolvers<never>()
+        const appliedTimeout = suspended ? suspendedTimeout : ms
+        const id = setTimeout(() => {
+          timer.reject(
+            error(
+              appliedTimeout,
+              suspended
+                ? `Tool execution produced no result for ${appliedTimeout}ms`
+                : `Model stream produced no events for ${appliedTimeout}ms`,
+            ),
+          )
+          abort()
+        }, appliedTimeout)
+        const next = await Promise.race([iterator.next(), timer.promise]).finally(() => clearTimeout(id))
+        if (next.done) return
+        suspended = updateSuspended(next.value)
+        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,
@@ -818,6 +826,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,12 +13,15 @@ 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 {
   const DOOM_LOOP_THRESHOLD = 3
+  const LOCAL_TOOL_TIMEOUT_MULTIPLIER = 12
   const log = Log.create({ service: "session.processor" })
 
   export type Info = Awaited<ReturnType<typeof create>>
@@ -51,9 +54,32 @@ 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,
+            })
+            const runningTools = new Set<string>()
+            const idleMs = Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS
 
-            for await (const value of stream.fullStream) {
+            for await (const value of StreamIdle.timeout(
+              stream.fullStream,
+              idleMs,
+              () => idle.controller.abort(),
+              (value) => {
+                if (
+                  value.type === "tool-call" &&
+                  (value.providerExecuted || typeof streamInput.tools?.[value.toolName]?.execute === "function")
+                ) {
+                  runningTools.add(value.toolCallId)
+                }
+                if (value.type === "tool-result" || value.type === "tool-error") {
+                  runningTools.delete(value.toolCallId)
+                }
+                return runningTools.size > 0
+              },
+              Math.min(idleMs * LOCAL_TOOL_TIMEOUT_MULTIPLIER, Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MAX),
+            )) {
               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/flag/flag.test.ts

@@ -0,0 +1,32 @@
+import { describe, expect, test } from "bun:test"
+import { Flag } from "../../src/flag/flag"
+
+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 = "2147483647"
+      expect(Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS).toBe(2_147_483_647)
+      process.env.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS = "2147483648"
+      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
+    }
+  })
+})