fix: add model stream idle timeout

EVENTS.md

@@ -170,8 +170,8 @@ Emitted immediately before `session_complete` when the session terminates abnorm
 }
 ```
 
-- `reason` (string, **required**) — one of `rate_limit`, `auth`, `timeout`, `oom`, `provider`, `interrupted`, `terminated`, `unknown`. `SIGINT` produces `interrupted`; `SIGTERM` produces `terminated`. Signals are not inferred to be timeouts.
-- `code` (string, optional) — provider HTTP status code, error code, or conventional signal-derived exit code (`130` for `SIGINT`, `143` for `SIGTERM`) when available.
+- `reason` (string, **required**) — one of `rate_limit`, `auth`, `timeout`, `oom`, `provider`, `interrupted`, `terminated`, `unknown`. A model stream idle timeout produces `timeout`; `SIGINT` produces `interrupted`; `SIGTERM` produces `terminated`. Signals are not inferred to be timeouts.
+- `code` (string, optional) — provider HTTP status code, error code, or conventional signal-derived exit code (`130` for `SIGINT`, `143` for `SIGTERM`) when available. A model stream idle timeout emits `MODEL_STREAM_IDLE_TIMEOUT` and persists a `StreamIdleTimeoutError` on the assistant message.
 - `message` (string, **required**) — human-readable error message.
 
 ## Message Events

README.md

@@ -34,36 +34,53 @@ aictrl run --model anthropic/claude-3-5-sonnet-latest "refactor the auth module"
 Aictrl is "headless first". When run in a non-TTY environment, it automatically switches to a mode optimized for automation.
 
 ### Stdin Piping
+
 You can pipe content directly into `aictrl`. This is useful for processing logs, code, or command output.
 
 ```bash
 cat logs.txt | aictrl run "summarize these errors"
 ```
 
 ### JSON Output
+
 For programmatic consumption, use `--format json` to get raw events.
 
 ```bash
 aictrl run --format json "review this PR" | jq '.type'
 ```
 
 ### Non-Interactive Execution
+
 In headless mode, Aictrl automatically rejects all interactive permission requests (like `question` or `plan_enter`), ensuring your pipelines never hang.
 
 ### CI/CD Integration
+
 Set `AICTRL_HEADLESS=true` in your environment to force headless behavior even in pseudo-TTYs.
 
+### Model Stream Idle Timeout
+
+Model stream idle timeouts are disabled by default. Set
+`AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS` to a decimal integer of milliseconds from 1
+through 2147483647 to enable one; for example, `300000` sets a five-minute timeout.
+`0` disables it. Missing, empty, negative, fractional, non-decimal, non-numeric, or
+unsupported values leave it disabled. The timer covers model stream setup and
+resets after every stream event, so responses that keep making progress are unaffected.
+Tool execution (local or provider-executed) uses a ceiling twelve times the
+configured model timeout, capped at 2147483647 ms (one hour for a five-minute timeout).
+
 ## GitHub Integration
 
 Aictrl includes a specialized GitHub agent that can be installed into your repositories to automate PR reviews, issue triage, and code generation.
 
 ### Setup
+
 ```bash
 # Install the GitHub agent in the current repo
 aictrl github install
 ```
 
 ### Features
+
 - **Auto-Push:** The agent can commit and push changes directly to your branches.
 - **PR Creation:** It can automatically open Pull Requests for its changes.
 - **Context Aware:** In GitHub Actions, it automatically fetches PR diffs, issue comments, and review history.
@@ -72,17 +89,21 @@ aictrl github install
 ## Developer Workflow
 
 ### PR Checkout
+
 Engineers can quickly checkout a PR and import the associated agent session:
 
 ```bash
 aictrl pr 123
 ```
+
 This command will:
+
 1. Fetch and checkout PR #123.
 2. Detect if an Aictrl session was used to generate the PR.
 3. Import that session locally so you can continue the conversation.
 
 ### MCP & Custom Tools
+
 Aictrl supports the [Model Context Protocol (MCP)](https://modelcontextprotocol.io).
 
 ```bash
@@ -101,11 +122,11 @@ Embed Aictrl directly into your TypeScript applications.
 import { createAictrlClient } from "@aictrl/sdk"
 
 const client = createAictrlClient({
-  baseUrl: "http://localhost:4096"
+  baseUrl: "http://localhost:4096",
 })
 
 const session = await client.session.create({
-  title: "My Automation Task"
+  title: "My Automation Task",
 })
 ```
 
@@ -122,4 +143,5 @@ aictrl acp
 Aictrl is a fork of the [OpenCode](https://opencode.ai) project and is licensed under the MIT License.
 
 ---
+
 [aictrl.dev](https://aictrl.dev/?utm_medium=referral&utm_source=github&utm_campaign=cli&utm_content=readme)

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 }
   }
@@ -37,13 +40,13 @@ export function classifySessionError(err: unknown): ClassifiedSessionError {
 }
 
 function extractMessage(err: unknown): string {
-  if (err instanceof Error) return err.message
-  if (typeof err === "string") return err
-  if (err && typeof err === "object" && "message" in err) return String((err as { message: unknown }).message)
   if (err && typeof err === "object" && "data" in err) {
     const data = (err as { data: unknown }).data
     if (data && typeof data === "object" && "message" in data) return String((data as { message: unknown }).message)
   }
+  if (err instanceof Error) return err.message
+  if (typeof err === "string") return err
+  if (err && typeof err === "object" && "message" in err) return String((err as { message: unknown }).message)
   return String(err)
 }
 

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 = 0
+  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,72 @@
+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 wait<T>(promise: Promise<T>, ms: number, abort: () => void): Promise<T> {
+    if (ms === 0) return promise
+    const timer = Promise.withResolvers<never>()
+    const id = setTimeout(() => {
+      timer.reject(error(ms, `Model stream setup produced no result for ${ms}ms`))
+      abort()
+    }, ms)
+    return Promise.race([promise, timer.promise]).finally(() => clearTimeout(id))
+  }
+
+  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

@@ -33,6 +33,13 @@ 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({
@@ -415,6 +422,7 @@ export namespace MessageV2 {
         NamedError.Unknown.Schema,
         OutputLengthError.Schema,
         AbortedError.Schema,
+        StreamIdleTimeoutError.Schema,
         StructuredOutputError.Schema,
         ContextOverflowError.Schema,
         APIError.Schema,
@@ -608,7 +616,8 @@ export namespace MessageV2 {
         if (
           msg.info.error &&
           !(
-            MessageV2.AbortedError.isInstance(msg.info.error) &&
+            (MessageV2.AbortedError.isInstance(msg.info.error) ||
+              MessageV2.StreamIdleTimeoutError.isInstance(msg.info.error)) &&
             msg.parts.some((part) => part.type !== "step-start" && part.type !== "reasoning")
           )
         ) {
@@ -836,6 +845,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,13 +13,16 @@ 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 { NamedError } from "@aictrl/util/error"
 import { ProviderTermination } from "@/provider/termination"
+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>>
@@ -91,9 +94,36 @@ export namespace SessionProcessor {
           try {
             let currentText: MessageV2.TextPart | undefined
             let reasoningMap: Record<string, MessageV2.ReasoningPart> = {}
-            const stream = await LLM.stream(streamInput)
+            const idleMs = Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS
+            const idle = StreamIdle.signal(streamInput.abort)
+            const stream = await StreamIdle.wait(
+              LLM.stream({
+                ...streamInput,
+                abort: idle.signal,
+              }),
+              idleMs,
+              () => idle.controller.abort(),
+            )
+            const runningTools = new Set<string>()
 
-            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

@@ -1,7 +1,30 @@
 import { describe, expect, test } from "bun:test"
 import { classifySessionError } from "../../src/cli/cmd/run.errors"
+import { MessageV2 } from "../../src/session/message-v2"
 
 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("live model stream timeout reports its human-readable message", () => {
+    const error = new MessageV2.StreamIdleTimeoutError({ message: "Model stream stalled", timeout: 25 })
+    expect(classifySessionError(error)).toEqual({
+      reason: "timeout",
+      code: "MODEL_STREAM_IDLE_TIMEOUT",
+      message: "Model stream stalled",
+    })
+  })
+
   test("HTTP 429 → rate_limit", () => {
     const res = classifySessionError({ status: 429, message: "Rate limit exceeded" })
     expect(res.reason).toBe("rate_limit")