import { afterEach, describe, expect, test, vi } from "vitest";
import { setLogSink } from "../src/log.js";
import { getModel } from "../src/models.js";
import { streamOpenAICompletions } from "../src/providers/openai-completions.js";
import type { AssistantMessage, Context, Model } from "../src/types.js";
import {
classifyStreamFailure,
extractStreamFailureInfo,
formatStreamFailureMessage,
parseProviderResetMs,
recordStreamFailure,
StreamFailureError,
streamFailureFromStopReason,
} from "../src/utils/stream-failure.js";
const originalFetch = global.fetch;
afterEach(() => {
setLogSink(undefined);
global.fetch = originalFetch;
vi.restoreAllMocks();
});
function makeOutput(overrides: Partial<AssistantMessage> = {}): AssistantMessage {
return {
role: "assistant",
content: [],
api: "anthropic-messages",
provider: "anthropic",
model: "test-model",
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "error",
timestamp: 0,
...overrides,
};
}
describe("classifyStreamFailure", () => {
test.each([
["overloaded_error", undefined, "overloaded"],
[undefined, 529, "overloaded"],
["rate_limit_error", undefined, "rate_limit"],
["usage_limit_reached", undefined, "rate_limit"],
["usage_not_included", 403, "rate_limit"],
["insufficient_quota", 429, "quota_exhausted"],
[undefined, 429, "rate_limit"],
["refusal", undefined, "refusal"],
["sensitive", undefined, "safety"],
["SAFETY", undefined, "safety"],
["PROHIBITED_CONTENT", undefined, "safety"],
["content_filter", undefined, "safety"],
["guardrail_intervened", undefined, "safety"],
["authentication_error", undefined, "auth"],
[undefined, 401, "auth"],
["permission_error", 403, "permission"],
["PermissionDeniedError", 403, "permission"],
[undefined, 403, "permission"],
["invalid_request_error", undefined, "invalid_request"],
["api_error", undefined, "server_error"],
["terminated", undefined, "server_error"],
[undefined, 503, "server_error"],
["something_else", undefined, "unknown"],
])("classifies %s / %s as %s", (type, status, expected) => {
expect(classifyStreamFailure(type, status)).toBe(expected);
});
test.each([
[
"depletion in the provider message",
429,
"You exceeded your current quota, please check your plan and billing details.",
"quota_exhausted",
],
["depletion in the credits text", 429, "Your account is out of credits.", "quota_exhausted"],
["depletion in the billing text", 429, "Billing hard limit reached.", "quota_exhausted"],
["keyword-free burst", 429, "Rate limit reached for gpt-5 on requests per min (RPM)", "rate_limit"],
["per-minute token limit", 429, "Number of request tokens has exceeded your per-minute rate limit", "rate_limit"],
] as const)("classifies a rate limit by its words: %s", (_name, status, message, expected) => {
expect(classifyStreamFailure(undefined, status, message)).toBe(expected);
});
test("quota words outside a rate limit never change the verdict", () => {
expect(classifyStreamFailure("invalid_request_error", 400, "unknown parameter quota")).toBe("invalid_request");
expect(classifyStreamFailure("authentication_error", 401, "billing profile missing")).toBe("auth");
});
});
describe("streamFailureFromStopReason", () => {
test("preserves the raw stop reason instead of a generic message", () => {
const error = streamFailureFromStopReason("refusal", { requestId: "req_abc" });
expect(error.message).toBe("Model refused to respond (refusal) [request_id: req_abc]");
expect(error.info).toMatchObject({ kind: "refusal", providerErrorType: "refusal", requestId: "req_abc" });
});
test("maps Gemini safety finish reasons", () => {
expect(streamFailureFromStopReason("SAFETY").info.kind).toBe("safety");
expect(streamFailureFromStopReason("MALFORMED_FUNCTION_CALL").info.kind).toBe("malformed_response");
});
test("still explains a missing stop reason", () => {
const error = streamFailureFromStopReason(undefined);
expect(error.info.kind).toBe("unknown");
expect(error.message).toContain("no stop reason");
});
});
describe("extractStreamFailureInfo", () => {
test("passes through StreamFailureError info", () => {
const info = { kind: "overloaded" as const, requestId: "req_1" };
expect(extractStreamFailureInfo(new StreamFailureError("x", info))).toBe(info);
});
test("extracts status, nested error type, and request id from SDK-shaped errors", () => {
const sdkError = Object.assign(new Error("529 overloaded"), {
status: 529,
error: { type: "error", error: { type: "overloaded_error", message: "Overloaded" } },
requestID: "req_sdk",
});
expect(extractStreamFailureInfo(sdkError)).toMatchObject({
kind: "overloaded",
providerErrorType: "overloaded_error",
status: 529,
requestId: "req_sdk",
});
});
test("extracts AWS SDK request id and exception name", () => {
const awsError = Object.assign(new Error("throttled"), {
name: "ThrottlingException",
$metadata: { requestId: "aws_req" },
});
expect(extractStreamFailureInfo(awsError)).toMatchObject({ requestId: "aws_req" });
});
test.each([
["Headers seconds", new Headers({ "retry-after": "120" }), 120000],
["retry-after-ms precedence", { "retry-after-ms": "1500", "retry-after": "2" }, 1500],
["record with mixed case", { "Retry-After": "120" }, 120000],
] as const)("extracts the server-requested retry delay: %s", (_name, headers, expected) => {
const error = Object.assign(new Error("429"), { status: 429, headers });
expect(extractStreamFailureInfo(error)).toMatchObject({ kind: "rate_limit", retryAfterMs: expected });
});
test("parses an HTTP-date Retry-After relative to now", () => {
const withDate = Object.assign(new Error("429"), {
status: 429,
headers: new Headers({ "retry-after": new Date(Date.now() + 60000).toUTCString() }),
});
const dateMs = extractStreamFailureInfo(withDate).retryAfterMs;
expect(dateMs).toBeGreaterThan(0);
expect(dateMs).toBeLessThanOrEqual(60000);
});
test("extracts retry delay embedded in provider message body without headers", () => {
const composite = new Error("429 You've hit your hourly rate limit. Please try again in 21m47s.");
expect(extractStreamFailureInfo(composite)).toMatchObject({
kind: "rate_limit",
retryAfterMs: 1307000,
});
const fixedNow = Date.parse("2026-09-17T12:00:00Z");
expect(
parseProviderResetMs(
"429 Your token-plan 1-week quota has been exhausted. The quota will reset at 09-21 20:06:00 UTC.",
fixedNow,
),
).toBe(374760000);
});
test("falls back to classifying the message text", () => {
expect(extractStreamFailureInfo(new Error("provider overloaded, retry later")).kind).toBe("overloaded");
expect(extractStreamFailureInfo("not an error").kind).toBe("unknown");
});
test("classifies an OpenAI-style exhausted-quota 429 as permanent", () => {
const body = {
type: "insufficient_quota",
message: "You exceeded your current quota, please check your plan and billing details.",
};
const error = Object.assign(new Error(`429 ${JSON.stringify({ error: body })}`), {
status: 429,
error: body,
});
expect(extractStreamFailureInfo(error)).toMatchObject({
kind: "quota_exhausted",
providerErrorType: "insufficient_quota",
status: 429,
});
});
test("classifies a Google-style RESOURCE_EXHAUSTED 429 as permanent", () => {
const body = {
code: 429,
message: "Quota exceeded for quota metric 'generate_content_free_tier_requests'",
status: "RESOURCE_EXHAUSTED",
};
const error = Object.assign(new Error(`429 ${JSON.stringify({ error: body })}`), { status: 429, error: body });
expect(extractStreamFailureInfo(error)).toMatchObject({ kind: "quota_exhausted", status: 429 });
});
test("keeps an Anthropic-style burst 429 retryable", () => {
const body = {
type: "rate_limit_error",
message: "Number of request tokens has exceeded your per-minute rate limit",
};
const error = Object.assign(new Error(`429 ${JSON.stringify({ error: body })}`), { status: 429, error: body });
expect(extractStreamFailureInfo(error)).toMatchObject({
kind: "rate_limit",
providerErrorType: "rate_limit_error",
status: 429,
});
});
test.each([
["Unauthorized: authentication failed", undefined, "unknown"],
["permission denied by policy", undefined, "unknown"],
["upstream authentication failed", 500, "server_error"],
["Unauthorized", 401, "auth"],
["permission denied", 403, "permission"],
] as const)("auth/permission need more than message text: %s / %s -> %s", (message, status, expected) => {
// Without a structured error type, only the status may decide auth or permission.
expect(extractStreamFailureInfo(Object.assign(new Error(message), { status })).kind).toBe(expected);
});
});
describe("formatStreamFailureMessage", () => {
test("condenses a classified SDK error to a one-liner instead of the raw payload", () => {
const sdkError = Object.assign(
new Error('401 {"type":"error","error":{"type":"authentication_error","message":"invalid x-api-key"}}'),
{
status: 401,
error: { type: "error", error: { type: "authentication_error", message: "invalid x-api-key" } },
requestID: "req_1",
},
);
expect(formatStreamFailureMessage(sdkError)).toBe(
"Provider authentication failed (authentication_error, 401): invalid x-api-key [request_id: req_1]",
);
});
test("passes unrecognized errors through verbatim", () => {
expect(formatStreamFailureMessage(new Error("fetch failed"))).toBe("fetch failed");
expect(formatStreamFailureMessage(new Error("Request was aborted"))).toBe("Request was aborted");
});
test("uses the StreamFailureError message as-is", () => {
const error = streamFailureFromStopReason("refusal");
expect(formatStreamFailureMessage(error)).toBe(error.message);
});
});
describe("recordStreamFailure", () => {
const model = { provider: "anthropic", id: "claude-fable-5", api: "anthropic-messages" };
test("appends a structured diagnostic and logs one entry", () => {
const logged: Record<string, unknown>[] = [];
setLogSink((entry) => logged.push(entry));
const output = makeOutput({ errorMessage: "Provider overloaded (overloaded_error) [request_id: req_9]" });
recordStreamFailure(model, output, new StreamFailureError("x", { kind: "overloaded", requestId: "req_9" }));
expect(output.diagnostics).toHaveLength(1);
expect(output.diagnostics?.[0]).toMatchObject({
type: "provider_stream_failure",
details: { kind: "overloaded", requestId: "req_9" },
});
expect(logged).toHaveLength(1);
expect(logged[0]).toMatchObject({
level: "error",
component: "ai.provider",
provider: "anthropic",
model: "claude-fable-5",
kind: "overloaded",
requestId: "req_9",
});
});
test("does nothing for user aborts", () => {
const logged: unknown[] = [];
setLogSink((entry) => logged.push(entry));
const output = makeOutput({ stopReason: "aborted" });
recordStreamFailure(model, output, new Error("Request was aborted"));
expect(output.diagnostics).toBeUndefined();
expect(logged).toEqual([]);
});
});
describe("provider retry ownership", () => {
const retryContext: Context = { messages: [{ role: "user", content: "hi", timestamp: 1 }] };
function completionsModel(): Model<"openai-completions"> {
const { compat: _compat, ...baseModel } = getModel("openai", "gpt-4o-mini")!;
return { ...baseModel, api: "openai-completions" } as Model<"openai-completions">;
}
test.each([
[
"makes exactly one request on a 500 and records a structured stream failure",
{ type: "server_error", message: "boom" },
{ status: 500 },
{ kind: "server_error", status: 500 },
],
[
"surfaces the server-requested Retry-After delay on rate limits",
{ type: "rate_limit_error", message: "slow down" },
{ status: 429, headers: { "retry-after": "30" } },
{ kind: "rate_limit", status: 429, retryAfterMs: 30000 },
],
] as const)("%s", async (_name, errorBody, init, expectedDetails) => {
const fetchMock = vi.fn(async () => new Response(JSON.stringify({ error: errorBody }), init));
global.fetch = fetchMock as typeof fetch;
let failed: AssistantMessage | undefined;
for await (const event of streamOpenAICompletions(completionsModel(), retryContext, { apiKey: "test-key" })) {
if (event.type === "error") {
failed = event.error;
break;
}
}
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(failed?.stopReason).toBe("error");
expect(failed?.diagnostics?.[0]).toMatchObject({ type: "provider_stream_failure", details: expectedDetails });
});
});