From 2f24d4457ee8f75c2b78feff9acd1906cda8948a Mon Sep 17 00:00:00 2001 From: Voon Foo Date: Fri, 7 Aug 2026 14:18:46 +0800 Subject: [PATCH] fix(ai,catalog): widen Bedrock stream-stall watchdog via model compat The lazy provider wrapper ignored model.compat.streamIdleTimeoutMs, so Bedrock reasoning models sat on the generic 300s idle watchdog despite ConverseStream sending no ping keepalives; long quiet thinking runs died with "Provider stream stalled while waiting for the next event" during plan writing and todo execution (issue #4758's Bedrock variant, worst on Fable 5 where the display default flipped to omitted). - catalog: BedrockCompat gains streamIdleTimeoutMs; reasoning models get a 600s floor, adaptive-thinking Claude (Opus 4.7+, Sonnet/Opus 5, Fable/Mythos 5) 900s to match direct Anthropic's ping-extended tolerance; explicit compat overrides still win (0 disables). - ai: forwardStream resolves options -> env -> model.compat -> default, and lazy terminal errors carry the structural errorId classification so session auto-retry classifies stalls without text matching. --- .../ai/src/providers/register-builtins.ts | 12 +++- packages/ai/test/register-builtins.test.ts | 59 ++++++++++++++++- packages/catalog/src/compat/bedrock.ts | 29 ++++++++- packages/catalog/src/types.ts | 13 ++++ .../test/amazon-bedrock-opus-5.test.ts | 2 + .../catalog/test/bedrock-prompt-cache.test.ts | 12 +++- .../test/bedrock-stream-idle-timeout.test.ts | 65 +++++++++++++++++++ 7 files changed, 185 insertions(+), 7 deletions(-) create mode 100644 packages/catalog/test/bedrock-stream-idle-timeout.test.ts diff --git a/packages/ai/src/providers/register-builtins.ts b/packages/ai/src/providers/register-builtins.ts index 8bb3a20f0..4eb067a7e 100644 --- a/packages/ai/src/providers/register-builtins.ts +++ b/packages/ai/src/providers/register-builtins.ts @@ -240,12 +240,19 @@ function forwardStream( (async () => { try { const providerHandlesStreamTimeouts = limits?.providerHandlesStreamTimeouts === true; + // Per-model catalog compat can widen the fallback watchdog for hosts + // with no keepalive events (e.g. Bedrock reasoning models that go + // quiet for minutes mid-thinking, issue #4758). Caller options and + // env overrides still take precedence over the compat fallback. + const compatIdleTimeoutMs = (model.compat as { streamIdleTimeoutMs?: number } | undefined) + ?.streamIdleTimeoutMs; + const idleTimeoutFallbackMs = compatIdleTimeoutMs ?? limits?.defaultIdleTimeoutMs; const idleTimeoutMs = providerHandlesStreamTimeouts ? undefined : (options.streamIdleTimeoutMs ?? (limits?.openAIIdleEnvFloorsFirstEvent - ? getOpenAIStreamIdleTimeoutMs(limits.defaultIdleTimeoutMs) - : getStreamIdleTimeoutMs(limits?.defaultIdleTimeoutMs))); + ? getOpenAIStreamIdleTimeoutMs(idleTimeoutFallbackMs) + : getStreamIdleTimeoutMs(idleTimeoutFallbackMs))); const firstItemTimeoutMs = providerHandlesStreamTimeouts ? 0 : (options.streamFirstEventTimeoutMs ?? @@ -311,6 +318,7 @@ function createLazyLoadErrorMessage( cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, stopReason, + errorId: stopReason === "error" ? AIError.classify(error, model.api) || undefined : undefined, errorMessage: stopReason === "aborted" ? "Request was aborted" : error instanceof Error ? error.message : String(error), timestamp: Date.now(), diff --git a/packages/ai/test/register-builtins.test.ts b/packages/ai/test/register-builtins.test.ts index 23326aa75..0d46fc550 100644 --- a/packages/ai/test/register-builtins.test.ts +++ b/packages/ai/test/register-builtins.test.ts @@ -1,21 +1,25 @@ import { describe, expect, it } from "bun:test"; +import * as AIError from "@oh-my-pi/pi-ai/error"; import { setBedrockProviderModule, streamBedrock } from "@oh-my-pi/pi-ai/providers/register-builtins"; import type { AssistantMessage, Context, Model } from "@oh-my-pi/pi-ai/types"; import type { AssistantMessageEventStream } from "@oh-my-pi/pi-ai/utils/event-stream"; import { buildModel } from "@oh-my-pi/pi-catalog/build"; -function createModel(): Model<"bedrock-converse-stream"> { +function createModel( + overrides: { reasoning?: boolean; compat?: { streamIdleTimeoutMs?: number } } = {}, +): Model<"bedrock-converse-stream"> { return buildModel({ id: "mock-bedrock", name: "Mock Bedrock", api: "bedrock-converse-stream", provider: "amazon-bedrock", baseUrl: "https://example.invalid", - reasoning: false, + reasoning: overrides.reasoning ?? false, input: ["text"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 8192, maxTokens: 2048, + compat: overrides.compat, }); } @@ -129,6 +133,57 @@ describe("register-builtins lazy streams", () => { expect(providerSignal?.aborted).toBe(true); expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe("Provider stream stalled while waiting for the next event"); + // The watchdog's StreamTimeoutError classification must survive onto the + // message so session-level auto-retry can classify it structurally. + expect(AIError.is(result.errorId, AIError.Flag.Transient)).toBe(true); + expect(AIError.is(result.errorId, AIError.Flag.Timeout)).toBe(true); + expect(AIError.retriable(result.errorId)).toBe(true); + }); + + it("honors model.compat.streamIdleTimeoutMs as the lazy watchdog fallback", async () => { + const partialMessage = createAssistantMessage("stop"); + let providerSignal: AbortSignal | undefined; + const source = { + async *[Symbol.asyncIterator]() { + yield { type: "start", partial: partialMessage } as const; + yield { type: "text_delta", contentIndex: 0, delta: "hello", partial: partialMessage } as const; + const { promise, reject } = Promise.withResolvers(); + if (providerSignal?.aborted) { + reject(new Error("Request was aborted")); + } + providerSignal?.addEventListener("abort", () => reject(new Error("Request was aborted")), { + once: true, + }); + await promise; + }, + } as unknown as AssistantMessageEventStream; + + setBedrockProviderModule({ + streamBedrock: (_model, _context, options) => { + providerSignal = options.signal; + return source; + }, + }); + + // No per-call option: the catalog compat override must reach the lazy + // watchdog (a stalled Bedrock stream previously waited the generic 300s + // default because model.compat was ignored on this path). + const model = createModel({ reasoning: true, compat: { streamIdleTimeoutMs: 20 } }); + expect(model.compat.streamIdleTimeoutMs).toBe(20); + const stream = streamBedrock(model, baseContext, {}); + // Real-clock race guard (matching this file's other lazy-stream tests): + // the lazy watchdog runs on the platform clock, so fake timers cannot + // drive it; the 20ms compat deadline settles the result long before the + // bound, which exists only to fail fast instead of hanging the test. + const result = await Promise.race([stream.result(), Bun.sleep(2_000).then(() => "timeout" as const)]); + + expect(result).not.toBe("timeout"); + if (result === "timeout") { + throw new Error("Timed out waiting for compat-driven stream stall result"); + } + expect(providerSignal?.aborted).toBe(true); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toBe("Provider stream stalled while waiting for the next event"); }); it("preserves caller aborts while forwarding lazy provider streams", async () => { diff --git a/packages/catalog/src/compat/bedrock.ts b/packages/catalog/src/compat/bedrock.ts index d8258c193..7ec13ea0f 100644 --- a/packages/catalog/src/compat/bedrock.ts +++ b/packages/catalog/src/compat/bedrock.ts @@ -1,3 +1,4 @@ +import { supportsAdaptiveThinkingDisplay } from "../identity/family"; import type { ModelSpec, ResolvedBedrockCompat } from "../types"; import { applyCompatOverrides } from "./apply"; @@ -118,9 +119,35 @@ function detectedBedrockCompat(modelId: string): ResolvedBedrockCompat { return NO_EXPLICIT_CHECKPOINTS; } -/** Resolve Bedrock Converse prompt-cache capabilities once per model. */ +/** + * Bedrock ConverseStream sends no ping/keepalive events, so a reasoning model + * that goes quiet mid-thinking (summarized-display gaps, `omitted` thinking, + * or the wedged long tool-call generation of issue #4900) reads as a dead + * stream to the generic 300s idle watchdog and dies with "Provider stream + * stalled while waiting for the next event" (issue #4758's Bedrock variant). + * Widen the floor to 600s for reasoning models, mirroring the GLM coding-plan + * floor; explicit `spec.compat.streamIdleTimeoutMs` overrides still win. + */ +const BEDROCK_REASONING_STREAM_IDLE_TIMEOUT_MS = 600_000; +/** + * Adaptive-thinking Claude (Opus 4.7+, Sonnet/Opus 5, Fable/Mythos 5) reasons + * for much longer stretches, and starting with Opus 4.7 / Fable 5 the + * Anthropic-side display default is `omitted` (issue #1373), so quiet gaps run + * longest on exactly this family — Fable 5 being the worst offender in the + * field. Direct Anthropic keeps these streams alive with ping keepalives and + * tolerates up to 3x the 300s idle budget of real-event silence (#4900); + * pingless Bedrock needs the same 900s tolerance in the raw idle floor. + */ +const BEDROCK_ADAPTIVE_THINKING_STREAM_IDLE_TIMEOUT_MS = 900_000; + +/** Resolve Bedrock Converse prompt-cache and stream-watchdog compat once per model. */ export function buildBedrockCompat(spec: ModelSpec<"bedrock-converse-stream">): ResolvedBedrockCompat { const compat = { ...detectedBedrockCompat(spec.id) }; + compat.streamIdleTimeoutMs = spec.reasoning + ? supportsAdaptiveThinkingDisplay(spec.id) + ? BEDROCK_ADAPTIVE_THINKING_STREAM_IDLE_TIMEOUT_MS + : BEDROCK_REASONING_STREAM_IDLE_TIMEOUT_MS + : undefined; applyCompatOverrides(compat, spec.compat); return compat; } diff --git a/packages/catalog/src/types.ts b/packages/catalog/src/types.ts index 4e289759e..614255eb2 100644 --- a/packages/catalog/src/types.ts +++ b/packages/catalog/src/types.ts @@ -518,6 +518,12 @@ export interface BedrockCompat { * Capability metadata only; zero means no explicit checkpoints. */ promptCacheMaximumCheckpoints?: number; + /** + * Stream-watchdog idle-timeout fallback in ms; 0 disables the idle watchdog. + * Undefined defers to `PI_STREAM_IDLE_TIMEOUT_MS`, then the legacy + * `PI_OPENAI_STREAM_IDLE_TIMEOUT_MS` alias, then the 300s default. + */ + streamIdleTimeoutMs?: number; } /** Fully-resolved Bedrock Converse prompt-cache capabilities, materialized once by `buildModel`. */ @@ -526,6 +532,13 @@ export interface ResolvedBedrockCompat { supportsLongPromptCacheRetention: boolean; promptCacheMinimumTokens: number; promptCacheMaximumCheckpoints: number; + /** + * Stream-watchdog idle-timeout fallback in ms for hosts with no keepalive + * events; 0 disables the idle watchdog. Undefined defers to + * `PI_STREAM_IDLE_TIMEOUT_MS`, then the legacy + * `PI_OPENAI_STREAM_IDLE_TIMEOUT_MS` alias, then the 300s default. + */ + streamIdleTimeoutMs?: number; } /** diff --git a/packages/catalog/test/amazon-bedrock-opus-5.test.ts b/packages/catalog/test/amazon-bedrock-opus-5.test.ts index 4b4d9591e..f73d096a1 100644 --- a/packages/catalog/test/amazon-bedrock-opus-5.test.ts +++ b/packages/catalog/test/amazon-bedrock-opus-5.test.ts @@ -142,6 +142,8 @@ describe("Amazon Bedrock Claude Opus 5", () => { supportsLongPromptCacheRetention: true, promptCacheMinimumTokens: 512, promptCacheMaximumCheckpoints: 4, + // reasoning:true adaptive-thinking family → 900s keepalive-free idle floor. + streamIdleTimeoutMs: 900_000, }); } }); diff --git a/packages/catalog/test/bedrock-prompt-cache.test.ts b/packages/catalog/test/bedrock-prompt-cache.test.ts index 38269feab..e9dd72c2c 100644 --- a/packages/catalog/test/bedrock-prompt-cache.test.ts +++ b/packages/catalog/test/bedrock-prompt-cache.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test"; import { buildModel } from "@oh-my-pi/pi-catalog/build"; +import { supportsAdaptiveThinkingDisplay } from "@oh-my-pi/pi-catalog/identity"; import { getBundledModel } from "@oh-my-pi/pi-catalog/models"; import type { ModelSpec } from "@oh-my-pi/pi-catalog/types"; @@ -90,6 +91,9 @@ describe("Bedrock prompt-cache compat", () => { supportsLongPromptCacheRetention: supportsLongRetention, promptCacheMinimumTokens: minimumTokens, promptCacheMaximumCheckpoints: minimumTokens === 0 ? 0 : 4, + // bedrockSpec is reasoning:true → keepalive-free idle floor applies + // (900s for the adaptive-thinking family, 600s otherwise). + streamIdleTimeoutMs: supportsAdaptiveThinkingDisplay(id) ? 900_000 : 600_000, }); } }); @@ -109,7 +113,11 @@ describe("Bedrock prompt-cache compat", () => { "us.amazon.nova-premier-v1:0", "global.amazon.nova-2-lite-v1:0", ] as const) { - expect(getBundledModel<"bedrock-converse-stream">("amazon-bedrock", id)?.compat).toEqual(expected); + const model = getBundledModel<"bedrock-converse-stream">("amazon-bedrock", id); + expect(model?.compat).toEqual({ + ...expected, + streamIdleTimeoutMs: model?.reasoning ? 600_000 : undefined, + }); } // AWS documents in-region model IDs plus geo/global inference-profile IDs. @@ -125,7 +133,7 @@ describe("Bedrock prompt-cache compat", () => { "jp.amazon.nova-2-lite-v1:0", "global.amazon.nova-2-lite-v1:0", ] as const) { - expect(buildModel(bedrockSpec({ id })).compat).toEqual(expected); + expect(buildModel(bedrockSpec({ id })).compat).toEqual({ ...expected, streamIdleTimeoutMs: 600_000 }); } }); diff --git a/packages/catalog/test/bedrock-stream-idle-timeout.test.ts b/packages/catalog/test/bedrock-stream-idle-timeout.test.ts new file mode 100644 index 000000000..43a4bb428 --- /dev/null +++ b/packages/catalog/test/bedrock-stream-idle-timeout.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, test } from "bun:test"; +import { buildModel } from "@oh-my-pi/pi-catalog/build"; +import type { ModelSpec } from "@oh-my-pi/pi-catalog/types"; + +function bedrockSpec( + overrides: Partial> = {}, +): ModelSpec<"bedrock-converse-stream"> { + return { + id: "global.anthropic.claude-fable-5", + name: "Claude Fable 5", + api: "bedrock-converse-stream", + provider: "amazon-bedrock", + baseUrl: "https://bedrock-runtime.us-east-1.amazonaws.com", + reasoning: true, + input: ["text"], + cost: { input: 5, output: 25, cacheRead: 0.5, cacheWrite: 6.25 }, + contextWindow: 1_000_000, + maxTokens: 128_000, + ...overrides, + }; +} + +// Bedrock ConverseStream sends no ping keepalives, so reasoning models that go +// quiet mid-thinking previously fell back to the generic 300s idle watchdog and +// died with "Provider stream stalled while waiting for the next event" during +// long plan-writing/reasoning phases (issue #4758's Bedrock variant, worst on +// the adaptive-thinking Fable/Opus 4.7+ family). +describe("Bedrock stream idle-timeout compat", () => { + test("widens the idle timeout to 900s for adaptive-thinking Claude", () => { + for (const id of [ + "global.anthropic.claude-fable-5", + "global.anthropic.claude-fable-5-20260120-v1:0", + "us.anthropic.claude-opus-4-8", + "us.anthropic.claude-sonnet-5", + "us.anthropic.claude-opus-5", + ]) { + expect(buildModel(bedrockSpec({ id })).compat.streamIdleTimeoutMs).toBe(900_000); + } + }); + + test("widens the idle timeout to 600s for other reasoning models", () => { + for (const id of [ + "anthropic.claude-opus-4-6-v1", + "anthropic.claude-3-7-sonnet-20250219-v1:0", + "us.amazon.nova-premier-v1:0", + ]) { + expect(buildModel(bedrockSpec({ id })).compat.streamIdleTimeoutMs).toBe(600_000); + } + }); + + test("leaves non-reasoning models on the generic default", () => { + expect( + buildModel(bedrockSpec({ id: "anthropic.claude-3-5-haiku-20241022-v1:0", reasoning: false })).compat + .streamIdleTimeoutMs, + ).toBeUndefined(); + }); + + test("explicit compat overrides win over the reasoning floors", () => { + expect(buildModel(bedrockSpec({ compat: { streamIdleTimeoutMs: 120_000 } })).compat.streamIdleTimeoutMs).toBe( + 120_000, + ); + // 0 disables the idle watchdog entirely. + expect(buildModel(bedrockSpec({ compat: { streamIdleTimeoutMs: 0 } })).compat.streamIdleTimeoutMs).toBe(0); + }); +});