From 1464da8edd038ad5416da13dca1e66940b8e1a01 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 29 Jul 2026 05:42:36 -0400 Subject: [PATCH] fix(gateway): keep Responses replacement streams consistent (#115799) Preserve append-only text, fail impossible rewritten streams after usage resolves, and keep required client-tool output buffered until validation. Refs: #85261 Co-authored-by: Achyut Mukund <143781487+cropsgg@users.noreply.github.com> --- src/gateway/openresponses-http.test.ts | 244 +++++++++++++++++++++++++ src/gateway/openresponses-http.ts | 53 ++++-- 2 files changed, 284 insertions(+), 13 deletions(-) diff --git a/src/gateway/openresponses-http.test.ts b/src/gateway/openresponses-http.test.ts index fa8a39d62d8e..eca4dc96164d 100644 --- a/src/gateway/openresponses-http.test.ts +++ b/src/gateway/openresponses-http.test.ts @@ -348,6 +348,175 @@ describe("OpenResponses HTTP API (e2e)", () => { }, ); + it.each([ + { name: "rewritten", replacementText: "final answer" }, + { name: "shortened", replacementText: "dra" }, + { name: "cleared", replacementText: "" }, + ])( + "fails an official SDK Responses 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 response run ID"); + } + emitAgentEvent({ + runId, + stream: "assistant", + data: { text: "draft answer", delta: "draft answer" }, + }); + emitAgentEvent({ + runId, + stream: "assistant", + data: { text: replacementText, replace: true, phase: "commentary" }, + }); + emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end" } }); + return { + payloads: [{ text: replacementText }], + 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, + }); + const events = await client.responses.create({ + model: "openclaw", + input: "Reject an incompatible replacement snapshot.", + stream: true, + }); + + const deltas: string[] = []; + const failures: Array<{ + status: string | undefined; + code: string | undefined; + usage: { input_tokens: number; output_tokens: number; total_tokens: number } | undefined; + }> = []; + let completionCount = 0; + for await (const event of events) { + if (event.type === "response.output_text.delta") { + deltas.push(event.delta); + } else if (event.type === "response.failed") { + failures.push({ + status: event.response.status, + code: event.response.error?.code, + usage: event.response.usage + ? { + input_tokens: event.response.usage.input_tokens, + output_tokens: event.response.usage.output_tokens, + total_tokens: event.response.usage.total_tokens, + } + : undefined, + }); + } else if (event.type === "response.completed") { + completionCount += 1; + } + } + + expect(deltas).toEqual(["draft answer"]); + expect(failures).toEqual([ + { + status: "failed", + code: "server_error", + usage: { input_tokens: 11, output_tokens: 7, total_tokens: 18 }, + }, + ]); + expect(completionCount).toBe(0); + expect(agentCommand).toHaveBeenCalledTimes(1); + }, + ); + + it.each([ + { + name: "a producer replacement snapshot without a delta", + previousDelta: undefined, + replacementDelta: undefined, + expectedDeltas: "final answer", + }, + { + name: "a producer replacement snapshot with its own delta", + previousDelta: undefined, + replacementDelta: "final answer", + expectedDeltas: "final answer", + }, + { + name: "an append-compatible replacement after streamed partial text", + previousDelta: "final ", + replacementDelta: "answer", + expectedDeltas: "final answer", + }, + ])( + "keeps official SDK Responses text consistent for $name", + async ({ previousDelta, replacementDelta, expectedDeltas }) => { + agentCommand.mockClear(); + agentCommand.mockImplementationOnce((async (opts: unknown) => { + const runId = (opts as { runId?: string }).runId; + if (!runId) { + throw new Error("expected a streaming response 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 client = new OpenAI({ + apiKey: "test", + baseURL: `http://127.0.0.1:${enabledPort}/v1`, + defaultHeaders: { "x-openclaw-scopes": "operator.write" }, + maxRetries: 0, + }); + const events = await client.responses.create({ + model: "openclaw", + input: "Stream a replacement snapshot exactly once.", + stream: true, + }); + + const deltas: string[] = []; + const doneTexts: string[] = []; + const completedTexts: string[] = []; + for await (const event of events) { + if (event.type === "response.output_text.delta") { + deltas.push(event.delta); + } else if (event.type === "response.output_text.done") { + doneTexts.push(event.text); + } else if (event.type === "response.completed") { + completedTexts.push( + ...event.response.output.flatMap((item) => + item.type === "message" + ? item.content.flatMap((part) => (part.type === "output_text" ? [part.text] : [])) + : [], + ), + ); + } + } + + expect(deltas.join("")).toBe(expectedDeltas); + expect(doneTexts).toEqual(["final answer"]); + expect(completedTexts).toEqual(["final answer"]); + expect(agentCommand).toHaveBeenCalledTimes(1); + }, + ); + it.each([ { name: "JSON-object output", text: { format: { type: "json_object" } } }, { @@ -1681,6 +1850,81 @@ describe("OpenResponses HTTP API (e2e)", () => { expect(response?.output?.[1]?.name).toBe("get_weather"); }); + it.each([ + { name: "an initial replacement", previousDelta: undefined, replacementDelta: undefined }, + { name: "a replacement after buffered text", previousDelta: "draft", replacementDelta: "" }, + { + name: "a replacement with an explicit delta", + previousDelta: "draft", + replacementDelta: "final answer", + }, + ])( + "buffers $name until the required client tool is validated", + 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 response 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 }), + }, + }); + return { + payloads: [{ text: "final answer" }], + meta: { + stopReason: "tool_calls", + pendingToolCalls: [ + { id: "call_1", name: "get_weather", arguments: '{"city":"Taipei"}' }, + ], + }, + }; + }) as never); + + const res = await postResponses(enabledPort, { + stream: true, + model: "openclaw", + input: "check the weather", + tools: WEATHER_TOOL, + tool_choice: "required", + }); + + expect(res.status).toBe(200); + const events = parseSseEvents(await res.text()); + const deltas = events + .filter((event) => event.event === "response.output_text.delta") + .map((event) => (parseSseData(event) as { delta?: string }).delta); + expect(deltas).toEqual(["final answer"]); + + const completed = parseSseData(findSseEvent(events, "response.completed")) as { + response?: { + status?: string; + output?: Array<{ type?: string; content?: Array<{ text?: string }> }>; + }; + }; + expect(completed.response?.status).toBe("incomplete"); + expect(completed.response?.output?.map((item) => item.type)).toEqual([ + "message", + "function_call", + ]); + expect(completed.response?.output?.[0]?.content?.[0]?.text).toBe("final answer"); + }, + ); + it("buffers replaceable assistant events for streaming responses", async () => { const port = enabledPort; agentCommand.mockClear(); diff --git a/src/gateway/openresponses-http.ts b/src/gateway/openresponses-http.ts index 72868969432e..ae41c9931f57 100644 --- a/src/gateway/openresponses-http.ts +++ b/src/gateway/openresponses-http.ts @@ -920,8 +920,10 @@ export async function handleOpenResponsesHttpRequest( setSseHeaders(res); let accumulatedText = ""; + let streamedAssistantText = ""; let bufferedReplaceableAssistantContent = ""; let sawAssistantDelta = false; + let unrepresentableAssistantReplacement = false; let closed = false; let unsubscribe = () => {}; let stopWatchingDisconnect = () => {}; @@ -1062,32 +1064,38 @@ export async function handleOpenResponsesHttpRequest( const text = evt.data?.text; const replace = evt.data?.replace === true; - const hadAssistantDelta = sawAssistantDelta; if (replace && typeof text === "string") { accumulatedText = text; - } - - const content = resolveAssistantStreamDeltaText(evt); - if (!content) { - if ( - replace && - typeof text === "string" && - text && - !toolChoiceConstraint && - !hadAssistantDelta - ) { + if (toolChoiceConstraint) { + return; + } + // Responses deltas can only append. A conflicting replacement must + // fail after usage resolves instead of claiming inconsistent success. + if (!text.startsWith(streamedAssistantText)) { + unrepresentableAssistantReplacement = true; + return; + } + unrepresentableAssistantReplacement = false; + const replacementDelta = text.slice(streamedAssistantText.length); + if (replacementDelta) { sawAssistantDelta = true; + streamedAssistantText = text; writeSseEvent(res, { type: "response.output_text.delta", item_id: outputItemId, output_index: 0, content_index: 0, - delta: text, + delta: replacementDelta, }); } return; } + const content = resolveAssistantStreamDeltaText(evt); + if (!content) { + return; + } + // Hold assistant prose until the tool-choice contract is confirmed. A // `required`/pinned request must reject text-only turns, so streaming // deltas now could leak output we may have to fail. Buffered text is @@ -1100,6 +1108,7 @@ export async function handleOpenResponsesHttpRequest( sawAssistantDelta = true; accumulatedText += content; + streamedAssistantText += content; writeSseEvent(res, { type: "response.output_text.delta", @@ -1150,6 +1159,24 @@ export async function handleOpenResponsesHttpRequest( finalUsage = extractUsageFromResult(result); + if (unrepresentableAssistantReplacement) { + rememberResponseSession(); + finalizeFailedResponse( + createResponseResource({ + id: responseId, + model, + status: "failed", + output: [], + error: { + code: "server_error", + message: "Assistant output cannot be represented as an append-only response stream.", + }, + usage: finalUsage, + }), + ); + return; + } + // Check for pending client tool calls BEFORE maybeFinalize() because the // lifecycle:end event may already have requested finalization. const resultAny = result as { payloads?: Array<{ text?: string }>; meta?: unknown };