fix(ai): retry safe truncated responses streams
This commit is contained in:
usr_bin_roygbiv
@@ -1,6 +1,9 @@
|
||||
# Changelog
|
||||
|
||||
## [Unreleased]
|
||||
### Fixed
|
||||
|
||||
- Retry one transient OpenAI Responses stream truncation before replay-unsafe output, preventing recoverable transport truncations from surfacing as failed turns ([#5908](https://github.com/can1357/oh-my-pi/issues/5908)).
|
||||
|
||||
## [17.0.3] - 2026-07-17
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { scheduler } from "node:timers/promises";
|
||||
import { hostMatchesUrl } from "@oh-my-pi/pi-catalog/hosts";
|
||||
import { $flag, logger, structuredCloneJSON } from "@oh-my-pi/pi-utils";
|
||||
import * as AIError from "../error";
|
||||
@@ -154,6 +155,33 @@ const OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE =
|
||||
"OpenAI responses stream timed out while waiting for the first event";
|
||||
/** Consecutive stale-previous-response failures before chaining is disabled for the session. */
|
||||
const OPENAI_RESPONSES_CHAIN_STALE_FAILURE_LIMIT = 3;
|
||||
const OPENAI_RESPONSES_MAX_TRANSIENT_STREAM_RETRIES = 1;
|
||||
const OPENAI_RESPONSES_TRANSIENT_STREAM_RETRY_DELAY_MS = 500;
|
||||
|
||||
function isOpenAIResponsesReplayUnsafeEvent(event: ResponseStreamEvent): boolean {
|
||||
switch (event.type) {
|
||||
case "response.output_text.delta":
|
||||
case "response.refusal.delta":
|
||||
case "response.reasoning_summary_text.delta":
|
||||
case "response.reasoning_text.delta":
|
||||
case "response.function_call_arguments.delta":
|
||||
case "response.custom_tool_call_input.delta":
|
||||
return typeof event.delta === "string" && event.delta.length > 0;
|
||||
case "response.reasoning_summary_part.done":
|
||||
return true;
|
||||
case "response.output_item.done":
|
||||
return true;
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function isRetryableOpenAIResponsesStreamFailure(error: unknown): boolean {
|
||||
return (
|
||||
AIError.isTransientStreamParseError(error) ||
|
||||
(error instanceof AIError.ProviderResponseError && error.kind === "incomplete-stream")
|
||||
);
|
||||
}
|
||||
|
||||
interface OpenAIResponsesProviderSessionState
|
||||
extends ProviderSessionState,
|
||||
@@ -452,7 +480,7 @@ const streamOpenAIResponsesOnce = (
|
||||
return payload;
|
||||
};
|
||||
chained = { ...chained, params: await applyPayloadReplacement(chained.params) };
|
||||
rawRequestDump = {
|
||||
const activeRawRequestDump: RawHttpRequestDump = {
|
||||
provider: model.provider,
|
||||
api: output.api,
|
||||
model: model.id,
|
||||
@@ -460,6 +488,7 @@ const streamOpenAIResponsesOnce = (
|
||||
url: requestUrl,
|
||||
body: chained.params,
|
||||
};
|
||||
rawRequestDump = activeRawRequestDump;
|
||||
const openResponsesStream = (requestParams: OpenAIResponsesSamplingParams) => {
|
||||
activeReasoningEffortFallbackKey = createOpenAIReasoningEffortFallbackKey(
|
||||
"responses",
|
||||
@@ -507,187 +536,257 @@ const streamOpenAIResponsesOnce = (
|
||||
{ provider: model.provider, signal: requestSignal },
|
||||
);
|
||||
};
|
||||
let openaiStream: AsyncIterable<ResponseStreamEvent>;
|
||||
let strictRetryAvailable = true;
|
||||
let activeStrictToolsApplied = builtParams.strictToolsApplied;
|
||||
let forceDisableStrictTools = false;
|
||||
while (true) {
|
||||
try {
|
||||
openaiStream = await openResponsesStream(chained.params);
|
||||
if (pendingReasoningEffortFallback) {
|
||||
rememberOpenAIReasoningEffortFallback(
|
||||
providerSessionState,
|
||||
pendingReasoningEffortFallback.key,
|
||||
pendingReasoningEffortFallback.fallback,
|
||||
);
|
||||
pendingReasoningEffortFallback = undefined;
|
||||
}
|
||||
break;
|
||||
} catch (error) {
|
||||
const capturedErrorResponse = error instanceof OpenAIHttpError ? error.captured : undefined;
|
||||
const reasoningEffortFallback =
|
||||
activeReasoningEffortFallbackKey && activeRequestParams && !requestSignal.aborted
|
||||
? resolveOpenAIReasoningEffortFallback(error, capturedErrorResponse, activeRequestParams, {
|
||||
explicitDisable: options?.disableReasoning === true && options.reasoning === undefined,
|
||||
})
|
||||
: undefined;
|
||||
if (reasoningEffortFallback !== undefined && activeReasoningEffortFallbackKey) {
|
||||
const retryMarker = `${activeReasoningEffortFallbackKey}:${String(reasoningEffortFallback)}`;
|
||||
if (attemptedReasoningEffortFallbacks.has(retryMarker)) throw error;
|
||||
attemptedReasoningEffortFallbacks.add(retryMarker);
|
||||
requestReasoningEffortFallbacks.set(activeReasoningEffortFallbackKey, reasoningEffortFallback);
|
||||
applyOpenAIReasoningEffortFallback(chained.params, reasoningEffortFallback);
|
||||
applyOpenAIReasoningEffortFallback(activeParams, reasoningEffortFallback);
|
||||
rawRequestDump.body = chained.params;
|
||||
pendingReasoningEffortFallback = {
|
||||
key: activeReasoningEffortFallbackKey,
|
||||
fallback: reasoningEffortFallback,
|
||||
};
|
||||
continue;
|
||||
}
|
||||
const compiledGrammarTooLarge =
|
||||
isOpenRouterAnthropicModel(model) &&
|
||||
isCompiledGrammarTooLargeStrictError(error, capturedErrorResponse);
|
||||
const canRetryWithoutStrictTools =
|
||||
strictRetryAvailable &&
|
||||
!requestSignal.aborted &&
|
||||
(compiledGrammarTooLarge ||
|
||||
shouldRetryWithoutStrictTools(
|
||||
error,
|
||||
capturedErrorResponse,
|
||||
activeStrictToolsApplied,
|
||||
context.tools,
|
||||
));
|
||||
if (canRetryWithoutStrictTools) {
|
||||
strictRetryAvailable = false;
|
||||
forceDisableStrictTools = true;
|
||||
disableStrictToolsForScope(providerSessionState, strictToolsScope);
|
||||
const fallbackBuilt = buildParams(
|
||||
const openResponsesStreamWithFallbacks = async (): Promise<AsyncIterable<ResponseStreamEvent>> => {
|
||||
let openaiStream: AsyncIterable<ResponseStreamEvent>;
|
||||
while (true) {
|
||||
try {
|
||||
openaiStream = await openResponsesStream(chained.params);
|
||||
if (pendingReasoningEffortFallback) {
|
||||
rememberOpenAIReasoningEffortFallback(
|
||||
providerSessionState,
|
||||
pendingReasoningEffortFallback.key,
|
||||
pendingReasoningEffortFallback.fallback,
|
||||
);
|
||||
pendingReasoningEffortFallback = undefined;
|
||||
}
|
||||
break;
|
||||
} catch (error) {
|
||||
const capturedErrorResponse = error instanceof OpenAIHttpError ? error.captured : undefined;
|
||||
const reasoningEffortFallback =
|
||||
activeReasoningEffortFallbackKey && activeRequestParams && !requestSignal.aborted
|
||||
? resolveOpenAIReasoningEffortFallback(error, capturedErrorResponse, activeRequestParams, {
|
||||
explicitDisable: options?.disableReasoning === true && options.reasoning === undefined,
|
||||
})
|
||||
: undefined;
|
||||
if (reasoningEffortFallback !== undefined && activeReasoningEffortFallbackKey) {
|
||||
const retryMarker = `${activeReasoningEffortFallbackKey}:${String(reasoningEffortFallback)}`;
|
||||
if (attemptedReasoningEffortFallbacks.has(retryMarker)) throw error;
|
||||
attemptedReasoningEffortFallbacks.add(retryMarker);
|
||||
requestReasoningEffortFallbacks.set(activeReasoningEffortFallbackKey, reasoningEffortFallback);
|
||||
applyOpenAIReasoningEffortFallback(chained.params, reasoningEffortFallback);
|
||||
applyOpenAIReasoningEffortFallback(activeParams, reasoningEffortFallback);
|
||||
activeRawRequestDump.body = chained.params;
|
||||
pendingReasoningEffortFallback = {
|
||||
key: activeReasoningEffortFallbackKey,
|
||||
fallback: reasoningEffortFallback,
|
||||
};
|
||||
continue;
|
||||
}
|
||||
const compiledGrammarTooLarge =
|
||||
isOpenRouterAnthropicModel(model) &&
|
||||
isCompiledGrammarTooLargeStrictError(error, capturedErrorResponse);
|
||||
const canRetryWithoutStrictTools =
|
||||
strictRetryAvailable &&
|
||||
!requestSignal.aborted &&
|
||||
(compiledGrammarTooLarge ||
|
||||
shouldRetryWithoutStrictTools(
|
||||
error,
|
||||
capturedErrorResponse,
|
||||
activeStrictToolsApplied,
|
||||
context.tools,
|
||||
));
|
||||
if (canRetryWithoutStrictTools) {
|
||||
strictRetryAvailable = false;
|
||||
forceDisableStrictTools = true;
|
||||
disableStrictToolsForScope(providerSessionState, strictToolsScope);
|
||||
const fallbackBuilt = buildParams(
|
||||
model,
|
||||
context,
|
||||
options,
|
||||
providerSessionState,
|
||||
strictToolsScope,
|
||||
true,
|
||||
);
|
||||
const fallbackParams = fallbackBuilt.params;
|
||||
if (chainState && !chainState.disabled) fallbackParams.store = true;
|
||||
let fallbackChained: OpenAIResponsesChainedParams =
|
||||
chainState && !chainState.disabled
|
||||
? buildOpenAIResponsesChainedParams(fallbackParams, chainState)
|
||||
: { params: fallbackParams };
|
||||
sentPreviousResponseId = fallbackChained.previousResponseId;
|
||||
fallbackChained = {
|
||||
...fallbackChained,
|
||||
params: await applyPayloadReplacement(fallbackChained.params),
|
||||
};
|
||||
chained = fallbackChained;
|
||||
activeRawRequestDump.body = chained.params;
|
||||
activeParams = fallbackParams;
|
||||
activeStrictToolsApplied = fallbackBuilt.strictToolsApplied;
|
||||
continue;
|
||||
}
|
||||
if (!chainState || !sentPreviousResponseId || requestSignal.aborted) {
|
||||
throw error;
|
||||
}
|
||||
const zdrRejection =
|
||||
error instanceof Error &&
|
||||
/previous[ _]?response/i.test(error.message) &&
|
||||
/zero[ _-]?data[ _-]?retention/i.test(error.message);
|
||||
const isPromptBlocked =
|
||||
error instanceof Error &&
|
||||
((error as { code?: string }).code === "invalid_prompt" ||
|
||||
/invalid_prompt|Request blocked/i.test(error.message));
|
||||
if (!zdrRejection && !isPromptBlocked && !isOpenAIResponsesStalePreviousResponseError(error)) {
|
||||
throw error;
|
||||
}
|
||||
// Server rejected the chain baseline: reset, count the failure (or
|
||||
// disable categorically on ZDR), and retry once with the full
|
||||
// transcript. Structurally cannot loop — the retry carries no
|
||||
// previous_response_id.
|
||||
if (zdrRejection) {
|
||||
markOpenAIResponsesChainZeroDataRetention(chainState, error);
|
||||
// ZDR orgs cannot store responses; the retry uses `store: false`.
|
||||
} else {
|
||||
registerOpenAIResponsesChainStaleFailure(chainState, error);
|
||||
}
|
||||
sentPreviousResponseId = undefined;
|
||||
const currentBuilt = buildParams(
|
||||
model,
|
||||
context,
|
||||
options,
|
||||
providerSessionState,
|
||||
strictToolsScope,
|
||||
true,
|
||||
forceDisableStrictTools,
|
||||
);
|
||||
const fallbackParams = fallbackBuilt.params;
|
||||
if (chainState && !chainState.disabled) fallbackParams.store = true;
|
||||
let fallbackChained: OpenAIResponsesChainedParams =
|
||||
chainState && !chainState.disabled
|
||||
? buildOpenAIResponsesChainedParams(fallbackParams, chainState)
|
||||
: { params: fallbackParams };
|
||||
sentPreviousResponseId = fallbackChained.previousResponseId;
|
||||
fallbackChained = {
|
||||
...fallbackChained,
|
||||
params: await applyPayloadReplacement(fallbackChained.params),
|
||||
};
|
||||
chained = fallbackChained;
|
||||
rawRequestDump.body = chained.params;
|
||||
activeParams = fallbackParams;
|
||||
activeStrictToolsApplied = fallbackBuilt.strictToolsApplied;
|
||||
continue;
|
||||
const currentParams = currentBuilt.params;
|
||||
// Only ZDR forces `store: false` (the org never persists responses). A
|
||||
// non-ZDR stale baseline is transient, so keep storing: the full-context
|
||||
// retry must be chainable next turn, and the consecutive stale-failure
|
||||
// breaker only trips when each retry stores and the next turn re-chains.
|
||||
currentParams.store = !zdrRejection;
|
||||
const retryParams = await applyPayloadReplacement(currentParams);
|
||||
chained = { params: retryParams };
|
||||
activeRawRequestDump.body = retryParams;
|
||||
activeParams = currentParams;
|
||||
activeStrictToolsApplied = currentBuilt.strictToolsApplied;
|
||||
}
|
||||
if (!chainState || !sentPreviousResponseId || requestSignal.aborted) {
|
||||
throw error;
|
||||
}
|
||||
const zdrRejection =
|
||||
error instanceof Error &&
|
||||
/previous[ _]?response/i.test(error.message) &&
|
||||
/zero[ _-]?data[ _-]?retention/i.test(error.message);
|
||||
const isPromptBlocked =
|
||||
error instanceof Error &&
|
||||
((error as { code?: string }).code === "invalid_prompt" ||
|
||||
/invalid_prompt|Request blocked/i.test(error.message));
|
||||
if (!zdrRejection && !isPromptBlocked && !isOpenAIResponsesStalePreviousResponseError(error)) {
|
||||
throw error;
|
||||
}
|
||||
// Server rejected the chain baseline: reset, count the failure (or
|
||||
// disable categorically on ZDR), and retry once with the full
|
||||
// transcript. Structurally cannot loop — the retry carries no
|
||||
// previous_response_id.
|
||||
if (zdrRejection) {
|
||||
markOpenAIResponsesChainZeroDataRetention(chainState, error);
|
||||
// ZDR orgs cannot store responses; the retry uses `store: false`.
|
||||
} else {
|
||||
registerOpenAIResponsesChainStaleFailure(chainState, error);
|
||||
}
|
||||
sentPreviousResponseId = undefined;
|
||||
const currentBuilt = buildParams(
|
||||
model,
|
||||
context,
|
||||
options,
|
||||
providerSessionState,
|
||||
strictToolsScope,
|
||||
forceDisableStrictTools,
|
||||
);
|
||||
const currentParams = currentBuilt.params;
|
||||
// Only ZDR forces `store: false` (the org never persists responses). A
|
||||
// non-ZDR stale baseline is transient, so keep storing: the full-context
|
||||
// retry must be chainable next turn, and the consecutive stale-failure
|
||||
// breaker only trips when each retry stores and the next turn re-chains.
|
||||
currentParams.store = !zdrRejection;
|
||||
const retryParams = await applyPayloadReplacement(currentParams);
|
||||
chained = { params: retryParams };
|
||||
rawRequestDump.body = retryParams;
|
||||
activeParams = currentParams;
|
||||
activeStrictToolsApplied = currentBuilt.strictToolsApplied;
|
||||
}
|
||||
}
|
||||
return openaiStream;
|
||||
};
|
||||
let openaiStream = await openResponsesStreamWithFallbacks();
|
||||
if (premiumRequestsTotal !== undefined) output.usage.premiumRequests = premiumRequestsTotal;
|
||||
stream.push({ type: "start", partial: output });
|
||||
|
||||
const nativeOutputItems: Array<Record<string, unknown>> = [];
|
||||
let sawTerminalResponseEvent = false;
|
||||
const timedOpenaiStream = iterateWithIdleTimeout(openaiStream, {
|
||||
idleTimeoutMs,
|
||||
firstItemTimeoutMs: firstEventTimeoutMs,
|
||||
firstItemErrorMessage: OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE,
|
||||
errorMessage: "OpenAI responses stream stalled while waiting for the next event",
|
||||
onFirstItemTimeout: () => abortTracker.abortLocally(firstEventTimeoutAbortError),
|
||||
onIdle: () => requestAbortController.abort(),
|
||||
abortSignal: options?.signal,
|
||||
isProgressItem: isOpenAIResponsesProgressEvent,
|
||||
});
|
||||
await processResponsesStream(timedOpenaiStream, output, stream, model, {
|
||||
onFirstToken: () => {
|
||||
if (!firstTokenTime) firstTokenTime = performance.now();
|
||||
},
|
||||
onOutputItemDone: item => {
|
||||
// `processResponsesStream` hands over a private clone already; no
|
||||
// second deep copy needed (reasoning items carry multi-KB blobs).
|
||||
nativeOutputItems.push(item as unknown as Record<string, unknown>);
|
||||
},
|
||||
onCompleted: () => {
|
||||
sawTerminalResponseEvent = true;
|
||||
},
|
||||
requestServiceTier: options?.serviceTier,
|
||||
});
|
||||
|
||||
const localAbortReason = abortTracker.getLocalAbortReason();
|
||||
if (localAbortReason) {
|
||||
throw localAbortReason;
|
||||
}
|
||||
if (abortTracker.wasCallerAbort()) {
|
||||
throw new AIError.AbortError();
|
||||
}
|
||||
|
||||
// Detect premature stream closure: the HTTP stream ended without the
|
||||
// provider sending a recognized terminal response event.
|
||||
// Custom/proxy providers may drop the connection mid-stream; without
|
||||
// this guard the incomplete output is silently surfaced as a successful
|
||||
// "stop".
|
||||
if (!sawTerminalResponseEvent) {
|
||||
throw new AIError.ProviderResponseError(
|
||||
"OpenAI responses stream closed before a terminal response event was received",
|
||||
{ provider: model.provider, kind: "incomplete-stream" },
|
||||
);
|
||||
}
|
||||
|
||||
if (output.stopReason === "aborted" || output.stopReason === "error") {
|
||||
throw new AIError.ProviderResponseError(output.errorMessage ?? "An unknown error occurred", {
|
||||
provider: model.provider,
|
||||
kind: "runtime",
|
||||
let transientStreamRetryAttempt = 0;
|
||||
while (true) {
|
||||
let sawReplayUnsafeOutput = false;
|
||||
let sawTerminalResponseEvent = false;
|
||||
const attemptStream = new AssistantMessageEventStream();
|
||||
let forwardAttemptLive = false;
|
||||
const forwardAttemptEvents = () => {
|
||||
for (const event of attemptStream.queue) stream.push(event);
|
||||
attemptStream.queue.length = 0;
|
||||
};
|
||||
nativeOutputItems.length = 0;
|
||||
const timedOpenaiStream = iterateWithIdleTimeout(openaiStream, {
|
||||
idleTimeoutMs,
|
||||
firstItemTimeoutMs: firstEventTimeoutMs,
|
||||
firstItemErrorMessage: OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE,
|
||||
errorMessage: "OpenAI responses stream stalled while waiting for the next event",
|
||||
onFirstItemTimeout: () => abortTracker.abortLocally(firstEventTimeoutAbortError),
|
||||
onIdle: () => requestAbortController.abort(),
|
||||
abortSignal: options?.signal,
|
||||
isProgressItem: isOpenAIResponsesProgressEvent,
|
||||
});
|
||||
const observedOpenaiStream = (async function* (): AsyncGenerator<ResponseStreamEvent> {
|
||||
for await (const event of timedOpenaiStream) {
|
||||
if (isOpenAIResponsesReplayUnsafeEvent(event)) {
|
||||
sawReplayUnsafeOutput = true;
|
||||
if (!forwardAttemptLive) {
|
||||
forwardAttemptEvents();
|
||||
forwardAttemptLive = true;
|
||||
}
|
||||
}
|
||||
yield event;
|
||||
if (forwardAttemptLive) forwardAttemptEvents();
|
||||
}
|
||||
})();
|
||||
|
||||
try {
|
||||
await processResponsesStream(observedOpenaiStream, output, attemptStream, model, {
|
||||
onFirstToken: () => {
|
||||
if (!firstTokenTime) firstTokenTime = performance.now();
|
||||
},
|
||||
onOutputItemDone: item => {
|
||||
// `processResponsesStream` hands over a private clone already; no
|
||||
// second deep copy needed (reasoning items carry multi-KB blobs).
|
||||
nativeOutputItems.push(item as unknown as Record<string, unknown>);
|
||||
},
|
||||
onCompleted: () => {
|
||||
sawTerminalResponseEvent = true;
|
||||
},
|
||||
requestServiceTier: options?.serviceTier,
|
||||
});
|
||||
|
||||
const localAbortReason = abortTracker.getLocalAbortReason();
|
||||
if (localAbortReason) throw localAbortReason;
|
||||
if (abortTracker.wasCallerAbort()) throw new AIError.AbortError();
|
||||
|
||||
// Detect premature stream closure: the HTTP stream ended without the
|
||||
// provider sending a recognized terminal response event.
|
||||
if (!sawTerminalResponseEvent) {
|
||||
throw new AIError.ProviderResponseError(
|
||||
"OpenAI responses stream closed before a terminal response event was received",
|
||||
{ provider: model.provider, kind: "incomplete-stream" },
|
||||
);
|
||||
}
|
||||
|
||||
if (output.stopReason === "aborted" || output.stopReason === "error") {
|
||||
throw new AIError.ProviderResponseError(output.errorMessage ?? "An unknown error occurred", {
|
||||
provider: model.provider,
|
||||
kind: "runtime",
|
||||
});
|
||||
}
|
||||
forwardAttemptEvents();
|
||||
break;
|
||||
} catch (error) {
|
||||
const streamFailure = abortTracker.getLocalAbortReason() ?? error;
|
||||
const canRetry =
|
||||
!sawReplayUnsafeOutput &&
|
||||
!requestSignal.aborted &&
|
||||
!abortTracker.wasCallerAbort() &&
|
||||
transientStreamRetryAttempt < OPENAI_RESPONSES_MAX_TRANSIENT_STREAM_RETRIES &&
|
||||
isRetryableOpenAIResponsesStreamFailure(streamFailure);
|
||||
if (!canRetry) {
|
||||
forwardAttemptEvents();
|
||||
throw streamFailure;
|
||||
}
|
||||
|
||||
transientStreamRetryAttempt++;
|
||||
logger.debug("OpenAI responses stream ended before replay-unsafe output; retrying", {
|
||||
provider: model.provider,
|
||||
model: model.id,
|
||||
attempt: transientStreamRetryAttempt,
|
||||
error: streamFailure instanceof Error ? streamFailure.message : String(streamFailure),
|
||||
});
|
||||
const retryOutput = createInitialResponsesAssistantMessage(model.api, model.provider, model.id);
|
||||
output.content.length = 0;
|
||||
output.responseId = undefined;
|
||||
output.upstreamProvider = undefined;
|
||||
output.errorMessage = undefined;
|
||||
output.errorStatus = undefined;
|
||||
output.errorId = undefined;
|
||||
output.stopDetails = undefined;
|
||||
output.providerPayload = undefined;
|
||||
output.usage = retryOutput.usage;
|
||||
if (premiumRequestsTotal !== undefined) output.usage.premiumRequests = premiumRequestsTotal;
|
||||
output.stopReason = "stop";
|
||||
output.duration = undefined;
|
||||
output.ttft = undefined;
|
||||
firstTokenTime = undefined;
|
||||
nativeOutputItems.length = 0;
|
||||
|
||||
if (options?.providerRetryWait) {
|
||||
await options.providerRetryWait(OPENAI_RESPONSES_TRANSIENT_STREAM_RETRY_DELAY_MS, options.signal);
|
||||
} else {
|
||||
await scheduler.wait(OPENAI_RESPONSES_TRANSIENT_STREAM_RETRY_DELAY_MS, { signal: options?.signal });
|
||||
}
|
||||
if (abortTracker.wasCallerAbort()) throw new AIError.AbortError();
|
||||
openaiStream = await openResponsesStreamWithFallbacks();
|
||||
}
|
||||
}
|
||||
|
||||
output.providerPayload = createOpenAIResponsesHistoryPayload(model.provider, nativeOutputItems);
|
||||
|
||||
@@ -0,0 +1,564 @@
|
||||
import { describe, expect, it, vi } from "bun:test";
|
||||
import { streamOpenAIResponses } from "@oh-my-pi/pi-ai/providers/openai-responses";
|
||||
import type {
|
||||
AssistantMessageEvent,
|
||||
AssistantMessageEventStream,
|
||||
Context,
|
||||
FetchImpl,
|
||||
Model,
|
||||
ProviderSessionState,
|
||||
} from "@oh-my-pi/pi-ai/types";
|
||||
import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
|
||||
|
||||
const model = getBundledModel("openai", "gpt-5-mini") as Model<"openai-responses">;
|
||||
const firstUser = { role: "user" as const, content: "Read the file", timestamp: 1_000 };
|
||||
const context: Context = { messages: [firstUser] };
|
||||
|
||||
function createSseResponse(events: unknown[]): Response {
|
||||
return new Response(`${events.map(event => `data: ${JSON.stringify(event)}`).join("\n\n")}\n\n`, {
|
||||
status: 200,
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
});
|
||||
}
|
||||
|
||||
function createTruncatedPendingToolResponse(): Response {
|
||||
const prefix = [
|
||||
{ type: "response.created", response: { id: "resp_partial", status: "in_progress" } },
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: {
|
||||
type: "function_call",
|
||||
id: "fc_partial",
|
||||
call_id: "call_partial",
|
||||
name: "read",
|
||||
arguments: "",
|
||||
status: "in_progress",
|
||||
},
|
||||
},
|
||||
];
|
||||
const truncatedEvent = 'data: {"type":"response.function_call_arguments.delta","item_id":"fc_partial","delta":';
|
||||
return new Response(`${prefix.map(event => `data: ${JSON.stringify(event)}`).join("\n\n")}\n\n${truncatedEvent}`, {
|
||||
status: 200,
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
});
|
||||
}
|
||||
|
||||
function createTruncatedReasoningPartDoneResponse(): Response {
|
||||
const prefix = [
|
||||
{ type: "response.created", response: { id: "resp_reasoning", status: "in_progress" } },
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: { type: "reasoning", id: "reasoning_partial", summary: [], status: "in_progress" },
|
||||
},
|
||||
{
|
||||
type: "response.reasoning_summary_part.added",
|
||||
item_id: "reasoning_partial",
|
||||
output_index: 0,
|
||||
summary_index: 0,
|
||||
part: { type: "summary_text", text: "" },
|
||||
},
|
||||
{
|
||||
type: "response.reasoning_summary_part.done",
|
||||
item_id: "reasoning_partial",
|
||||
output_index: 0,
|
||||
summary_index: 0,
|
||||
part: { type: "summary_text", text: "" },
|
||||
},
|
||||
];
|
||||
const truncatedEvent = 'data: {"type":"response.output_text.delta","item_id":"missing","delta":';
|
||||
return new Response(`${prefix.map(event => `data: ${JSON.stringify(event)}`).join("\n\n")}\n\n${truncatedEvent}`, {
|
||||
status: 200,
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
});
|
||||
}
|
||||
|
||||
function createCompletedToolResponse(responseId = "resp_retry"): Response {
|
||||
const argumentsJson = JSON.stringify({ path: "README.md" });
|
||||
return createSseResponse([
|
||||
{ type: "response.created", response: { id: responseId, status: "in_progress" } },
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: {
|
||||
type: "function_call",
|
||||
id: "fc_recovered",
|
||||
call_id: "call_recovered",
|
||||
name: "read",
|
||||
arguments: "",
|
||||
status: "in_progress",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.function_call_arguments.delta",
|
||||
output_index: 0,
|
||||
item_id: "fc_recovered",
|
||||
delta: argumentsJson,
|
||||
},
|
||||
{
|
||||
type: "response.function_call_arguments.done",
|
||||
output_index: 0,
|
||||
item_id: "fc_recovered",
|
||||
arguments: argumentsJson,
|
||||
},
|
||||
{
|
||||
type: "response.output_item.done",
|
||||
output_index: 0,
|
||||
item: {
|
||||
type: "function_call",
|
||||
id: "fc_recovered",
|
||||
call_id: "call_recovered",
|
||||
name: "read",
|
||||
arguments: argumentsJson,
|
||||
status: "completed",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.completed",
|
||||
response: {
|
||||
id: responseId,
|
||||
status: "completed",
|
||||
usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8, input_tokens_details: { cached_tokens: 0 } },
|
||||
},
|
||||
},
|
||||
]);
|
||||
}
|
||||
|
||||
function createCompletedTextResponse(text: string, responseId: string): Response {
|
||||
return createSseResponse([
|
||||
{ type: "response.created", response: { id: responseId, status: "in_progress" } },
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: { type: "message", id: `msg_${responseId}`, role: "assistant", status: "in_progress", content: [] },
|
||||
},
|
||||
{ type: "response.output_text.delta", output_index: 0, item_id: `msg_${responseId}`, delta: text },
|
||||
{
|
||||
type: "response.output_item.done",
|
||||
output_index: 0,
|
||||
item: {
|
||||
type: "message",
|
||||
id: `msg_${responseId}`,
|
||||
role: "assistant",
|
||||
status: "completed",
|
||||
content: [{ type: "output_text", text }],
|
||||
},
|
||||
},
|
||||
{ type: "response.completed", response: { id: responseId, status: "completed" } },
|
||||
]);
|
||||
}
|
||||
|
||||
function createGatedTextAndToolResponse(): {
|
||||
response: Response;
|
||||
terminalRequested: Promise<void>;
|
||||
releaseTerminal: () => void;
|
||||
} {
|
||||
const nonTerminalEvents = [
|
||||
{ type: "response.created", response: { id: "resp_live", status: "in_progress" } },
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: { type: "message", id: "msg_live", role: "assistant", status: "in_progress", content: [] },
|
||||
},
|
||||
{ type: "response.output_text.delta", output_index: 0, item_id: "msg_live", delta: "draft" },
|
||||
{
|
||||
type: "response.output_item.done",
|
||||
output_index: 0,
|
||||
item: {
|
||||
type: "message",
|
||||
id: "msg_live",
|
||||
role: "assistant",
|
||||
status: "completed",
|
||||
content: [{ type: "output_text", text: "draft" }],
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 1,
|
||||
item: {
|
||||
type: "function_call",
|
||||
id: "fc_live",
|
||||
call_id: "call_live",
|
||||
name: "read",
|
||||
arguments: "",
|
||||
status: "in_progress",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.function_call_arguments.delta",
|
||||
output_index: 1,
|
||||
item_id: "fc_live",
|
||||
delta: '{"path":"README',
|
||||
},
|
||||
];
|
||||
const terminalEvents = [
|
||||
{
|
||||
type: "response.function_call_arguments.done",
|
||||
output_index: 1,
|
||||
item_id: "fc_live",
|
||||
arguments: '{"path":"README.md"}',
|
||||
},
|
||||
{
|
||||
type: "response.output_item.done",
|
||||
output_index: 1,
|
||||
item: {
|
||||
type: "function_call",
|
||||
id: "fc_live",
|
||||
call_id: "call_live",
|
||||
name: "read",
|
||||
arguments: '{"path":"README.md"}',
|
||||
status: "completed",
|
||||
},
|
||||
},
|
||||
{ type: "response.completed", response: { id: "resp_live", status: "completed" } },
|
||||
];
|
||||
const terminalGate = Promise.withResolvers<void>();
|
||||
const terminalRequest = Promise.withResolvers<void>();
|
||||
let sentNonTerminal = false;
|
||||
const body = new ReadableStream<Uint8Array>(
|
||||
{
|
||||
async pull(controller) {
|
||||
if (!sentNonTerminal) {
|
||||
sentNonTerminal = true;
|
||||
controller.enqueue(
|
||||
new TextEncoder().encode(
|
||||
`${nonTerminalEvents.map(event => `data: ${JSON.stringify(event)}`).join("\n\n")}\n\n`,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
terminalRequest.resolve();
|
||||
await terminalGate.promise;
|
||||
controller.enqueue(
|
||||
new TextEncoder().encode(
|
||||
`${terminalEvents.map(event => `data: ${JSON.stringify(event)}`).join("\n\n")}\n\n`,
|
||||
),
|
||||
);
|
||||
controller.close();
|
||||
},
|
||||
},
|
||||
{ highWaterMark: 0 },
|
||||
);
|
||||
return {
|
||||
response: new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }),
|
||||
terminalRequested: terminalRequest.promise,
|
||||
releaseTerminal: terminalGate.resolve,
|
||||
};
|
||||
}
|
||||
|
||||
function parseBody(init: RequestInit | undefined): Record<string, unknown> {
|
||||
return JSON.parse(String(init?.body)) as Record<string, unknown>;
|
||||
}
|
||||
|
||||
async function collectEvents(stream: AssistantMessageEventStream): Promise<AssistantMessageEvent[]> {
|
||||
const events: AssistantMessageEvent[] = [];
|
||||
for await (const event of stream) events.push(event);
|
||||
return events;
|
||||
}
|
||||
|
||||
describe("OpenAI Responses transient stream retry", () => {
|
||||
it("retries a truncated pending tool call with a fresh request and clean state", async () => {
|
||||
const sentRequests: Array<Record<string, unknown>> = [];
|
||||
let attempt = 0;
|
||||
let payloadCalls = 0;
|
||||
const fetchMock = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => {
|
||||
sentRequests.push(parseBody(init));
|
||||
attempt++;
|
||||
if (attempt === 1) return createTruncatedPendingToolResponse();
|
||||
if (attempt === 2) return createCompletedToolResponse();
|
||||
return createCompletedTextResponse("Follow-up", "resp_followup");
|
||||
}) as FetchImpl;
|
||||
const providerSessionState = new Map<string, ProviderSessionState>();
|
||||
const options = {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
providerRetryWait: async () => {},
|
||||
providerSessionState,
|
||||
sessionId: "stream-retry-session",
|
||||
statefulResponses: true,
|
||||
onPayload: (payload: unknown) => {
|
||||
payloadCalls++;
|
||||
return { ...(payload as Record<string, unknown>), metadata: { retry_test: "preserved" } };
|
||||
},
|
||||
};
|
||||
|
||||
const responseStream = streamOpenAIResponses(model, context, options);
|
||||
const events = await collectEvents(responseStream);
|
||||
const result = await responseStream.result();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
expect(payloadCalls).toBe(1);
|
||||
expect(sentRequests[0]?.metadata).toEqual({ retry_test: "preserved" });
|
||||
expect(sentRequests[1]).toEqual(sentRequests[0]);
|
||||
expect(result.stopReason).toBe("toolUse");
|
||||
expect(JSON.parse(JSON.stringify(result.content))).toEqual([
|
||||
{ type: "toolCall", id: "call_recovered|fc_recovered", name: "read", arguments: { path: "README.md" } },
|
||||
]);
|
||||
expect(events.map(event => event.type)).toEqual([
|
||||
"start",
|
||||
"toolcall_start",
|
||||
"toolcall_delta",
|
||||
"toolcall_end",
|
||||
"done",
|
||||
]);
|
||||
expect(JSON.stringify(result.providerPayload)).not.toContain("partial");
|
||||
|
||||
const followup = await streamOpenAIResponses(
|
||||
model,
|
||||
{
|
||||
messages: [firstUser, result, { role: "user", content: "What did it contain?", timestamp: 1_001 }],
|
||||
},
|
||||
options,
|
||||
).result();
|
||||
expect(followup.stopReason).toBe("stop");
|
||||
expect(sentRequests[2]?.previous_response_id).toBe("resp_retry");
|
||||
});
|
||||
|
||||
it("falls back to full transcript when a fresh stream retry finds a stale chain baseline", async () => {
|
||||
const sentRequests: Array<Record<string, unknown>> = [];
|
||||
const fetchMock = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => {
|
||||
const request = parseBody(init);
|
||||
sentRequests.push(request);
|
||||
switch (sentRequests.length) {
|
||||
case 1:
|
||||
return createCompletedTextResponse("Baseline", "resp_baseline");
|
||||
case 2:
|
||||
return createTruncatedPendingToolResponse();
|
||||
case 3:
|
||||
return new Response(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
message: "Previous response with id 'resp_baseline' not found.",
|
||||
type: "invalid_request_error",
|
||||
param: "previous_response_id",
|
||||
code: "previous_response_not_found",
|
||||
},
|
||||
}),
|
||||
{ status: 404, headers: { "content-type": "application/json" } },
|
||||
);
|
||||
case 4:
|
||||
return createCompletedTextResponse("Recovered", "resp_recovered");
|
||||
default:
|
||||
return createCompletedTextResponse("Follow-up", "resp_followup");
|
||||
}
|
||||
}) as FetchImpl;
|
||||
const providerSessionState = new Map<string, ProviderSessionState>();
|
||||
const options = {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
providerRetryWait: async () => {},
|
||||
providerSessionState,
|
||||
sessionId: "stream-retry-stale-chain-session",
|
||||
statefulResponses: true,
|
||||
};
|
||||
|
||||
const baseline = await streamOpenAIResponses(model, context, options).result();
|
||||
const secondUser = { role: "user" as const, content: "Continue after baseline", timestamp: 1_001 };
|
||||
const responseStream = streamOpenAIResponses(model, { messages: [firstUser, baseline, secondUser] }, options);
|
||||
const events = await collectEvents(responseStream);
|
||||
const recovered = await responseStream.result();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(4);
|
||||
expect(sentRequests[0]?.previous_response_id).toBeUndefined();
|
||||
expect(sentRequests[1]?.previous_response_id).toBe("resp_baseline");
|
||||
expect(sentRequests[2]).toEqual(sentRequests[1]);
|
||||
expect(JSON.stringify(sentRequests[1]?.input)).toContain("Continue after baseline");
|
||||
expect(JSON.stringify(sentRequests[1]?.input)).not.toContain("Read the file");
|
||||
expect(sentRequests[3]?.previous_response_id).toBeUndefined();
|
||||
expect(sentRequests[3]?.store).toBe(true);
|
||||
expect(JSON.stringify(sentRequests[3]?.input)).toContain("Read the file");
|
||||
expect(JSON.stringify(sentRequests[3]?.input)).toContain("Baseline");
|
||||
expect(JSON.stringify(sentRequests[3]?.input)).toContain("Continue after baseline");
|
||||
expect(recovered.responseId).toBe("resp_recovered");
|
||||
expect(events.map(event => event.type)).toEqual(["start", "text_start", "text_delta", "text_end", "done"]);
|
||||
|
||||
const followup = await streamOpenAIResponses(
|
||||
model,
|
||||
{
|
||||
messages: [
|
||||
firstUser,
|
||||
baseline,
|
||||
secondUser,
|
||||
recovered,
|
||||
{ role: "user", content: "One more question", timestamp: 1_002 },
|
||||
],
|
||||
},
|
||||
options,
|
||||
).result();
|
||||
expect(followup.stopReason).toBe("stop");
|
||||
expect(fetchMock).toHaveBeenCalledTimes(5);
|
||||
expect(sentRequests[4]?.previous_response_id).toBe("resp_recovered");
|
||||
});
|
||||
|
||||
it("forwards text and tool deltas live with their delta-time partial state", async () => {
|
||||
const gated = createGatedTextAndToolResponse();
|
||||
const fetchMock = vi.fn(async () => gated.response) as FetchImpl;
|
||||
const responseStream = streamOpenAIResponses(model, context, {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
providerRetryWait: async () => {},
|
||||
});
|
||||
const nonTerminalEvents = (async () => {
|
||||
const observed: Array<{ type: AssistantMessageEvent["type"]; content: unknown }> = [];
|
||||
for await (const event of responseStream) {
|
||||
observed.push({
|
||||
type: event.type,
|
||||
content:
|
||||
event.type !== "start" && "partial" in event ? structuredClone(event.partial.content) : undefined,
|
||||
});
|
||||
if (event.type === "toolcall_delta") return observed;
|
||||
}
|
||||
throw new Error("stream ended before the tool delta");
|
||||
})();
|
||||
|
||||
await gated.terminalRequested;
|
||||
const observedBeforeTerminal = await nonTerminalEvents;
|
||||
const deltaText = { type: "text", text: "draft", textSignature: JSON.stringify({ v: 1, id: "msg_live" }) };
|
||||
expect(observedBeforeTerminal).toEqual([
|
||||
{ type: "start", content: undefined },
|
||||
{ type: "text_start", content: [deltaText] },
|
||||
{ type: "text_delta", content: [deltaText] },
|
||||
{ type: "text_end", content: [deltaText] },
|
||||
{
|
||||
type: "toolcall_start",
|
||||
content: [deltaText, { type: "toolCall", id: "call_live|fc_live", name: "read", arguments: {} }],
|
||||
},
|
||||
{
|
||||
type: "toolcall_delta",
|
||||
content: [
|
||||
deltaText,
|
||||
{ type: "toolCall", id: "call_live|fc_live", name: "read", arguments: { path: "README" } },
|
||||
],
|
||||
},
|
||||
]);
|
||||
|
||||
gated.releaseTerminal();
|
||||
const result = await responseStream.result();
|
||||
expect(result.stopReason).toBe("toolUse");
|
||||
expect(JSON.parse(JSON.stringify(result.content[1]))).toEqual({
|
||||
type: "toolCall",
|
||||
id: "call_live|fc_live",
|
||||
name: "read",
|
||||
arguments: { path: "README.md" },
|
||||
});
|
||||
});
|
||||
|
||||
it("does not retry after a tool argument delta was emitted", async () => {
|
||||
const partialWithDelta = createSseResponse([
|
||||
{ type: "response.created", response: { id: "resp_partial", status: "in_progress" } },
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: {
|
||||
type: "function_call",
|
||||
id: "fc_partial",
|
||||
call_id: "call_partial",
|
||||
name: "read",
|
||||
arguments: "",
|
||||
status: "in_progress",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.function_call_arguments.delta",
|
||||
output_index: 0,
|
||||
item_id: "fc_partial",
|
||||
delta: '{"path":"README.md"}',
|
||||
},
|
||||
]);
|
||||
const fetchMock = vi.fn(async () => partialWithDelta) as FetchImpl;
|
||||
|
||||
const result = await streamOpenAIResponses(model, context, {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
providerRetryWait: async () => {},
|
||||
}).result();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(result.stopReason).toBe("error");
|
||||
});
|
||||
|
||||
it("does not retry after a reasoning summary part completion emitted thinking", async () => {
|
||||
const fetchMock = vi.fn(async () => createTruncatedReasoningPartDoneResponse()) as FetchImpl;
|
||||
const responseStream = streamOpenAIResponses(model, context, {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
providerRetryWait: async () => {},
|
||||
});
|
||||
|
||||
const events = await collectEvents(responseStream);
|
||||
const result = await responseStream.result();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(events.map(event => event.type)).toEqual(["start", "thinking_start", "thinking_delta", "error"]);
|
||||
expect(result.stopReason).toBe("error");
|
||||
});
|
||||
|
||||
it("bounds repeated pre-output stream corruption to one retry", async () => {
|
||||
const fetchMock = vi.fn(async () => createTruncatedPendingToolResponse()) as FetchImpl;
|
||||
|
||||
const result = await streamOpenAIResponses(model, context, {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
providerRetryWait: async () => {},
|
||||
}).result();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
expect(result.stopReason).toBe("error");
|
||||
});
|
||||
|
||||
it("honors caller abort during the retry wait", async () => {
|
||||
const controller = new AbortController();
|
||||
const fetchMock = vi.fn(async () => createTruncatedPendingToolResponse()) as FetchImpl;
|
||||
|
||||
const responseStream = streamOpenAIResponses(model, context, {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
signal: controller.signal,
|
||||
providerRetryWait: async () => controller.abort(),
|
||||
});
|
||||
const events = await collectEvents(responseStream);
|
||||
const result = await responseStream.result();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(result.stopReason).toBe("aborted");
|
||||
expect(events.map(event => event.type)).toEqual(["start", "error"]);
|
||||
expect(result.content).toEqual([]);
|
||||
expect(result.responseId).toBeUndefined();
|
||||
expect(result.providerPayload).toBeUndefined();
|
||||
expect(result.usage).toEqual({
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 0,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
});
|
||||
expect(result.ttft).toBeUndefined();
|
||||
});
|
||||
|
||||
for (const error of [
|
||||
{ code: "invalid_request_error", message: "Tool schema is invalid" },
|
||||
{ code: "insufficient_quota", message: "Persistent quota exhausted" },
|
||||
]) {
|
||||
it(`does not retry terminal ${error.code} failures`, async () => {
|
||||
const fetchMock = vi.fn(async () =>
|
||||
createSseResponse([
|
||||
{
|
||||
type: "response.failed",
|
||||
response: { id: "resp_failed", status: "failed", error },
|
||||
},
|
||||
]),
|
||||
) as FetchImpl;
|
||||
|
||||
const result = await streamOpenAIResponses(model, context, {
|
||||
apiKey: "test-key",
|
||||
fetch: fetchMock,
|
||||
providerRetryWait: async () => {},
|
||||
}).result();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(result.stopReason).toBe("error");
|
||||
});
|
||||
}
|
||||
});
|
||||
Reference in New Issue
Block a user