diff --git a/packages/ai/src/transports/openai-completions-transport.ts b/packages/ai/src/transports/openai-completions-transport.ts index 413b2eedb5fd..b0bba89d5960 100644 --- a/packages/ai/src/transports/openai-completions-transport.ts +++ b/packages/ai/src/transports/openai-completions-transport.ts @@ -335,7 +335,10 @@ export function createOpenAICompletionsTransportStreamFn(): StreamFn { firstEventAbort = createFirstStreamEventAbortController(options?.signal); const responseStream = (await client.chat.completions.create( params as never, - buildOpenAISdkRequestOptions(model, firstEventAbort.signal), + buildOpenAISdkRequestOptions(model, firstEventAbort.signal, { + timeoutMs: options?.timeoutMs, + maxRetries: options?.maxRetries, + }), )) as unknown as AsyncIterable; stream.push({ type: "start", partial: output as never }); await processOpenAICompletionsStream(responseStream, output, model, stream, { diff --git a/packages/ai/src/transports/openai-responses-client.ts b/packages/ai/src/transports/openai-responses-client.ts index aaeb9e02a446..36f140c1577b 100644 --- a/packages/ai/src/transports/openai-responses-client.ts +++ b/packages/ai/src/transports/openai-responses-client.ts @@ -190,11 +190,11 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti } const requestStartedAt = Date.now(); firstEventAbort = createFirstStreamEventAbortController(options?.signal); - const requestOptions = buildOpenAISdkRequestOptions( - model, - firstEventAbort.signal, - config.streamRequest ? { stream: true } : undefined, - ); + const requestOptions = buildOpenAISdkRequestOptions(model, firstEventAbort.signal, { + stream: config.streamRequest, + timeoutMs: options?.timeoutMs, + maxRetries: options?.maxRetries, + }); emitModelTransportDebug( log, `[responses] start provider=${model.provider} api=${model.api} model=${model.id} ` + diff --git a/packages/ai/src/transports/openai-transport-params.session-header.test.ts b/packages/ai/src/transports/openai-transport-params.session-header.test.ts index f22173394873..04553591b85e 100644 --- a/packages/ai/src/transports/openai-transport-params.session-header.test.ts +++ b/packages/ai/src/transports/openai-transport-params.session-header.test.ts @@ -1,7 +1,10 @@ import type { Context, Model } from "@openclaw/llm-core"; import { describe, expect, it } from "vitest"; import { OPENAI_PROMPT_CACHE_KEY_MAX_LENGTH } from "../providers/openai-prompt-cache.js"; -import { buildOpenAIClientHeaders } from "./openai-transport-params.js"; +import { + buildOpenAIClientHeaders, + buildOpenAISdkRequestOptions, +} from "./openai-transport-params.js"; const codexModel = { id: "gpt-5.6-luna", @@ -35,3 +38,27 @@ describe("buildOpenAIClientHeaders session_id affinity header", () => { expect(headers.session_id).toBe("abc-123"); }); }); + +describe("buildOpenAISdkRequestOptions turn controls", () => { + const model = { + id: "gpt-5.6-luna", + provider: "openai", + api: "openai-responses", + baseUrl: "https://api.openai.com/v1", + } as Model; + + it("forwards an explicit turn timeout and zero retries to the SDK", () => { + const signal = new AbortController().signal; + + expect( + buildOpenAISdkRequestOptions(model, signal, { + timeoutMs: 1_234, + maxRetries: 0, + }), + ).toEqual({ signal, timeout: 1_234, maxRetries: 0 }); + }); + + it("does not add a retry policy when the turn does not specify one", () => { + expect(buildOpenAISdkRequestOptions(model)).toBeUndefined(); + }); +}); diff --git a/packages/ai/src/transports/openai-transport-params.ts b/packages/ai/src/transports/openai-transport-params.ts index c8f3c3f71a65..3afda0a0f0af 100644 --- a/packages/ai/src/transports/openai-transport-params.ts +++ b/packages/ai/src/transports/openai-transport-params.ts @@ -225,8 +225,8 @@ export function buildOpenAIClientHeaders( return resolvedHeaders; } -function resolveOpenAISdkTimeoutMs(model: Model): number | undefined { - return resolveModelRequestTimeoutMs(model, undefined); +function resolveOpenAISdkTimeoutMs(model: Model, timeoutMs?: number): number | undefined { + return resolveModelRequestTimeoutMs(model, timeoutMs); } export function buildOpenAISdkClientOptions(model: Model): { timeout?: number } { @@ -237,20 +237,28 @@ export function buildOpenAISdkClientOptions(model: Model): { timeout?: number } export function buildOpenAISdkRequestOptions( model: Model, signal?: AbortSignal, - options?: { stream?: boolean }, -): { signal?: AbortSignal; timeout?: number; headers?: Record } | undefined { - const timeout = resolveOpenAISdkTimeoutMs(model); + options?: { stream?: boolean; timeoutMs?: number; maxRetries?: number }, +): + | { + signal?: AbortSignal; + timeout?: number; + maxRetries?: number; + headers?: Record; + } + | undefined { + const timeout = resolveOpenAISdkTimeoutMs(model, options?.timeoutMs); const headers = options?.stream === true && usesNativeOpenAICodexResponsesBackend(model) ? { Accept: "text/event-stream" } : undefined; - if (timeout === undefined && !signal && !headers) { + if (timeout === undefined && options?.maxRetries === undefined && !signal && !headers) { return undefined; } return { ...(headers ? { headers } : {}), ...(signal ? { signal } : {}), ...(timeout !== undefined ? { timeout } : {}), + ...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}), }; } diff --git a/src/agents/openai-transport-stream.streaming.test.ts b/src/agents/openai-transport-stream.streaming.test.ts index 3bca5045885d..a198434c047c 100644 --- a/src/agents/openai-transport-stream.streaming.test.ts +++ b/src/agents/openai-transport-stream.streaming.test.ts @@ -1,5 +1,9 @@ import { createServer } from "node:http"; -import { createOpenAICompletionsTransportStreamFn } from "@openclaw/ai/transports"; +import { + createAzureOpenAIResponsesTransportStreamFn, + createOpenAICompletionsTransportStreamFn, + createOpenAIResponsesTransportStreamFn, +} from "@openclaw/ai/transports"; import type { Model } from "openclaw/plugin-sdk/llm"; import { describe, expect, it, vi } from "vitest"; import { @@ -82,6 +86,85 @@ describe("openai transport stream", () => { ).toBeUndefined(); }); + it.each([ + { + api: "openai-responses" as const, + provider: "openai", + createStream: createOpenAIResponsesTransportStreamFn, + }, + { + api: "azure-openai-responses" as const, + provider: "azure-openai", + createStream: createAzureOpenAIResponsesTransportStreamFn, + }, + { + api: "openai-completions" as const, + provider: "openai", + createStream: createOpenAICompletionsTransportStreamFn, + }, + ])("honors turn timeout and zero retries over real $api HTTP", async (transport) => { + const capturedTimeouts: Array = []; + const server = createServer((request, response) => { + const timeout = request.headers["x-stainless-timeout"]; + capturedTimeouts.push(Array.isArray(timeout) ? timeout[0] : timeout); + request.resume(); + request.on("end", () => { + response.writeHead(500, { "content-type": "application/json" }); + response.end( + JSON.stringify({ + error: { type: "server_error", message: "turn retry regression" }, + }), + ); + }); + }); + + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + try { + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Missing loopback server address"); + } + + const model = { + id: "gpt-5.6-luna", + name: "GPT-5.6 Luna", + api: transport.api, + provider: transport.provider, + baseUrl: `http://127.0.0.1:${address.port}/v1`, + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128_000, + maxTokens: 4_096, + requestTimeoutMs: 900_000, + } satisfies Model & { requestTimeoutMs: number }; + + const stream = await transport.createStream()( + model, + { + messages: [{ role: "user", content: "Reply OK", timestamp: Date.now() }], + tools: [], + }, + { apiKey: "test-key", timeoutMs: 1_234, maxRetries: 0 }, + ); + + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + + expect(eventTypes).toContain("error"); + // The SDK advertises request timeouts in whole seconds on the wire. + expect(capturedTimeouts).toEqual(["1"]); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + } + }); + it("streams OpenAI-compatible loopback requests with the configured SDK timeout", async () => { let captured: { path?: string; timeout?: string; model?: string; roles?: string[] } = {}; const server = createServer((req, res) => {