Add model stream idle-timeout handling#99
Closed
byapparov wants to merge 3 commits into
Closed
Conversation
![aictrl-dev[bot]](https://avatars.githubusercontent.com/in/2817739?s=60&v=4){kind=small}
| return | ||
| } | ||
|
|
||
| const iterator = stream[Symbol.asyncIterator]() |
There was a problem hiding this comment.
🟠 Idle wrapper never releases inner iterator on exit.
🤖 Fix with your agent
Fix this code review finding (aictrl-dev/cli PR #99, packages/cli/src/session/idle.ts:10-25):
Problem: Idle wrapper never releases inner iterator on exit
Detail: `StreamIdle.timeout` pulls `stream[Symbol.asyncIterator]()` and drives it manually with `iterator.next()`, but the generator has no `try/finally` that calls `iterator.return()` when it unwinds via the idle-timeout rejection, a propagated provider error, or a consumer `break`/`return`. The only cleanup hook is the `abort()` callback, which fires solely on the timeout path; every non-timeout early exit (and any timeout where the provider SDK does not synchronously settle the in-flight `next()` on abort) leaves the underlying HTTP socket / SSE reader and the still-pending `iterator.next()` promise dangling. This violates the async-iterator protocol that wrappers must forward `.return()` to the inner iterator, and over many sessions can leak file descriptors / sockets (slow-burn DoS). The included `idle.test.ts` does not catch it because its mock iterator exposes no `return()` method.
Suggested fix: Wrap the `while` loop in `try { ... } finally { try { await iterator.return?.() } catch {} }` so the inner iterator is released on every exit path (timeout, propagated error, early break, normal completion). Also consider awaiting `iterator.return?.()` inside the `setTimeout` handler right after `abort()` so the cleanup happens immediately on timeout rather than relying on the generator's outer finally.
Implement the fix on the PR head branch and add a regression test that fails before the fix and passes after.
Why this matters
StreamIdle.timeout pulls stream[Symbol.asyncIterator]() and drives it manually with iterator.next(), but the generator has no try/finally that calls iterator.return() when it unwinds via the idle-timeout rejection, a propagated provider error, or a consumer break/return. The only cleanup hook is the abort() callback, which fires solely on the timeout path; every non-timeout early exit (and any timeout where the provider SDK does not synchronously settle the in-flight next() on abort) leaves the underlying HTTP socket / SSE reader and the still-pending iterator.next() promise dangling. This violates the async-iterator protocol that wrappers must forward .return() to the inner iterator, and over many sessions can leak file descriptors / sockets (slow-burn DoS). The included idle.test.ts does not catch it because its mock iterator exposes no return() method.
if (ms === 0) {
yield* stream
return
}
const iterator = stream[Symbol.asyncIterator]()
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
}| const controller = new AbortController() | ||
| const stream = await LLM.stream({ | ||
| ...streamInput, | ||
| abort: AbortSignal.any([streamInput.abort, controller.signal]), |
There was a problem hiding this comment.
🟠 AbortSignal.any() throws if streamInput.abort is undefined.
🤖 Fix with your agent
Fix this code review finding (aictrl-dev/cli PR #99, packages/cli/src/session/processor.ts:59):
Problem: AbortSignal.any() throws if streamInput.abort is undefined
Detail: The new call `AbortSignal.any([streamInput.abort, controller.signal])` throws a synchronous `TypeError` when any element is not an `AbortSignal`. The prior `LLM.stream(streamInput)` tolerated an undefined `abort` field; this change turns a previously working session into a hard crash for any caller of `SessionProcessor.create` whose input lacks `abort`. The processor-idle integration test only exercises one caller (`SessionPrompt.prompt`) where `abort` happens to be defined, so the regression is not caught by tests. No null-guard is present.
Suggested fix: Filter undefined signals before combining: `const signals = [streamInput.abort, controller.signal].filter((s): s is AbortSignal => !!s); const abort = signals.length > 1 ? AbortSignal.any(signals) : signals[0];` then pass `abort`. Alternatively guard explicitly: `abort: streamInput.abort ? AbortSignal.any([streamInput.abort, controller.signal]) : controller.signal`.
Implement the fix on the PR head branch and add a regression test that fails before the fix and passes after.
Why this matters
The new call AbortSignal.any([streamInput.abort, controller.signal]) throws a synchronous TypeError when any element is not an AbortSignal. The prior LLM.stream(streamInput) tolerated an undefined abort field; this change turns a previously working session into a hard crash for any caller of SessionProcessor.create whose input lacks abort. The processor-idle integration test only exercises one caller (SessionPrompt.prompt) where abort happens to be defined, so the regression is not caught by tests. No null-guard is present.
const controller = new AbortController()
const stream = await LLM.stream({
...streamInput,
abort: AbortSignal.any([streamInput.abort, controller.signal]),
})| 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.MODEL_STREAM_IDLE_TIMEOUT_DEFAULT |
There was a problem hiding this comment.
🟡 Idle-timeout env parser too lenient (hex/scientific, no max).
🤖 Fix with your agent
Fix this code review finding (aictrl-dev/cli PR #99, packages/cli/src/flag/flag.ts:51-56):
Problem: Idle-timeout env parser too lenient (hex/scientific, no max)
Detail: The getter uses `Number(value)` and accepts any non-negative integer. This admits hex (`0xFF`) and scientific (`1e9`) notation, imposes no upper bound (a value like `9007199254740992` effectively disables the safety net this PR introduces), and silently swallows invalid values back to the default with no diagnostic. The behavior is operator-configured rather than attacker-controlled, so this is a robustness/usability nit rather than a security issue, but `parseInt(value, 10)` plus a sane max (e.g. 24h) and a stderr warning on invalid input would be friendlier.
Suggested fix: Switch to `Number.parseInt(value, 10)` to reject hex/scientific forms, clamp to a maximum (e.g. `Math.min(parsed, 24 * 60 * 60 * 1000)`), and optionally warn on stderr when the env value is non-empty but invalid so misconfigurations are visible.
Implement the fix on the PR head branch and add a regression test that fails before the fix and passes after.
Why this matters
The getter uses Number(value) and accepts any non-negative integer. This admits hex (0xFF) and scientific (1e9) notation, imposes no upper bound (a value like 9007199254740992 effectively disables the safety net this PR introduces), and silently swallows invalid values back to the default with no diagnostic. The behavior is operator-configured rather than attacker-controlled, so this is a robustness/usability nit rather than a security issue, but parseInt(value, 10) plus a sane max (e.g. 24h) and a stderr warning on invalid input would be friendlier.
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.MODEL_STREAM_IDLE_TIMEOUT_DEFAULT
const parsed = Number(value)
return Number.isInteger(parsed) && parsed >= 0 ? parsed : Flag.MODEL_STREAM_IDLE_TIMEOUT_DEFAULT
},
enumerable: true,
configurable: false,
})| import { Question } from "@/question" | ||
| import { NamedError } from "@aictrl/util/error" | ||
| import { StreamIdle } from "./idle" | ||
| import { Flag } from "@/flag/flag" |
There was a problem hiding this comment.
🟡 New imports break the file's import grouping.
Suggested change
| import { Flag } from "@/flag/flag" | |
| Move `import { StreamIdle } from \"./idle\"` next to `import { SessionCompaction } from \"./compaction\"`, and `import { Flag } from \"@/flag/flag\"` next to `import { PermissionNext } from \"@/permission/next\"`. |
🤖 Fix with your agent
Fix this code review finding (aictrl-dev/cli PR #99, packages/cli/src/session/processor.ts:20-21):
Problem: New imports break the file's import grouping
Detail: The existing import block in `processor.ts` groups by source kind: relative (`./compaction`), then `@/` aliases (`@/permission/next`, `@/question`), then external (`@aictrl/util/error`). The new `import { StreamIdle } from \"./idle\"` (relative) and `import { Flag } from \"@/flag/flag\"` (alias) are appended after `@aictrl/util/error`, violating that grouping.
Suggested fix: Move `import { StreamIdle } from \"./idle\"` next to `import { SessionCompaction } from \"./compaction\"`, and `import { Flag } from \"@/flag/flag\"` next to `import { PermissionNext } from \"@/permission/next\"`.
Implement the fix on the PR head branch and add a regression test that fails before the fix and passes after.
Why this matters
The existing import block in processor.ts groups by source kind: relative (./compaction), then @/ aliases (@/permission/next, @/question), then external (@aictrl/util/error). The new import { StreamIdle } from \"./idle\" (relative) and import { Flag } from \"@/flag/flag\" (alias) are appended after @aictrl/util/error, violating that grouping.
import { SessionCompaction } from "./compaction"
import { PermissionNext } from "@/permission/next"
import { Question } from "@/question"
import { NamedError } from "@aictrl/util/error"
import { StreamIdle } from "./idle"
import { Flag } from "@/flag/flag"| } | ||
|
|
||
| export namespace Flag { | ||
| export const MODEL_STREAM_IDLE_TIMEOUT_DEFAULT = 5 * 60 * 1000 |
There was a problem hiding this comment.
⚪ Default const breaks AICTRL_ naming convention.
🤖 Fix with your agent
Fix this code review finding (aictrl-dev/cli PR #99, packages/cli/src/flag/flag.ts:7):
Problem: Default const breaks AICTRL_ naming convention
Detail: Every other const inside the `Flag` namespace is named after an env var with the `AICTRL_` prefix (`AICTRL_GIT_BASH_PATH`, `AICTRL_ENABLE_QUESTION_TOOL`, etc.). `MODEL_STREAM_IDLE_TIMEOUT_DEFAULT` is the only non-`AICTRL_`-prefixed member and is interleaved with the env-var consts, making the namespace look inconsistent. The omission may be intentional (it is a default, not an env var), but the placement is noisy.
Suggested fix: Either keep the name but move the declaration adjacent to the `Object.defineProperty` getter that consumes it (so it sits with its only user), or rename to align with the file's convention, e.g. `AICTRL_MODEL_STREAM_IDLE_TIMEOUT_DEFAULT`.
Implement the fix on the PR head branch and add a regression test that fails before the fix and passes after.
Why this matters
Every other const inside the Flag namespace is named after an env var with the AICTRL_ prefix (AICTRL_GIT_BASH_PATH, AICTRL_ENABLE_QUESTION_TOOL, etc.). MODEL_STREAM_IDLE_TIMEOUT_DEFAULT is the only non-AICTRL_-prefixed member and is interleaved with the env-var consts, making the namespace look inconsistent. The omission may be intentional (it is a default, not an env var), but the placement is noisy.
}
export namespace Flag {
export const MODEL_STREAM_IDLE_TIMEOUT_DEFAULT = 5 * 60 * 1000
export const AICTRL_GIT_BASH_PATH = process.env["AICTRL_GIT_BASH_PATH"]| } | ||
| } | ||
|
|
||
| Object.defineProperty(Flag, "AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS", { |
There was a problem hiding this comment.
⚪ Lazy flag getter missing rationale comment its sibling has.
🤖 Fix with your agent
Fix this code review finding (aictrl-dev/cli PR #99, packages/cli/src/flag/flag.ts:48-56):
Problem: Lazy flag getter missing rationale comment its sibling has
Detail: The pre-existing `AICTRL_DISABLE_PROJECT_CONFIG` getter is preceded by a 3-line comment explaining it is defined via `Object.defineProperty` so the env var is read at access time rather than module load. The new `AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS` getter uses the same pattern (and relies on the same lazy-read semantics — the test mutates `process.env` at runtime and re-reads `Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS`), but omits the rationale comment, breaking the local convention for documenting this pattern.
Suggested fix: Add a short comment mirroring the sibling, e.g. `// Dynamic getter for AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS. Evaluated at access time, not module load, so env overrides and tests that mutate process.env take effect.`
Implement the fix on the PR head branch and add a regression test that fails before the fix and passes after.
Why this matters
The pre-existing AICTRL_DISABLE_PROJECT_CONFIG getter is preceded by a 3-line comment explaining it is defined via Object.defineProperty so the env var is read at access time rather than module load. The new AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS getter uses the same pattern (and relies on the same lazy-read semantics — the test mutates process.env at runtime and re-reads Flag.AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS), but omits the rationale comment, breaking the local convention for documenting this pattern.
// 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
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.MODEL_STREAM_IDLE_TIMEOUT_DEFAULT
const parsed = Number(value)
return Number.isInteger(parsed) && parsed >= 0 ? parsed : Flag.MODEL_STREAM_IDLE_TIMEOUT_DEFAULT
},
enumerable: true,
configurable: false,
})
Code reviewVerdict: Address the major findings before merging. · 🔴 0 · 🟠 2 · 🟡 2 · ⚪ 2 · 0/6 resolved
🤖 Fix all 6 open findings with your agent📋 Out-of-diff findings (6)
Reviewed 9 files · 0 inline · view all 6 findings ↗ aictrl · AI code review for fast-moving teams · aictrl.dev |
Contributor
Author
Review findings addressed at
|
| Finding | Verdict / action | Resolution | Verification |
|---|---|---|---|
| Default constant naming | TRUE · FIX | Renamed to AICTRL_MODEL_STREAM_IDLE_TIMEOUT_DEFAULT to match the Flag namespace convention. |
bun run typecheck — PASS |
| Lazy getter rationale | TRUE · FIX | Added the access-time evaluation rationale immediately above the property definition. | Diff inspection + typecheck — PASS |
| Environment parser robustness | TRUE in part · FIX | Kept explicit 0 disable, but now accepts only safe non-negative decimal integers. Hex, scientific, fractional, negative, unsafe, and non-numeric inputs use the documented default. I did not use parseInt: parseInt("1e9", 10) returns 1 and parseInt("0xFF", 10) returns 0, which would silently reinterpret input and can accidentally disable the guard. I also did not impose an undocumented 24h cap. |
Focused flag tests cover 0, decimal override, hex, scientific, unsafe integer, and invalid fallback — PASS |
| Inner iterator cleanup | TRUE · FIX | The wrapper now forwards return() in finally on timeout, provider failure, and consumer exit. Cleanup is deliberately non-blocking because awaiting an async-generator return() queued behind a permanently pending next() would recreate the hang. |
Behavioral tests assert cleanup on timeout and early consumer break — PASS |
| Import grouping | TRUE · FIX | Moved the relative and alias imports back into their existing groups. | git diff --check + typecheck — PASS |
| Undefined caller abort | FALSE contract claim · defensive FIX | LLM.StreamInput.abort is required by TypeScript, so supported callers cannot omit it. StreamIdle.signal() nevertheless now falls back to its own controller at runtime and combines a supplied caller signal when present. |
Behavioral tests cover undefined and supplied caller signals — PASS |
Verification:
bun test test/session/idle.test.ts test/session/processor-idle.test.ts test/cli/classify-session-error.test.ts test/cli/exception-handler.test.ts— 23 passed, 0 failedbun run typecheckfrompackages/cli— PASSbun testfrompackages/cli— 1,334 passed, 7 skipped, 0 failed- pre-push workspace typecheck — 6/6 tasks successful
A single bounded re-review has been triggered for this new head by the synchronized PR workflows.
| }) | ||
|
|
||
| test("releases the inner iterator after an idle timeout", async () => { | ||
| const pending = Promise.withResolvers<IteratorResult<string>>() |
There was a problem hiding this comment.
🟡 "releases iterator" test gives false confidence.
Suggested change
| const pending = Promise.withResolvers<IteratorResult<string>>() | |
| Either rename to "calls return() on the inner iterator after an idle timeout" to reflect what is actually asserted, or add a second test using a real `async function*` whose `next()` is gated on a promise that `abort()` resolves, demonstrating that release actually requires abort to propagate to the underlying stream. |
🤖 Fix with your agent
Fix this code review finding (aictrl-dev/cli PR #99, packages/cli/test/session/idle.test.ts:48-65):
Problem: "releases iterator" test gives false confidence
Detail: The test "releases the inner iterator after an idle timeout" uses a hand-rolled iterator whose `return()` is a plain async function — invoking it runs the body synchronously and sets `released = true`. Real iterator targets (the LLM SDK's `fullStream`, async generators in general) queue `.return()` behind an in-flight `.next()` per ECMA-262: the enqueued return-completion only runs once the pending next() settles. The production code's own comment in `idle.ts` (lines 36-38) acknowledges this ("may never settle for the stalled stream we are escaping"). With `next: () => pending.promise` (never resolves) and `abort: () => {}` (no-op), production cleanup depends entirely on `abort()` causing the provider's pending next() to settle; if the provider ignores the abort signal, `return()` never executes. The test asserts `released === true` unconditionally and so masks this production gap — a refactor that breaks abort propagation (or drops the `iterator.return?.()` call in favour of relying solely on abort) could still pass this test for the wrong reason.
Suggested fix: Either rename to "calls return() on the inner iterator after an idle timeout" to reflect what is actually asserted, or add a second test using a real `async function*` whose `next()` is gated on a promise that `abort()` resolves, demonstrating that release actually requires abort to propagate to the underlying stream.
Implement the fix on the PR head branch and add a regression test that fails before the fix and passes after.
Why this matters
The test "releases the inner iterator after an idle timeout" uses a hand-rolled iterator whose return() is a plain async function — invoking it runs the body synchronously and sets released = true. Real iterator targets (the LLM SDK's fullStream, async generators in general) queue .return() behind an in-flight .next() per ECMA-262: the enqueued return-completion only runs once the pending next() settles. The production code's own comment in idle.ts (lines 36-38) acknowledges this ("may never settle for the stalled stream we are escaping"). With next: () => pending.promise (never resolves) and abort: () => {} (no-op), production cleanup depends entirely on abort() causing the provider's pending next() to settle; if the provider ignores the abort signal, return() never executes. The test asserts released === true unconditionally and so masks this production gap — a refactor that breaks abort propagation (or drops the iterator.return?.() call in favour of relying solely on abort) could still pass this test for the wrong reason.
test("releases the inner iterator after an idle timeout", async () => {
const pending = Promise.withResolvers<IteratorResult<string>>()
let released = false
const stream = {
[Symbol.asyncIterator]() {
return {
next: () => pending.promise,
async return() {
released = true
return { done: true as const, value: undefined }
},
}
},
}
await StreamIdle.timeout(stream, 10, () => {}).next().catch(() => {})
expect(released).toBe(true)
})
Code reviewVerdict: Looks good — only minor / nit comments below. · 🔴 0 · 🟠 0 · 🟡 1 · ⚪ 0 · 0/1 resolved
🤖 Fix all 1 open findings with your agent📋 Out-of-diff findings (1)
Reviewed 9 files · 0 inline · view all 1 findings ↗ aictrl · AI code review for fast-moving teams · aictrl.dev |
Contributor
Author
Final review finding addressed at
|
| Finding | Verdict / action | Resolution | Verification |
|---|---|---|---|
| “releases iterator” test gives false confidence | TRUE · FIX | Renamed the test to “calls return on the inner iterator after an idle timeout” and changed released to called. The assertion now describes exactly what the hand-rolled iterator proves: the wrapper invokes the iterator protocol cleanup hook. It intentionally does not claim that a real async generator completes return() while a permanently pending next() ignores abort; idle.ts already documents that limitation. |
bun test test/session/idle.test.ts — 8 passed, 0 failed; bun run typecheck — PASS; pre-push workspace typecheck — 6/6 tasks successful |
No production behavior changed, so the previously green full package suite remains applicable (1,334 passed, 7 skipped, 0 failed at f842b2af6). This is the final bounded response; no further re-review is requested.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #80
Intent
Bound provider streams that stop producing activity so a headless session cannot remain busy indefinitely.
Expected outcomes
Expected impact
Unresponsive providers no longer wedge sessions or downstream automation indefinitely, while legitimately long streams continue as long as they produce activity. Operators retain a documented compatibility escape hatch.
Summary
AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MSpositive overrides and0opt-outMODEL_STREAM_IDLE_TIMEOUTVerification
bun test test/session/processor-idle.test.ts test/session/idle.test.ts test/cli/classify-session-error.test.tsbun run typecheck(packages/cli)bun test(packages/cli: 1330 pass, 7 skip, 0 fail)Risk / rollout
The default is intentionally generous and idle-based, so active long-running reasoning streams continue normally. Set the environment variable to
0to preserve unlimited idle time if a deployment requires it.