diff --git a/src/gateway/openai-http.test.ts b/src/gateway/openai-http.test.ts index 06e3fd9466a6..3a44d13d327a 100644 --- a/src/gateway/openai-http.test.ts +++ b/src/gateway/openai-http.test.ts @@ -2166,34 +2166,234 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { expect(finishChoice?.finish_reason).toBe("stop"); }); + it.each([ + { name: "resolved", reject: false }, + { name: "rejected", reject: true }, + ])( + "fails an official SDK stream when an error lifecycle precedes a $name run", + async ({ reject }) => { + const idleRootCount = getActiveGatewayRootWorkCount(); + const wireResponse = createDeferred(); + agentCommand.mockClear(); + agentCommand.mockImplementationOnce((async (opts: unknown) => { + const runId = (opts as { runId?: string }).runId; + if (!runId) { + throw new Error("expected a streaming chat-completion run ID"); + } + emitAgentEvent({ runId, stream: "assistant", data: { delta: "partial answer" } }); + emitAgentEvent({ + runId, + stream: "lifecycle", + data: { phase: "error", error: "All model fallback candidates failed" }, + }); + emitAgentEvent({ + runId, + stream: "lifecycle", + data: { phase: "error", error: "A later lifecycle event must not replace the failure" }, + }); + if (reject) { + throw new Error("private upstream failure"); + } + return { + payloads: [{ text: "partial answer" }], + meta: { agentMeta: { usage: { input: 11, output: 7, total: 18 } } }, + }; + }) as never); + + const client = new OpenAI({ + apiKey: "test", + baseURL: `http://127.0.0.1:${enabledPort}/v1`, + defaultHeaders: { "x-openclaw-scopes": "operator.write" }, + maxRetries: 0, + fetch: async (input, init) => { + const response = await fetch(input, init); + void response.clone().text().then(wireResponse.resolve, wireResponse.reject); + return response; + }, + }); + const stream = await client.chat.completions.create({ + model: "openclaw", + messages: [{ role: "user", content: "Report the provider failure." }], + stream: true, + }); + const deliveredContent: string[] = []; + const deliveredFinishReasons: Array = []; + + await expect(async () => { + for await (const chunk of stream) { + for (const choice of chunk.choices) { + if (typeof choice.delta.content === "string") { + deliveredContent.push(choice.delta.content); + } + deliveredFinishReasons.push(choice.finish_reason); + } + } + }).rejects.toMatchObject({ + message: "All model fallback candidates failed", + type: "api_error", + }); + + const data = parseSseDataLines(await wireResponse.promise); + const chunks = data + .filter((line) => line !== "[DONE]") + .map((line) => JSON.parse(line) as Record); + expect(deliveredContent).toEqual(["partial answer"]); + expect(deliveredFinishReasons.every((reason) => reason === null)).toBe(true); + expect(chunks.filter((chunk) => "error" in chunk)).toEqual([ + { error: { message: "All model fallback candidates failed", type: "api_error" } }, + ]); + expect(data.at(-1)).toBe("[DONE]"); + expect(agentCommand).toHaveBeenCalledTimes(1); + await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(idleRootCount)); + }, + ); + + it.each([ + { name: "rewritten", replacementText: "final answer" }, + { name: "shortened", replacementText: "dra" }, + { name: "cleared", replacementText: "" }, + ])("fails an official SDK stream when streamed text is $name", async ({ replacementText }) => { + agentCommand.mockClear(); + agentCommand.mockImplementationOnce((async (opts: unknown) => { + const runId = (opts as { runId?: string }).runId; + if (!runId) { + throw new Error("expected a streaming chat-completion run ID"); + } + emitAgentEvent({ + runId, + stream: "assistant", + data: { text: "draft answer", delta: "draft answer" }, + }); + emitAgentEvent({ + runId, + stream: "assistant", + data: { text: replacementText, delta: "", replace: true, phase: "commentary" }, + }); + emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end" } }); + return { payloads: [{ text: replacementText }] }; + }) as never); + + const stream = await createOpenAiChatClient(enabledPort).chat.completions.create({ + model: "openclaw", + messages: [{ role: "user", content: "Reject an incompatible replacement snapshot." }], + stream: true, + }); + const deliveredContent: string[] = []; + await expect(async () => { + for await (const chunk of stream) { + for (const choice of chunk.choices) { + if (typeof choice.delta.content === "string") { + deliveredContent.push(choice.delta.content); + } + } + } + }).rejects.toMatchObject({ + message: "Assistant output cannot be represented as an append-only response stream.", + type: "api_error", + }); + expect(deliveredContent).toEqual(["draft answer"]); + expect(agentCommand).toHaveBeenCalledTimes(1); + }); + + it.each([ + { + name: "a producer replacement snapshot without a delta", + previousDelta: undefined, + replacementDelta: undefined, + }, + { + name: "a producer replacement snapshot with its own delta", + previousDelta: undefined, + replacementDelta: "final answer", + }, + { + name: "an append-compatible replacement after streamed partial text", + previousDelta: "final ", + replacementDelta: "answer", + }, + ])( + "keeps official SDK text consistent for $name", + async ({ previousDelta, replacementDelta }) => { + agentCommand.mockClear(); + agentCommand.mockImplementationOnce((async (opts: unknown) => { + const runId = (opts as { runId?: string }).runId; + if (!runId) { + throw new Error("expected a streaming chat-completion run ID"); + } + if (previousDelta) { + emitAgentEvent({ + runId, + stream: "assistant", + data: { text: previousDelta, delta: previousDelta }, + }); + } + emitAgentEvent({ + runId, + stream: "assistant", + data: { + text: "final answer", + replace: true, + phase: "commentary", + ...(replacementDelta === undefined ? {} : { delta: replacementDelta }), + }, + }); + emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end" } }); + return { payloads: [{ text: "final answer" }] }; + }) as never); + + const stream = await createOpenAiChatClient(enabledPort).chat.completions.create({ + model: "openclaw", + messages: [{ role: "user", content: "Preserve an append-compatible replacement." }], + stream: true, + }); + const deliveredContent: string[] = []; + const finishReasons: Array = []; + for await (const chunk of stream) { + for (const choice of chunk.choices) { + if (typeof choice.delta.content === "string") { + deliveredContent.push(choice.delta.content); + } + finishReasons.push(choice.finish_reason); + } + } + expect(deliveredContent.join("")).toBe("final answer"); + expect(finishReasons.at(-1)).toBe("stop"); + expect(agentCommand).toHaveBeenCalledTimes(1); + }, + ); + it.each([ { name: "successful completion without a provider terminal", fail: false, providerTerminal: false, expected: "hello", + protocolError: false, }, { name: "successful completion with a provider terminal", fail: false, providerTerminal: true, expected: "hello", + protocolError: false, }, { name: "internal agent error without a provider terminal", fail: true, providerTerminal: false, expected: "Error: internal error", + protocolError: false, }, { name: "internal agent error with a provider terminal", fail: true, providerTerminal: true, - expected: "Error: internal error", + expected: "Agent run failed", + protocolError: true, }, ])( "separates streamed content from the terminal finish for an official SDK $name", - async ({ fail, providerTerminal, expected }) => { + async ({ fail, providerTerminal, expected, protocolError }) => { const idleRootCount = getActiveGatewayRootWorkCount(); const terminalAdmission = createDeferred<{ active: number; @@ -2271,8 +2471,18 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { delta: { content?: string | null }; finish_reason: string | null; }> = []; - for await (const chunk of stream) { - choices.push(...chunk.choices); + const consumeStream = async () => { + for await (const chunk of stream) { + choices.push(...chunk.choices); + } + }; + if (protocolError) { + await expect(consumeStream()).rejects.toMatchObject({ + message: expected, + type: "api_error", + }); + } else { + await consumeStream(); } const [admission, wire] = await Promise.all([ @@ -2285,13 +2495,17 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { expect(lifecycleTerminals).toEqual([fail ? "error" : "end"]); const contentChoices = choices.filter((choice) => typeof choice.delta.content === "string"); - expect(contentChoices.map((choice) => choice.delta.content).join("")).toBe(expected); + expect(contentChoices.map((choice) => choice.delta.content).join("")).toBe( + protocolError ? "" : expected, + ); expect(contentChoices.every((choice) => choice.finish_reason === null)).toBe(true); const terminalChoices = choices.filter((choice) => choice.finish_reason === "stop"); - expect(terminalChoices).toHaveLength(1); - expect(terminalChoices[0]?.delta).toEqual({}); - expect(choices.at(-1)).toEqual(terminalChoices[0]); + expect(terminalChoices).toHaveLength(protocolError ? 0 : 1); + if (!protocolError) { + expect(terminalChoices[0]?.delta).toEqual({}); + expect(choices.at(-1)).toEqual(terminalChoices[0]); + } await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(idleRootCount)); } finally { continueAgent.resolve(); diff --git a/src/gateway/openai-http.ts b/src/gateway/openai-http.ts index 63626fc303e3..3c0ba30fdf2d 100644 --- a/src/gateway/openai-http.ts +++ b/src/gateway/openai-http.ts @@ -1191,6 +1191,7 @@ export async function handleOpenAiHttpRequest( let resultResolved = false; let closed = false; let observedTerminalLifecycle = false; + let terminalStreamError: { message: string; type: string; code?: string } | undefined; let terminalLifecyclePhase: "end" | "error" = "end"; let stopWatchingDisconnect = () => {}; @@ -1211,6 +1212,10 @@ export async function handleOpenAiHttpRequest( if (closed) { return; } + if (terminalStreamError) { + finishStreamWithError(terminalStreamError); + return; + } closed = true; stopWatchingDisconnect(); unsubscribe(); @@ -1257,6 +1262,20 @@ export async function handleOpenAiHttpRequest( return; } + // SSE deltas cannot retract bytes already delivered to the OpenAI client. + if ( + replace && + typeof text === "string" && + !toolChoiceConstraint && + !text.startsWith(streamedAssistantText) + ) { + terminalStreamError ??= { + message: "Assistant output cannot be represented as an append-only response stream.", + type: "api_error", + }; + return; + } + // Snapshots include prefixes held during tag-boundary filtering; the raw // delta alone can omit a literal leading less-than. const content = @@ -1297,11 +1316,29 @@ export async function handleOpenAiHttpRequest( } if (phase === "end" || phase === "error") { observedTerminalLifecycle = true; + if (phase === "error" && terminalLifecyclePhase !== "error") { + terminalStreamError ??= { + message: normalizeOptionalString(evt.data?.error) ?? "Agent run failed", + type: "api_error", + }; + } requestFinalize(); } } }); + const finishStreamWithError = (error: { message: string; type: string; code?: string }) => { + if (closed) { + return; + } + closed = true; + stopWatchingDisconnect(); + unsubscribe(); + writeSse(res, { error }); + writeDone(res); + res.end(); + }; + // Agent cleanup and deferred SSE delivery have independent lifetimes; // shutdown must wait until both have settled, whichever finishes last. const releaseAgentRootWork = retainGatewayRootWorkAdmissionContinuation(); @@ -1332,6 +1369,11 @@ export async function handleOpenAiHttpRequest( return; } + if (terminalStreamError) { + finishStreamWithError(terminalStreamError); + return; + } + finalUsage = resolveChatCompletionUsage(result); const meta = (result as { meta?: unknown } | null)?.meta; const { stopReason, pendingToolCalls } = resolveStopReasonAndPendingToolCalls(meta); @@ -1346,17 +1388,10 @@ export async function handleOpenAiHttpRequest( pendingToolCalls, }) ) { - closed = true; - stopWatchingDisconnect(); - unsubscribe(); - writeSse(res, { - error: { - message: resolveUnsatisfiedToolChoiceMessage(toolChoiceConstraint), - type: "api_error", - }, + finishStreamWithError({ + message: resolveUnsatisfiedToolChoiceMessage(toolChoiceConstraint), + type: "api_error", }); - writeDone(res); - res.end(); return; } @@ -1416,26 +1451,22 @@ export async function handleOpenAiHttpRequest( terminalLifecyclePhase = "error"; logWarn(`openai-compat: streaming chat completion failed: ${String(err)}`); if (isClientToolNameConflictError(err)) { - closed = true; - stopWatchingDisconnect(); - unsubscribe(); - writeSse(res, { - error: { message: "invalid tool configuration", type: "invalid_request_error" }, + finishStreamWithError({ + message: "invalid tool configuration", + type: "invalid_request_error", }); - writeDone(res); - res.end(); return; } const mapped = resolveOpenAiCompatError(err); if (mapped) { - closed = true; - stopWatchingDisconnect(); - unsubscribe(); - writeSse(res, { error: mapped.error }); - writeDone(res); - res.end(); + finishStreamWithError(mapped.error); return; } + if (terminalStreamError) { + finishStreamWithError(terminalStreamError); + return; + } + // Runs without a producer-owned terminal retain the visible-error fallback. const content = "Error: internal error"; writeAssistantContentChunk(res, { runId,