vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-28 11:51:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: Responses streams mishandle interleaved output items and silent truncation (#109615)

Browse Source 
* fix(ai): track Responses output items per index and require terminal stream events

* test(ai): add live Responses stream coverage

* test(infra): mirror packages live glob in live-config test
This commit is contained in:
Peter Steinberger
2026-07-16 23:15:35 -07:00
committed by GitHub
parent 45ab24dac4
commit 03d287d111
11 changed files with 1220 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
packages/ai/src/providers/azure-openai-responses.test.ts
View File

@@ -104,4 +104,24 @@ describe("azure-openai-responses", () => {
}
}
});
it("disables response storage and clamps small output limits", async () => {
let sentParams: { max_output_tokens?: unknown; store?: unknown } | undefined;
const hostFetch: typeof fetch = async (input, init) => {
sentParams = (await new Request(input, init).json()) as typeof sentParams;
return Response.json({ error: { message: "captured" } }, { status: 400 });
};
configureAiTransportHost({ buildModelFetch: () => hostFetch });
try {
await streamSimpleAzureOpenAIResponses(azureResponsesModel, context, {
apiKey: "test-api-key",
maxTokens: 1,
}).result();
expect(sentParams).toMatchObject({ max_output_tokens: 16, store: false });
} finally {
configureAiTransportHost({});
}
});
});

1
packages/ai/src/providers/azure-openai-responses.ts
View File

@@ -239,6 +239,7 @@ function buildParams(
options?.cacheRetention === "none"
? undefined
: clampOpenAIPromptCacheKey(options?.promptCacheKey ?? options?.sessionId),
store: false,
};
applyCommonResponsesParams(params, model, context, options);

44
packages/ai/src/providers/openai-chatgpt-responses.test.ts
View File

@@ -160,6 +160,28 @@ describe("streamOpenAICodexResponses transport", () => {
expect(providerToken).toBe(`Bearer ${realToken}`);
});
it("clamps session headers to the backend limit", async () => {
const jwt = createJwt({ "https://api.openai.com/auth": { chatgpt_account_id: "acct" } });
let headers: Headers | undefined;
vi.stubGlobal(
"fetch",
vi.fn(async (_input, init) => {
headers = new Headers(init?.headers);
return completedSseResponse();
}),
);
const result = await streamOpenAICodexResponses(model, context, {
apiKey: jwt,
sessionId: "s".repeat(80),
transport: "sse",
}).result();
expect(result.stopReason).toBe("stop");
expect(headers?.get("session_id")).toBe("s".repeat(64));
expect(headers?.get("x-client-request-id")).toBe("s".repeat(64));
});
it("builds the first Node request with an OS-specific user agent", async () => {
vi.resetModules();
const freshProvider = await import("./openai-chatgpt-responses.js");
@@ -822,6 +844,28 @@ describe("streamOpenAICodexResponses transport", () => {
expect(payload).toMatchObject({ prompt_cache_key: "stable-cache-key" });
});
it("does not retry the ChatGPT transport when maxRetries is zero", async () => {
const jwt = createJwt({ "https://api.openai.com/auth": { chatgpt_account_id: "acct" } });
const fetchMock = vi.fn<typeof fetch>().mockResolvedValue(
new Response("rate limited", {
status: 429,
headers: { "retry-after": "1" },
}),
);
vi.stubGlobal("fetch", fetchMock);
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
const result = await streamOpenAICodexResponses(model, context, {
apiKey: jwt,
maxRetries: 0,
transport: "sse",
}).result();
expect(result.stopReason).toBe("error");
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(setTimeoutSpy).not.toHaveBeenCalled();
});
it.each([
"1.5",
"0x10",

20
packages/ai/src/providers/openai-chatgpt-responses.ts
View File

@@ -75,7 +75,7 @@ import { buildBaseOptions } from "./simple-options.js";
// ============================================================================
const DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api";
const MAX_RETRIES = 3;
const DEFAULT_MAX_RETRIES = 3;
const BASE_DELAY_MS = 1000;
const REQUEST_COMPRESSION_ZSTD_LEVEL = 3;
const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "opencode"]);
@@ -271,14 +271,9 @@ export const streamOpenAICodexResponses: StreamFunction<
// per request, which forfeits session-affinity routing on the WS transport (the
// backend routes by session_id/x-client-request-id). Left as-is for this fix;
// see the SSE-path session_id addition in buildOpenAIClientHeaders (agents/openai-transport-stream.ts).
const websocketRequestId = options?.sessionId || createCodexRequestId();
const sseHeaders = buildSSEHeaders(
modelHeaders,
optionHeaders,
accountId,
apiKey,
options?.sessionId,
);
const sessionId = clampOpenAIPromptCacheKey(options?.sessionId);
const websocketRequestId = sessionId || createCodexRequestId();
const sseHeaders = buildSSEHeaders(modelHeaders, optionHeaders, accountId, apiKey, sessionId);
const websocketHeaders = buildWebSocketHeaders(
modelHeaders,
optionHeaders,
@@ -371,8 +366,9 @@ export const streamOpenAICodexResponses: StreamFunction<
// Fetch with retry logic for rate limits and transient errors
let response: Response | undefined;
let lastError: Error | undefined;
const maxRetries = options?.maxRetries ?? DEFAULT_MAX_RETRIES;
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
for (let attempt = 0; attempt <= maxRetries; attempt++) {
if (activeSignal?.aborted) {
throw new Error("Request was aborted");
}
@@ -394,7 +390,7 @@ export const streamOpenAICodexResponses: StreamFunction<
}
const errorText = await readChatGptResponsesErrorTextLimited(response);
if (attempt < MAX_RETRIES && isRetryableError(response.status, errorText)) {
if (attempt < maxRetries && isRetryableError(response.status, errorText)) {
let delayMs = BASE_DELAY_MS * 2 ** attempt;
const retryAfterMs = response.headers.get("retry-after-ms");
@@ -453,7 +449,7 @@ export const streamOpenAICodexResponses: StreamFunction<
}
lastError = error instanceof Error ? error : new Error(String(error));
// Network errors are retryable
if (attempt < MAX_RETRIES && !lastError.message.includes("usage limit")) {
if (attempt < maxRetries && !lastError.message.includes("usage limit")) {
const delayMs = BASE_DELAY_MS * 2 ** attempt;
await sleepWithAbort(delayMs, activeSignal);
continue;

541
packages/ai/src/providers/openai-responses-shared.test.ts
View File

@@ -1,5 +1,6 @@
// OpenAI Responses shared tests cover tool conversion and response item mapping.
import type {
ResponseCreateParamsStreaming,
ResponseStreamEvent,
Tool as OpenAIResponsesTool,
} from "openai/resources/responses/responses.js";
@@ -22,9 +23,9 @@ type ResponsesFunctionTool = Extract<OpenAIResponsesTool, { type: "function" }>;
type OpenAIResponsesStreamEvent =
Parameters<typeof processResponsesStream>[0] extends AsyncIterable<infer Event> ? Event : never;
async function* streamResponsesEvents(
events: readonly OpenAIResponsesStreamEvent[],
): AsyncGenerator<OpenAIResponsesStreamEvent> {
async function* streamResponsesEvents<T extends OpenAIResponsesStreamEvent>(
events: readonly T[],
): AsyncGenerator<T> {
for (const event of events) {
yield event;
}
@@ -303,6 +304,20 @@ describe("Responses temperature support", () => {
});
});
describe("Responses output token limits", () => {
it.each([
[1, 16],
[15, 16],
[16, 16],
[32, 32],
])("normalizes maxTokens %i to %i", (maxTokens, expected) => {
const params = {} as ResponseCreateParamsStreaming;
applyCommonResponsesParams(params, nativeOpenAIModel, { messages: [] }, { maxTokens });
expect(params.max_output_tokens).toBe(expected);
});
});
describe("Responses reasoning effort", () => {
it("omits unsupported default-off reasoning for GPT-5.6 Sol", () => {
const params = {} as never;
@@ -365,6 +380,21 @@ describe("convertResponsesMessages", () => {
});
});
it("uses the system role when a Responses-compatible provider opts out of developer", () => {
const compatModel = {
...nativeOpenAIModel,
compat: { supportsDeveloperRole: false },
} satisfies Model<"openai-responses">;
const input = convertResponsesMessages(
compatModel,
{ systemPrompt: "system", messages: [] },
allowedToolCallProviders,
);
expect(input[0]).toMatchObject({ type: "message", role: "system" });
});
it("strips the internal cache boundary marker from the system prompt message", () => {
const input = convertResponsesMessages(
nativeOpenAIModel,
@@ -899,6 +929,484 @@ describe("processResponsesStream", () => {
}
});
it.each([
[undefined, 0],
[2, 2],
])("passes SDK maxRetries %s as %i", async (maxRetries, expected) => {
let requestMaxRetries: number | undefined;
const output = createAssistantOutput();
const stream = new AssistantMessageEventStream();
await runResponsesStreamLifecycle({
stream,
model: nativeOpenAIModel,
output,
options: maxRetries === undefined ? undefined : { maxRetries },
createClient: () => ({
responses: {
create: (_params, requestOptions) => {
requestMaxRetries = requestOptions.maxRetries;
return {
withResponse: async () => ({
data: streamResponsesEvents([
{
type: "response.completed",
sequence_number: 1,
response: { id: "resp_retry", status: "completed" },
} as ResponseStreamEvent,
]),
response: new Response(null, { status: 200 }),
}),
};
},
},
}),
buildParams: () => ({ model: nativeOpenAIModel.id, input: [], stream: true }),
formatError: (error) => (error instanceof Error ? error.message : String(error)),
});
expect(requestMaxRetries).toBe(expected);
expect(output.stopReason).toBe("stop");
});
it("keeps interleaved reasoning items bound to their output indices", async () => {
const output = createAssistantOutput();
const { stream, events } = createCapturedAssistantMessageEventStream();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
output_index: 0,
item: { type: "reasoning", id: "rs_first", summary: [] },
},
{
type: "response.output_item.added",
output_index: 1,
item: { type: "reasoning", id: "rs_second", summary: [] },
},
{ type: "response.reasoning_text.delta", output_index: 0, delta: "first" },
{ type: "response.reasoning_text.delta", output_index: 1, delta: "second" },
{
type: "response.output_item.done",
output_index: 1,
item: {
type: "reasoning",
id: "rs_second",
summary: [],
content: [{ type: "reasoning_text", text: "second" }],
encrypted_content: "cipher-second",
},
},
{
type: "response.output_item.done",
output_index: 0,
item: {
type: "reasoning",
id: "rs_first",
summary: [],
content: [{ type: "reasoning_text", text: "first" }],
encrypted_content: "cipher-first",
},
},
{ type: "response.completed", response: { id: "resp_reasoning", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
expect(output.content).toMatchObject([
{ type: "thinking", thinking: "first" },
{ type: "thinking", thinking: "second" },
]);
const signatures = output.content.map((block) =>
block.type === "thinking" && block.thinkingSignature
? JSON.parse(block.thinkingSignature)
: undefined,
);
expect(signatures).toMatchObject([
{ id: "rs_first", encrypted_content: "cipher-first" },
{ id: "rs_second", encrypted_content: "cipher-second" },
]);
expect(
events.map((event) => [event.type, "contentIndex" in event ? event.contentIndex : undefined]),
).toEqual([
["thinking_start", 0],
["thinking_start", 1],
["thinking_delta", 0],
["thinking_delta", 1],
["thinking_end", 1],
["thinking_end", 0],
]);
});
it("preserves reasoning while a tool call streams on another output index", async () => {
const output = createAssistantOutput();
const { stream, events } = createCapturedAssistantMessageEventStream();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
output_index: 0,
item: { type: "reasoning", id: "rs_tool", summary: [] },
},
{ type: "response.reasoning_text.delta", output_index: 0, delta: "before " },
{
type: "response.output_item.added",
output_index: 1,
item: {
type: "function_call",
id: "fc_read",
call_id: "call_read",
name: "read",
arguments: "",
},
},
{
type: "response.function_call_arguments.delta",
output_index: 1,
item_id: "fc_read",
delta: '{"path":"README.md"}',
},
{ type: "response.reasoning_text.delta", output_index: 0, delta: "after" },
{
type: "response.output_item.done",
output_index: 1,
item: {
type: "function_call",
id: "fc_read",
call_id: "call_read",
name: "read",
arguments: '{"path":"README.md"}',
},
},
{
type: "response.output_item.done",
output_index: 0,
item: {
type: "reasoning",
id: "rs_tool",
summary: [],
encrypted_content: "cipher-tool",
},
},
{ type: "response.completed", response: { id: "resp_tool", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
expect(output.content).toMatchObject([
{ type: "thinking", thinking: "before after" },
{ type: "toolCall", id: "call_read|fc_read", arguments: { path: "README.md" } },
]);
expect(
events
.filter((event) => event.type.endsWith("_delta"))
.map((event) => [event.type, "contentIndex" in event ? event.contentIndex : undefined]),
).toEqual([
["thinking_delta", 0],
["toolcall_delta", 1],
["thinking_delta", 0],
]);
});
it("backfills terminal encrypted reasoning for stateless replay", async () => {
const output = createAssistantOutput();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
output_index: 0,
item: { type: "reasoning", id: "rs_backfill", summary: [] },
},
{
type: "response.output_item.done",
output_index: 0,
item: { type: "reasoning", id: "rs_backfill", summary: [] },
},
{
type: "response.completed",
response: {
id: "resp_backfill",
status: "completed",
output: [
{
type: "reasoning",
id: "rs_backfill",
summary: [],
encrypted_content: "cipher-from-terminal",
},
],
},
},
]),
output,
new AssistantMessageEventStream(),
nativeOpenAIModel,
);
const replay = convertResponsesMessages(
nativeOpenAIModel,
{
messages: [
{ role: "user", content: "first", timestamp: 1 },
output,
{ role: "user", content: "again", timestamp: 2 },
],
},
testAllowedToolCallProviders,
);
expect(replay.find((item) => item.type === "reasoning")).toMatchObject({
id: "rs_backfill",
encrypted_content: "cipher-from-terminal",
});
});
it("tolerates a completed message item with null content", async () => {
const output = createAssistantOutput();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
output_index: 0,
item: {
type: "message",
id: "msg_null",
content: [{ type: "output_text", text: "" }],
},
},
{ type: "response.output_text.delta", output_index: 0, delta: "streamed" },
{
type: "response.output_item.done",
output_index: 0,
item: { type: "message", id: "msg_null", content: null },
},
{ type: "response.completed", response: { id: "resp_null", status: "completed" } },
]),
output,
new AssistantMessageEventStream(),
nativeOpenAIModel,
);
expect(output.content).toMatchObject([{ type: "text", text: "streamed" }]);
});
it("keeps deferred text before an interleaved output item", async () => {
const output = createAssistantOutput();
const { stream, events } = createCapturedAssistantMessageEventStream();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
output_index: 0,
item: {
type: "message",
id: "msg_first",
content: [{ type: "output_text", text: "" }],
},
},
{
type: "response.output_item.done",
output_index: 0,
item: {
type: "message",
id: "msg_first",
content: [{ type: "output_text", text: "first" }],
},
},
{
type: "response.output_item.added",
output_index: 1,
item: {
type: "message",
id: "msg_second",
content: [{ type: "output_text", text: "" }],
},
},
{ type: "response.output_text.delta", output_index: 1, delta: "first extended" },
{
type: "response.output_item.added",
output_index: 2,
item: { type: "reasoning", id: "rs_after", summary: [] },
},
{
type: "response.output_item.done",
output_index: 2,
item: {
type: "reasoning",
id: "rs_after",
summary: [],
content: [{ type: "reasoning_text", text: "thought" }],
},
},
{
type: "response.output_item.done",
output_index: 1,
item: {
type: "message",
id: "msg_second",
content: [{ type: "output_text", text: "first extended" }],
},
},
{ type: "response.completed", response: { id: "resp_order", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
expect(output.content).toMatchObject([
{ type: "text", text: "first" },
{ type: "text", text: "first extended" },
{ type: "thinking", thinking: "thought" },
]);
expect(
events
.filter((event) => event.type.endsWith("_start"))
.map((event) => [event.type, "contentIndex" in event ? event.contentIndex : undefined]),
).toEqual([
["text_start", 0],
["text_start", 1],
["thinking_start", 2],
]);
});
it("finalizes incomplete terminal events with usage", async () => {
const output = createAssistantOutput();
await processResponsesStream(
responseEvents([
{
type: "response.incomplete",
response: {
id: "resp_incomplete",
status: "incomplete",
usage: {
input_tokens: 30,
output_tokens: 12,
total_tokens: 42,
input_tokens_details: { cached_tokens: 5, cache_write_tokens: 3 },
},
},
},
]),
output,
new AssistantMessageEventStream(),
nativeOpenAIModel,
);
expect(output.stopReason).toBe("length");
expect(output.usage).toMatchObject({
input: 22,
output: 12,
cacheRead: 5,
cacheWrite: 3,
totalTokens: 42,
});
});
it("reports content-filtered incomplete responses as errors", async () => {
const output = createAssistantOutput();
await processResponsesStream(
responseEvents([
{
type: "response.incomplete",
response: {
id: "resp_filtered",
status: "incomplete",
incomplete_details: { reason: "content_filter" },
},
},
]),
output,
new AssistantMessageEventStream(),
nativeOpenAIModel,
);
expect(output.stopReason).toBe("error");
expect(output.errorMessage).toBe("Provider incomplete_reason: content_filter");
const lifecycleOutput = createAssistantOutput();
await runResponsesStreamLifecycle({
stream: new AssistantMessageEventStream(),
model: nativeOpenAIModel,
output: lifecycleOutput,
createClient: () => ({
responses: {
create: () => ({
withResponse: async () => ({
data: streamResponsesEvents([
{
type: "response.incomplete",
response: {
id: "resp_filtered_lifecycle",
status: "incomplete",
incomplete_details: { reason: "content_filter" },
},
} as ResponseStreamEvent,
]),
response: new Response(null, { status: 200 }),
}),
}),
},
}),
buildParams: () => ({ model: nativeOpenAIModel.id, input: [], stream: true }),
formatError: (error) => (error instanceof Error ? error.message : String(error)),
});
expect(lifecycleOutput.stopReason).toBe("error");
expect(lifecycleOutput.errorMessage).toBe("Provider incomplete_reason: content_filter");
});
it("preserves failed terminal response details", async () => {
const output = createAssistantOutput();
await processResponsesStream(
responseEvents([
{
type: "response.failed",
response: {
id: "resp_failed",
status: "failed",
error: { code: "server_error", message: "provider failed" },
},
},
]),
output,
new AssistantMessageEventStream(),
nativeOpenAIModel,
);
expect(output).toMatchObject({
responseId: "resp_failed",
stopReason: "error",
errorMessage: "server_error: provider failed",
});
});
it("rejects streams that end without a terminal response event", async () => {
const output = createAssistantOutput();
output.usage.input = 7;
await expect(
processResponsesStream(
responseEvents([{ type: "response.created", response: { id: "resp_truncated" } }]),
output,
new AssistantMessageEventStream(),
nativeOpenAIModel,
),
).rejects.toThrow("OpenAI Responses stream ended before a terminal response event");
expect(output.usage.input).toBe(7);
});
it.each([
["omits arguments", undefined],
["sends empty arguments", ""],
@@ -990,6 +1498,7 @@ describe("processResponsesStream", () => {
type: "response.output_item.done",
item: { type: "function_call", name: "computer", arguments: "{}" },
},
{ type: "response.completed", response: { id: "resp_idless", status: "completed" } },
]),
output,
stream,
@@ -1037,6 +1546,11 @@ describe("processResponsesStream", () => {
status: "completed",
},
},
{
type: "response.completed",
sequence_number: 3,
response: { id: "resp_without_item_id", status: "completed" },
} as ResponseStreamEvent,
];
const output = createAssistantOutput();
@@ -1084,6 +1598,11 @@ describe("processResponsesStream", () => {
status: "completed",
},
},
{
type: "response.completed",
sequence_number: 3,
response: { id: "resp_weather", status: "completed" },
} as ResponseStreamEvent,
];
const output = createAssistantOutput();
const { stream, events } = createCapturedAssistantMessageEventStream();
@@ -1221,6 +1740,11 @@ describe("processResponsesStream", () => {
status: "completed",
},
},
{
type: "response.completed",
sequence_number: 7,
response: { id: "resp_interleaved", status: "completed" },
} as ResponseStreamEvent,
];
const output = createAssistantOutput();
const { stream, events } = createCapturedAssistantMessageEventStream();
@@ -1740,6 +2264,7 @@ describe("processResponsesStream", () => {
arguments: '{"slot":1}',
},
},
{ type: "response.completed", response: { id: "resp_suffix", status: "completed" } },
]),
output,
stream,
@@ -1811,6 +2336,10 @@ describe("processResponsesStream", () => {
arguments: '{"slot":0}',
},
},
{
type: "response.completed",
response: { id: "resp_omitted_completions", status: "completed" },
},
]),
output,
stream,
@@ -1867,6 +2396,10 @@ describe("processResponsesStream", () => {
arguments: '{"slot":0}',
},
},
{
type: "response.completed",
response: { id: "resp_identity_mismatch", status: "completed" },
},
]),
output,
stream,
@@ -1931,6 +2464,7 @@ describe("processResponsesStream", () => {
arguments: '{"slot":1}',
},
},
{ type: "response.completed", response: { id: "resp_sequential", status: "completed" } },
]),
output,
stream,
@@ -2013,6 +2547,7 @@ describe("processResponsesStream", () => {
status: "completed",
},
},
{ type: "response.completed", response: { id: "resp_item_only", status: "completed" } },
]),
output,
stream,

697
packages/ai/src/providers/openai-responses-shared.ts
View File

@@ -4,12 +4,12 @@ import type OpenAI from "openai";
import type {
ResponseCreateParamsStreaming,
ResponseFunctionCallOutputItemList,
ResponseFunctionToolCall,
ResponseInput,
ResponseInputItem,
ResponseInputContent,
ResponseInputImage,
ResponseInputText,
ResponseOutputItem,
ResponseOutputMessage,
ResponseReasoningItem,
ResponseStreamEvent,
@@ -326,7 +326,9 @@ export function convertResponsesMessages<TApi extends Api>(
const includeSystemPrompt = options?.includeSystemPrompt ?? true;
if (includeSystemPrompt && context.systemPrompt) {
const role = model.reasoning ? "developer" : "system";
const compat = model.compat as { supportsDeveloperRole?: boolean } | undefined;
const role =
model.reasoning && compat?.supportsDeveloperRole !== false ? "developer" : "system";
messages.push({
type: "message",
role,
@@ -554,7 +556,7 @@ export function applyCommonResponsesParams<TApi extends Api>(
config?: { setDefaultReasoningOff?: boolean },
): void {
if (options?.maxTokens) {
params.max_output_tokens = options.maxTokens;
params.max_output_tokens = Math.max(options.maxTokens, 16);
}
if (options?.temperature !== undefined && supportsOpenAITemperature(model)) {
@@ -596,7 +598,7 @@ function buildResponsesRequestOptions(
return {
...(options?.signal ? { signal: options.signal } : {}),
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}),
maxRetries: options?.maxRetries ?? 0,
};
}
@@ -665,7 +667,7 @@ export async function runResponsesStreamLifecycle<TApi extends Api>(params: {
}
if (output.stopReason === "aborted" || output.stopReason === "error") {
throw new Error("An unknown error occurred");
throw new Error(output.errorMessage ?? "An unknown error occurred");
}
stream.push({ type: "done", reason: output.stopReason, message: output });
@@ -692,30 +694,101 @@ export async function processResponsesStream<TApi extends Api>(
model: Model<TApi>,
options?: OpenAIResponsesProcessStreamOptions,
): Promise<void> {
let currentItem:
| ResponseReasoningItem
| ResponsesStreamOutputMessage
| ResponseFunctionToolCall
| null = null;
let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null =
null;
type StreamingToolCallBlock = ToolCall & { partialJson: string };
type StreamingToolCallState = ResponsesToolCallState & {
block: StreamingToolCallBlock;
contentIndex: number;
};
const streamingToolCalls = createResponsesToolCallTracker<StreamingToolCallState>();
let lastTextBlock: {
type TextBlockReference = {
block: TextContent;
index: number;
phase: TextSignatureV1["phase"] | undefined;
} | null = null;
// While a message item may still be a cumulative snapshot of lastTextBlock,
// its public block is deferred so a collapsed item never leaves an
// unbalanced text_start behind (#91959). null = no deferral in progress.
let pendingMessageText: string | null = null;
};
type ResponsesOutputSlot =
| {
type: "thinking";
item: ResponseReasoningItem;
block: ThinkingContent;
contentIndex: number;
}
| {
type: "text";
item: ResponsesStreamOutputMessage;
block: TextContent | null;
contentIndex: number | undefined;
pendingText: string | null;
collapseCandidate: TextBlockReference | null;
}
| { type: "toolCall"; toolCall: StreamingToolCallState };
const streamingToolCalls = createResponsesToolCallTracker<StreamingToolCallState>();
const outputSlots = new Map<number, ResponsesOutputSlot>();
const reasoningBlocksById = new Map<string, ThinkingContent>();
let unindexedOutputSlot: ResponsesOutputSlot | undefined;
let terminalResponseEvent: "finalized" | "failed" | undefined;
let lastTextBlock: TextBlockReference | null = null;
const blocks = output.content;
const blockIndex = () => blocks.length - 1;
const readOutputIndex = (event: object): number | undefined => {
const outputIndex = (event as { output_index?: unknown }).output_index;
return typeof outputIndex === "number" && Number.isInteger(outputIndex) && outputIndex >= 0
? outputIndex
: undefined;
};
const registerOutputSlot = (event: object, slot: ResponsesOutputSlot): void => {
const outputIndex = readOutputIndex(event);
if (outputIndex === undefined) {
if (unindexedOutputSlot) {
throw new Error("Responses stream added overlapping unindexed output items");
}
unindexedOutputSlot = slot;
return;
}
if (outputSlots.has(outputIndex)) {
throw new Error(`Responses stream reused active output index ${outputIndex}`);
}
outputSlots.set(outputIndex, slot);
};
const resolveOutputSlot = <TType extends ResponsesOutputSlot["type"]>(
event: object,
type: TType,
): Extract<ResponsesOutputSlot, { type: TType }> | undefined => {
const outputIndex = readOutputIndex(event);
let slot = outputIndex === undefined ? unindexedOutputSlot : outputSlots.get(outputIndex);
if (outputIndex === undefined && !slot) {
const matchingSlots = [...outputSlots.values()].filter(
(candidate) => candidate.type === type,
);
slot = matchingSlots.length === 1 ? matchingSlots[0] : undefined;
}
return slot?.type === type
? (slot as Extract<ResponsesOutputSlot, { type: TType }>)
: undefined;
};
const forgetOutputSlot = (event: object, slot: ResponsesOutputSlot): void => {
const outputIndex = readOutputIndex(event);
if (outputIndex === undefined) {
if (unindexedOutputSlot === slot) {
unindexedOutputSlot = undefined;
} else {
for (const [indexedOutput, indexedSlot] of outputSlots) {
if (indexedSlot === slot) {
outputSlots.delete(indexedOutput);
}
}
}
return;
}
if (outputSlots.get(outputIndex) === slot) {
outputSlots.delete(outputIndex);
}
};
const forgetToolCallOutputSlot = (toolCall: StreamingToolCallState): void => {
for (const [outputIndex, slot] of outputSlots) {
if (slot.type === "toolCall" && slot.toolCall === toolCall) {
outputSlots.delete(outputIndex);
}
}
};
const readIdentityValue = (value: unknown): string | undefined => {
const identity = typeof value === "string" ? value.trim() : "";
return identity || undefined;
@@ -739,30 +812,194 @@ export async function processResponsesStream<TApi extends Api>(
}
return name;
};
const appendPendingMessageDelta = (delta: string) => {
pendingMessageText = `${pendingMessageText ?? ""}${delta}`;
const priorText = lastTextBlock?.block.text ?? "";
if (priorText.startsWith(pendingMessageText) || pendingMessageText.startsWith(priorText)) {
const createOutputSlot = (
event: object,
item: ResponseOutputItem | ResponsesStreamOutputMessage,
): ResponsesOutputSlot | undefined => {
if (item.type === "reasoning") {
const block: ThinkingContent = { type: "thinking", thinking: "" };
const slot = {
type: "thinking",
item,
block,
contentIndex: blocks.length,
} satisfies ResponsesOutputSlot;
blocks.push(block);
registerOutputSlot(event, slot);
stream.push({ type: "thinking_start", contentIndex: slot.contentIndex, partial: output });
return slot;
}
if (item.type === "message") {
const messageItem = item as ResponsesStreamOutputMessage;
const collapseCandidate = lastTextBlock;
const block: TextContent | null = collapseCandidate
? null
: {
type: "text",
text: "",
...(messageItem.phase
? { textSignature: encodeTextSignatureV1(messageItem.id, messageItem.phase) }
: {}),
};
const slot = {
type: "text",
item: messageItem,
block,
contentIndex: block ? blocks.length : undefined,
pendingText: collapseCandidate ? "" : null,
collapseCandidate,
} satisfies ResponsesOutputSlot;
if (block) {
blocks.push(block);
}
registerOutputSlot(event, slot);
if (slot.contentIndex !== undefined) {
stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output });
}
return slot;
}
return undefined;
};
const resolveOutputItemSlot = (
event: object,
item: ResponseOutputItem | ResponsesStreamOutputMessage,
): ResponsesOutputSlot | undefined => {
if (item.type === "reasoning") {
return resolveOutputSlot(event, "thinking");
}
if (item.type === "message") {
return resolveOutputSlot(event, "text");
}
const outputIndex = readOutputIndex(event);
return outputIndex === undefined ? undefined : outputSlots.get(outputIndex);
};
const getOrCreateOutputSlot = (
event: object,
item: ResponseOutputItem | ResponsesStreamOutputMessage,
): ResponsesOutputSlot | undefined => {
return resolveOutputItemSlot(event, item) ?? createOutputSlot(event, item);
};
const materializeDeferredTextSlot = (
slot: Extract<ResponsesOutputSlot, { type: "text" }>,
): void => {
if (slot.block || slot.pendingText === null) {
return;
}
const text = slot.pendingText;
slot.block = {
type: "text",
text,
...(slot.item.phase
? { textSignature: encodeTextSignatureV1(slot.item.id, slot.item.phase) }
: {}),
};
blocks.push(slot.block);
slot.contentIndex = blockIndex();
stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output });
if (text) {
stream.push({
type: "text_delta",
contentIndex: slot.contentIndex,
delta: text,
partial: output,
});
}
if (lastTextBlock === slot.collapseCandidate) {
lastTextBlock = null;
}
slot.pendingText = null;
slot.collapseCandidate = null;
};
const materializeDeferredTextSlots = (except?: ResponsesOutputSlot): void => {
for (const slot of outputSlots.values()) {
if (slot !== except && slot.type === "text") {
materializeDeferredTextSlot(slot);
}
}
if (unindexedOutputSlot !== except && unindexedOutputSlot?.type === "text") {
materializeDeferredTextSlot(unindexedOutputSlot);
}
};
const appendPendingMessageDelta = (
slot: Extract<ResponsesOutputSlot, { type: "text" }>,
delta: string,
) => {
slot.pendingText = `${slot.pendingText ?? ""}${delta}`;
const priorText = slot.collapseCandidate?.block.text ?? "";
if (priorText.startsWith(slot.pendingText) || slot.pendingText.startsWith(priorText)) {
return;
}
// Diverged from the prior text: this is a distinct message, so open its
// block now and replay the withheld text as one delta.
currentBlock = {
type: "text",
text: pendingMessageText,
...(currentItem?.type === "message" && currentItem.phase
? { textSignature: encodeTextSignatureV1(currentItem.id, currentItem.phase ?? undefined) }
: {}),
};
blocks.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: pendingMessageText,
partial: output,
});
pendingMessageText = null;
materializeDeferredTextSlot(slot);
};
const backfillReasoningSignatures = (responseOutput: ResponseOutputItem[]): void => {
for (const item of responseOutput) {
if (item.type !== "reasoning" || !item.encrypted_content) {
continue;
}
const block = reasoningBlocksById.get(item.id);
if (!block?.thinkingSignature) {
continue;
}
const storedItem = JSON.parse(block.thinkingSignature) as ResponseReasoningItem;
if (storedItem.encrypted_content) {
continue;
}
block.thinkingSignature = JSON.stringify({
...storedItem,
encrypted_content: item.encrypted_content,
});
}
};
const finalizeResponse = (
response: Extract<
ResponseStreamEvent,
{ type: "response.completed" | "response.incomplete" }
>["response"],
): void => {
terminalResponseEvent = "finalized";
backfillReasoningSignatures(response.output ?? []);
if (response.id) {
output.responseId = response.id;
}
if (response.usage) {
const inputTokenDetails = response.usage.input_tokens_details as
| ResponsesInputTokensDetails
| null
| undefined;
const cachedTokens = inputTokenDetails?.cached_tokens || 0;
const cacheWriteTokens = inputTokenDetails?.cache_write_tokens || 0;
output.usage = {
// OpenAI includes cache reads and writes in input_tokens, so split both priced buckets.
input: Math.max(0, (response.usage.input_tokens || 0) - cachedTokens - cacheWriteTokens),
output: response.usage.output_tokens || 0,
cacheRead: cachedTokens,
cacheWrite: cacheWriteTokens,
totalTokens: response.usage.total_tokens || 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
};
}
calculateCost(model, output.usage);
if (options?.applyServiceTierPricing) {
const serviceTier = options.resolveServiceTier
? options.resolveServiceTier(response.service_tier, options.serviceTier)
: (response.service_tier ?? options.serviceTier);
options.applyServiceTierPricing(output.usage, serviceTier);
}
if (
response.status === "incomplete" &&
response.incomplete_details?.reason === "content_filter"
) {
output.stopReason = "error";
output.errorMessage = "Provider incomplete_reason: content_filter";
} else {
output.stopReason = mapStopReason(response.status);
}
if (output.content.some((block) => block.type === "toolCall") && output.stopReason === "stop") {
output.stopReason = "toolUse";
}
};
const guardedStream = withFirstStreamEventTimeout(openaiStream, {
@@ -779,32 +1016,15 @@ export async function processResponsesStream<TApi extends Api>(
if (event.type === "response.created") {
output.responseId = event.response.id;
} else if (event.type === "response.output_item.added") {
materializeDeferredTextSlots();
const item = event.item;
if (item.type !== "message") {
// Snapshot collapse only applies to back-to-back message items; any
// other item is a real boundary (see resolveResponsesMessageSnapshotCollapse).
lastTextBlock = null;
pendingMessageText = null;
}
if (item.type === "reasoning") {
currentItem = item;
currentBlock = { type: "thinking", thinking: "" };
output.content.push(currentBlock);
stream.push({ type: "thinking_start", contentIndex: blockIndex(), partial: output });
} else if (item.type === "message") {
currentItem = item;
if (lastTextBlock) {
currentBlock = null;
pendingMessageText = "";
} else {
currentBlock = {
type: "text",
text: "",
...(item.phase ? { textSignature: encodeTextSignatureV1(item.id, item.phase) } : {}),
};
output.content.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
}
if (item.type === "reasoning" || item.type === "message") {
createOutputSlot(event, item);
} else if (item.type === "function_call") {
const toolCallBlock: StreamingToolCallBlock = {
type: "toolCall",
@@ -821,129 +1041,144 @@ export async function processResponsesStream<TApi extends Api>(
...readResponsesToolCallItemIdentity(item),
};
streamingToolCalls.register(event, toolCallState);
currentItem = item;
currentBlock = toolCallBlock;
if (readOutputIndex(event) !== undefined) {
registerOutputSlot(event, { type: "toolCall", toolCall: toolCallState });
}
output.content.push(toolCallBlock);
stream.push({ type: "toolcall_start", contentIndex, partial: output });
}
} else if (event.type === "response.reasoning_summary_part.added") {
if (currentItem && currentItem.type === "reasoning") {
currentItem.summary = currentItem.summary || [];
currentItem.summary.push(event.part);
const slot = resolveOutputSlot(event, "thinking");
if (!slot) {
continue;
}
slot.item.summary = slot.item.summary || [];
slot.item.summary.push(event.part);
} else if (event.type === "response.reasoning_summary_text.delta") {
if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") {
currentItem.summary = currentItem.summary || [];
const lastPart = currentItem.summary[currentItem.summary.length - 1];
if (lastPart) {
currentBlock.thinking += event.delta;
lastPart.text += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
const slot = resolveOutputSlot(event, "thinking");
if (!slot) {
continue;
}
slot.item.summary = slot.item.summary || [];
const lastPart = slot.item.summary[slot.item.summary.length - 1];
if (!lastPart) {
continue;
}
slot.block.thinking += event.delta;
lastPart.text += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
} else if (event.type === "response.reasoning_summary_part.done") {
if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") {
currentItem.summary = currentItem.summary || [];
const lastPart = currentItem.summary[currentItem.summary.length - 1];
if (lastPart) {
currentBlock.thinking += "\n\n";
lastPart.text += "\n\n";
stream.push({
type: "thinking_delta",
contentIndex: blockIndex(),
delta: "\n\n",
partial: output,
});
}
const slot = resolveOutputSlot(event, "thinking");
if (!slot) {
continue;
}
slot.item.summary = slot.item.summary || [];
const lastPart = slot.item.summary[slot.item.summary.length - 1];
if (!lastPart) {
continue;
}
slot.block.thinking += "\n\n";
lastPart.text += "\n\n";
stream.push({
type: "thinking_delta",
contentIndex: slot.contentIndex,
delta: "\n\n",
partial: output,
});
} else if (event.type === "response.reasoning_text.delta") {
if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") {
currentBlock.thinking += event.delta;
const slot = resolveOutputSlot(event, "thinking");
if (!slot) {
continue;
}
slot.block.thinking += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
} else if (event.type === "response.content_part.added") {
const slot = resolveOutputSlot(event, "text");
if (!slot) {
continue;
}
slot.item.content = slot.item.content || [];
if (
event.part.type === OPENAI_RESPONSES_OUTPUT_TEXT_CONTENT_PART_TYPE ||
event.part.type === AZURE_RESPONSES_TEXT_CONTENT_PART_TYPE ||
event.part.type === "refusal"
) {
slot.item.content.push(event.part);
}
} else if (event.type === "response.output_text.delta") {
const slot = resolveOutputSlot(event, "text");
if (!slot?.item.content || slot.item.content.length === 0) {
continue;
}
const lastPart = slot.item.content[slot.item.content.length - 1];
if (!isResponsesTextContentPartType(lastPart?.type)) {
continue;
}
lastPart.text += event.delta;
if (slot.pendingText !== null) {
appendPendingMessageDelta(slot, event.delta);
} else if (slot.block && slot.contentIndex !== undefined) {
slot.block.text += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: blockIndex(),
type: "text_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
}
} else if (event.type === "response.content_part.added") {
if (currentItem?.type === "message") {
currentItem.content = currentItem.content || [];
if (
event.part.type === OPENAI_RESPONSES_OUTPUT_TEXT_CONTENT_PART_TYPE ||
event.part.type === AZURE_RESPONSES_TEXT_CONTENT_PART_TYPE ||
event.part.type === "refusal"
) {
currentItem.content.push(event.part);
}
}
} else if (event.type === "response.output_text.delta") {
if (currentItem?.type === "message") {
if (!currentItem.content || currentItem.content.length === 0) {
continue;
}
const lastPart = currentItem.content[currentItem.content.length - 1];
if (isResponsesTextContentPartType(lastPart?.type)) {
lastPart.text += event.delta;
if (pendingMessageText !== null) {
appendPendingMessageDelta(event.delta);
} else if (currentBlock?.type === "text") {
currentBlock.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
}
}
} else if (isAzureResponsesTextDeltaEvent(event)) {
if (currentItem?.type === "message") {
currentItem.content = currentItem.content || [];
let lastPart = currentItem.content[currentItem.content.length - 1];
if (lastPart?.type !== "text") {
lastPart = { type: "text", text: "" };
currentItem.content.push(lastPart);
}
lastPart.text += event.delta;
if (pendingMessageText !== null) {
appendPendingMessageDelta(event.delta);
} else if (currentBlock?.type === "text") {
currentBlock.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
const slot = resolveOutputSlot(event, "text");
if (!slot) {
continue;
}
slot.item.content = slot.item.content || [];
let lastPart = slot.item.content[slot.item.content.length - 1];
if (lastPart?.type !== "text") {
lastPart = { type: "text", text: "" };
slot.item.content.push(lastPart);
}
lastPart.text += event.delta;
if (slot.pendingText !== null) {
appendPendingMessageDelta(slot, event.delta);
} else if (slot.block && slot.contentIndex !== undefined) {
slot.block.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
}
} else if (event.type === "response.refusal.delta") {
if (currentItem?.type === "message") {
if (!currentItem.content || currentItem.content.length === 0) {
continue;
}
const lastPart = currentItem.content[currentItem.content.length - 1];
if (lastPart?.type === "refusal") {
lastPart.refusal += event.delta;
if (pendingMessageText !== null) {
appendPendingMessageDelta(event.delta);
} else if (currentBlock?.type === "text") {
currentBlock.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
}
const slot = resolveOutputSlot(event, "text");
if (!slot?.item.content || slot.item.content.length === 0) {
continue;
}
const lastPart = slot.item.content[slot.item.content.length - 1];
if (lastPart?.type !== "refusal") {
continue;
}
lastPart.refusal += event.delta;
if (slot.pendingText !== null) {
appendPendingMessageDelta(slot, event.delta);
} else if (slot.block && slot.contentIndex !== undefined) {
slot.block.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
}
} else if (event.type === "response.function_call_arguments.delta") {
const toolCall = streamingToolCalls.resolve(event);
@@ -992,78 +1227,98 @@ export async function processResponsesStream<TApi extends Api>(
const item = event.item;
if (item.type !== "message") {
lastTextBlock = null;
pendingMessageText = null;
}
if (item.type === "reasoning" && currentBlock?.type === "thinking") {
const existingOutputSlot = resolveOutputItemSlot(event, item);
materializeDeferredTextSlots(existingOutputSlot);
const outputSlot = existingOutputSlot ?? getOrCreateOutputSlot(event, item);
if (item.type === "reasoning" && outputSlot?.type === "thinking") {
const summaryText = item.summary?.map((s) => s.text).join("\n\n") || "";
const contentText = item.content?.map((c) => c.text).join("\n\n") || "";
currentBlock.thinking = summaryText || contentText || currentBlock.thinking;
currentBlock.thinkingSignature = JSON.stringify(item);
outputSlot.block.thinking = summaryText || contentText || outputSlot.block.thinking;
outputSlot.block.thinkingSignature = JSON.stringify(item);
if (typeof item.id === "string") {
reasoningBlocksById.set(item.id, outputSlot.block);
}
stream.push({
type: "thinking_end",
contentIndex: blockIndex(),
content: currentBlock.thinking,
contentIndex: outputSlot.contentIndex,
content: outputSlot.block.thinking,
partial: output,
});
currentBlock = null;
forgetOutputSlot(event, outputSlot);
} else if (
item.type === "message" &&
(currentBlock?.type === "text" || pendingMessageText !== null)
outputSlot?.type === "text" &&
(outputSlot.block || outputSlot.pendingText !== null)
) {
// Support both OpenAI "output_text" and Azure "text" content types
const finalText = item.content
.map((c) => (c.type === "output_text" || c.type === "text" ? c.text : c.refusal))
.join("");
const streamedText = outputSlot.pendingText ?? outputSlot.block?.text ?? "";
const finalText =
item.content == null
? streamedText
: item.content
.map((c) => (c.type === "output_text" || c.type === "text" ? c.text : c.refusal))
.join("");
const phase = item.phase ?? undefined;
const collapse =
pendingMessageText !== null
outputSlot.pendingText !== null
? resolveResponsesMessageSnapshotCollapse({
prior: lastTextBlock && {
text: lastTextBlock.block.text,
phase: lastTextBlock.phase,
prior: outputSlot.collapseCandidate && {
text: outputSlot.collapseCandidate.block.text,
phase: outputSlot.collapseCandidate.phase,
},
nextText: finalText,
nextPhase: phase,
})
: ({ kind: "keep" } as const);
pendingMessageText = null;
if (collapse.kind === "extend" && lastTextBlock) {
outputSlot.pendingText = null;
if (collapse.kind === "extend" && outputSlot.collapseCandidate) {
// Cumulative snapshot of the prior message item: replace its text
// instead of appending another copy. The deferred block was never
// started publicly, and the newest item's signature is kept so
// replay carries the item that produced this content (#91959).
lastTextBlock.block.text = collapse.text;
lastTextBlock.block.textSignature = encodeTextSignatureV1(item.id, phase);
outputSlot.collapseCandidate.block.text = collapse.text;
outputSlot.collapseCandidate.block.textSignature = encodeTextSignatureV1(item.id, phase);
stream.push({
type: "text_end",
contentIndex: lastTextBlock.index,
contentIndex: outputSlot.collapseCandidate.index,
content: collapse.text,
partial: output,
});
lastTextBlock = outputSlot.collapseCandidate;
} else {
if (currentBlock?.type !== "text") {
if (!outputSlot.block) {
// Deferred distinct message: open its block now, balanced with the
// text_end below.
currentBlock = {
outputSlot.block = {
type: "text",
text: "",
...(phase ? { textSignature: encodeTextSignatureV1(item.id, phase) } : {}),
};
blocks.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
blocks.push(outputSlot.block);
outputSlot.contentIndex = blockIndex();
stream.push({
type: "text_start",
contentIndex: outputSlot.contentIndex,
partial: output,
});
}
currentBlock.text = finalText;
currentBlock.textSignature = encodeTextSignatureV1(item.id, phase);
lastTextBlock = { block: currentBlock, index: blockIndex(), phase };
outputSlot.block.text = finalText;
outputSlot.block.textSignature = encodeTextSignatureV1(item.id, phase);
const contentIndex = outputSlot.contentIndex;
if (contentIndex === undefined) {
throw new Error("Responses stream finalized text without a content index");
}
lastTextBlock = { block: outputSlot.block, index: contentIndex, phase };
stream.push({
type: "text_end",
contentIndex: blockIndex(),
content: currentBlock.text,
contentIndex,
content: outputSlot.block.text,
partial: output,
});
}
currentBlock = null;
forgetOutputSlot(event, outputSlot);
} else if (item.type === "function_call") {
const streamingToolCall = streamingToolCalls.resolve(
event,
@@ -1117,10 +1372,7 @@ export async function processResponsesStream<TApi extends Api>(
if (streamingToolCall) {
streamingToolCalls.forget(streamingToolCall);
}
if (currentBlock === toolCall) {
currentBlock = null;
currentItem = null;
forgetToolCallOutputSlot(streamingToolCall);
}
stream.push({
type: "toolcall_end",
@@ -1129,43 +1381,11 @@ export async function processResponsesStream<TApi extends Api>(
partial: output,
});
}
} else if (event.type === "response.completed") {
} else if (event.type === "response.completed" || event.type === "response.incomplete") {
if (streamingToolCalls.hasActive()) {
throw new Error("Responses stream completed with unresolved tool calls");
}
const response = event.response;
if (response?.id) {
output.responseId = response.id;
}
if (response?.usage) {
const inputTokenDetails = response.usage.input_tokens_details as
| ResponsesInputTokensDetails
| null
| undefined;
const cachedTokens = inputTokenDetails?.cached_tokens || 0;
const cacheWriteTokens = inputTokenDetails?.cache_write_tokens || 0;
output.usage = {
// OpenAI includes cache reads and writes in input_tokens, so split both priced buckets.
input: Math.max(0, (response.usage.input_tokens || 0) - cachedTokens - cacheWriteTokens),
output: response.usage.output_tokens || 0,
cacheRead: cachedTokens,
cacheWrite: cacheWriteTokens,
totalTokens: response.usage.total_tokens || 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
};
}
calculateCost(model, output.usage);
if (options?.applyServiceTierPricing) {
const serviceTier = options.resolveServiceTier
? options.resolveServiceTier(response?.service_tier, options.serviceTier)
: (response?.service_tier ?? options.serviceTier);
options.applyServiceTierPricing(output.usage, serviceTier);
}
// Map status to stop reason
output.stopReason = mapStopReason(response?.status);
if (output.content.some((b) => b.type === "toolCall") && output.stopReason === "stop") {
output.stopReason = "toolUse";
}
finalizeResponse(event.response);
} else if (event.type === "error") {
throw new Error(
event.message ? `Error Code ${event.code}: ${event.message}` : "Unknown error",
@@ -1173,17 +1393,26 @@ export async function processResponsesStream<TApi extends Api>(
} else if (event.type === "response.failed") {
const error = event.response?.error;
const details = event.response?.incomplete_details;
const msg = error
output.responseId = event.response.id;
output.stopReason = "error";
output.errorMessage = error
? `${error.code || "unknown"}: ${error.message || "no message"}`
: details?.reason
? `incomplete: ${details.reason}`
: "Unknown error (no error details in response)";
throw new Error(msg);
terminalResponseEvent = "failed";
break;
}
}
if (terminalResponseEvent === "failed") {
return;
}
if (streamingToolCalls.hasActive()) {
throw new Error("Responses stream ended with unresolved tool calls");
}
if (!terminalResponseEvent) {
throw new Error("OpenAI Responses stream ended before a terminal response event");
}
}
function mapStopReason(status: OpenAI.Responses.ResponseStatus | undefined): StopReason {

116
packages/ai/src/providers/openai-responses.live.test.ts Normal file
View File

@@ -0,0 +1,116 @@
import { Type } from "typebox";
import { describe, expect, it } from "vitest";
import type { Context, Model } from "../types.js";
import { streamOpenAIResponses } from "./openai-responses.js";
// Live coverage for the Responses stream state machine: real streams interleave
// reasoning/message/tool items, so unit fakes alone cannot prove slot tracking,
// terminal-event handling, or the max_output_tokens floor against the real API.
const LIVE = process.env.OPENCLAW_LIVE_TEST === "1";
const OPENAI_KEY = process.env.OPENAI_API_KEY ?? "";
const describeLive = LIVE && OPENAI_KEY ? describe : describe.skip;
const LIVE_MODEL_ID = process.env.OPENCLAW_LIVE_RESPONSES_MODEL || "gpt-5.6-luna";
const LIVE_TIMEOUT_MS = 120_000;
function liveModel(overrides: Partial<Model<"openai-responses">> = {}) {
return {
id: LIVE_MODEL_ID,
name: LIVE_MODEL_ID,
api: "openai-responses",
provider: "openai",
baseUrl: "https://api.openai.com/v1",
reasoning: true,
input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: 200_000,
maxTokens: 8192,
...overrides,
} satisfies Model<"openai-responses">;
}
describeLive("OpenAI Responses live", () => {
it(
"streams a reply through a terminal event with usage accounting",
async () => {
const context: Context = {
messages: [{ role: "user", content: "Reply with exactly: OK", timestamp: 0 }],
};
const result = await streamOpenAIResponses(liveModel(), context, {
apiKey: OPENAI_KEY,
maxTokens: 256,
}).result();
expect(result.errorMessage).toBeUndefined();
expect(result.stopReason).toBe("stop");
const text = result.content
.filter((block) => block.type === "text")
.map((block) => block.text)
.join("");
expect(text).toContain("OK");
expect(result.usage.totalTokens).toBeGreaterThan(0);
expect(result.usage.output).toBeGreaterThan(0);
},
LIVE_TIMEOUT_MS,
);
it(
"clamps a sub-floor max token budget instead of failing the request",
async () => {
const context: Context = {
messages: [{ role: "user", content: "Reply with exactly: OK", timestamp: 0 }],
};
const result = await streamOpenAIResponses(liveModel(), context, {
apiKey: OPENAI_KEY,
maxTokens: 1,
}).result();
// Pre-floor behavior was a hard 400 from the API; truncation is expected.
expect(result.errorMessage).toBeUndefined();
expect(["stop", "length"]).toContain(result.stopReason);
},
LIVE_TIMEOUT_MS,
);
it(
"keeps reasoning and tool-call items separable on a real interleaved stream",
async () => {
const context: Context = {
messages: [
{
role: "user",
content: "Call the live_probe tool with value set to exactly LIVE_OK.",
timestamp: 0,
},
],
tools: [
{
name: "live_probe",
description: "Records a probe value.",
parameters: Type.Object({ value: Type.String() }),
},
],
};
const result = await streamOpenAIResponses(liveModel(), context, {
apiKey: OPENAI_KEY,
maxTokens: 1024,
reasoningEffort: "low",
}).result();
expect(result.errorMessage).toBeUndefined();
expect(result.stopReason).toBe("toolUse");
const toolCalls = result.content.filter((block) => block.type === "toolCall");
expect(toolCalls).toHaveLength(1);
const probeCall = toolCalls[0];
expect(probeCall?.name).toBe("live_probe");
const probeArguments = (probeCall?.arguments ?? {}) as { value?: string };
expect(probeArguments.value).toBe("LIVE_OK");
for (const block of result.content) {
if (block.type === "thinking") {
expect(block.thinkingSignature).toBeTruthy();
}
}
},
LIVE_TIMEOUT_MS,
);
});

23
packages/ai/src/providers/openai-responses.test.ts
View File

@@ -2,12 +2,18 @@ import { afterEach, describe, expect, it, vi } from "vitest";
import { configureAiTransportHost } from "../host.js";
import type { Context, Model } from "../types.js";
const openAiMockState = vi.hoisted(() => ({ configs: [] as unknown[] }));
const openAiMockState = vi.hoisted(() => ({
configs: [] as unknown[],
params: [] as unknown[],
requestOptions: [] as unknown[],
}));
vi.mock("openai", () => ({
default: class MockOpenAI {
responses = {
create: vi.fn(() => {
create: vi.fn((params: unknown, requestOptions: unknown) => {
openAiMockState.params.push(params);
openAiMockState.requestOptions.push(requestOptions);
throw new Error("stop after constructor");
}),
};
@@ -43,6 +49,8 @@ function model(overrides: Partial<Model<"openai-responses">> = {}) {
describe("OpenAI Responses provider", () => {
afterEach(() => {
openAiMockState.configs = [];
openAiMockState.params = [];
openAiMockState.requestOptions = [];
configureAiTransportHost({});
});
@@ -83,4 +91,15 @@ describe("OpenAI Responses provider", () => {
);
expect(config.fetch).toBe(hostFetch);
});
it("clamps small output limits and disables implicit SDK retries", async () => {
const result = await streamOpenAIResponses(model(), context, {
apiKey: String(1),
maxTokens: 1,
}).result();
expect(result.stopReason).toBe("error");
expect(openAiMockState.params[0]).toMatchObject({ max_output_tokens: 16, store: false });
expect(openAiMockState.requestOptions[0]).toMatchObject({ maxRetries: 0 });
});
});

2
packages/llm-core/src/types.ts
View File

@@ -475,6 +475,8 @@ export interface OpenAICompletionsCompat {
/** Compatibility settings for OpenAI Responses APIs. */
export interface OpenAIResponsesCompat {
/** Whether the provider supports the `developer` role (vs `system`). Default: true. */
supportsDeveloperRole?: boolean;
/** Whether the model accepts the `temperature` parameter. Default: true. */
supportsTemperature?: boolean;
/** Whether to send the OpenAI `session_id` cache-affinity header from `options.sessionId` when caching is enabled. Default: true. */

1
src/infra/vitest-live-config.test.ts
View File

@@ -22,6 +22,7 @@ describe("live vitest config", () => {
expect(liveConfig.test?.include).toEqual([
"src/**/*.live.test.ts",
"test/**/*.live.test.ts",
"packages/*/src/**/*.live.test.ts",
BUNDLED_PLUGIN_LIVE_TEST_GLOB,
]);
expect(normalizeConfigPaths(liveConfig.test?.setupFiles)).toEqual([

7
test/vitest/vitest.live.config.ts
View File

@@ -27,7 +27,12 @@ export default defineConfig({
[...(baseTest.setupFiles ?? []), "test/setup-openclaw-runtime.ts"].map(resolveRepoRootPath),
),
],
include: ["src/**/*.live.test.ts", "test/**/*.live.test.ts", BUNDLED_PLUGIN_LIVE_TEST_GLOB],
include: [
"src/**/*.live.test.ts",
"test/**/*.live.test.ts",
"packages/*/src/**/*.live.test.ts",
BUNDLED_PLUGIN_LIVE_TEST_GLOB,
],
exclude,
},
});