From 3ec0070a179e67c0aa7398d9a7e2b3e7ad85115e Mon Sep 17 00:00:00 2001 From: Dallin Romney Date: Thu, 13 Aug 2026 21:13:47 +0800 Subject: [PATCH] refactor(gateway): trim responses terminal coverage --- .../openai-http-terminal-outcome.test.ts | 80 ++ src/gateway/openai-http-terminal-outcome.ts | 87 +- src/gateway/openai-http.test.ts | 376 +----- src/gateway/openai-http.ts | 116 +- src/gateway/openresponses-http.test.ts | 1107 +++-------------- src/gateway/openresponses-http.ts | 97 +- 6 files changed, 369 insertions(+), 1494 deletions(-) create mode 100644 src/gateway/openai-http-terminal-outcome.test.ts diff --git a/src/gateway/openai-http-terminal-outcome.test.ts b/src/gateway/openai-http-terminal-outcome.test.ts new file mode 100644 index 000000000000..76189d3169c0 --- /dev/null +++ b/src/gateway/openai-http-terminal-outcome.test.ts @@ -0,0 +1,80 @@ +import { describe, expect, it } from "vitest"; +import { + resolveOpenAiHttpAgentRunTerminalOutcome, + resolveOpenAiHttpResultText, +} from "./openai-http-terminal-outcome.js"; + +describe("OpenAI HTTP terminal outcome", () => { + it.each([ + { + name: "accepts a visible fallback after a failed attempt", + result: { + payloads: [ + { text: "private provider failure", isError: true }, + { text: "fallback recovered" }, + ], + }, + reason: "completed", + }, + { + name: "accepts a media-only fallback after a failed attempt", + result: { + payloads: [ + { text: "private provider failure", isError: true }, + { mediaUrl: "https://example.invalid/recovered.png" }, + ], + }, + reason: "completed", + }, + { + name: "retains failure when only transient notices follow", + result: { + payloads: [ + { text: "private provider failure", isError: true }, + { text: "commentary", isCommentary: true }, + { text: "compaction", isCompactionNotice: true }, + { text: "fallback", isFallbackNotice: true }, + { text: "reasoning", isReasoningSnapshot: true }, + { text: "status", isStatusNotice: true }, + { text: "hidden", visible: false }, + ], + }, + reason: "failed", + }, + { + name: "retains failure when only whitespace follows", + result: { + payloads: [{ text: "private provider failure", isError: true }, { text: " \t\n " }], + }, + reason: "failed", + }, + { + name: "keeps replay-invalid success", + result: { meta: { replayInvalid: true } }, + reason: "completed", + }, + { + name: "classifies bare abort as cancellation", + result: { meta: { aborted: true } }, + reason: "aborted", + }, + { + name: "preserves hard timeout attribution", + result: { meta: { timeoutPhase: "provider" } }, + reason: "hard_timeout", + }, + ])("$name", ({ result, reason }) => { + expect(resolveOpenAiHttpAgentRunTerminalOutcome(result)).toMatchObject({ reason }); + }); + + it("filters historical error text from recovered output", () => { + expect( + resolveOpenAiHttpResultText({ + payloads: [ + { text: "private provider failure", isError: true }, + { text: "fallback recovered" }, + ], + }), + ).toBe("fallback recovered"); + }); +}); diff --git a/src/gateway/openai-http-terminal-outcome.ts b/src/gateway/openai-http-terminal-outcome.ts index 5b380f0e0185..3699d0f73985 100644 --- a/src/gateway/openai-http-terminal-outcome.ts +++ b/src/gateway/openai-http-terminal-outcome.ts @@ -5,30 +5,51 @@ import { type AgentRunTerminalOutcome, } from "../agents/agent-run-terminal-outcome.js"; import { hasVisibleAgentPayload } from "../agents/embedded-agent-runner/message-visibility.js"; +import { isReplyPayloadStatusNotice, type ReplyPayload } from "../auto-reply/reply-payload.js"; + +type LifecycleData = NonNullable< + Parameters[0]["data"] +>; type OpenAiHttpAgentResult = { - payloads?: Array<{ - isError?: boolean; - isCommentary?: boolean; - isCompactionNotice?: boolean; - isFallbackNotice?: boolean; - isReasoningSnapshot?: boolean; - isStatusNotice?: boolean; - text?: string; - visible?: boolean; - }>; - meta?: { - aborted?: boolean; - error?: unknown; - stopReason?: unknown; - livenessState?: unknown; - timeoutPhase?: unknown; - providerStarted?: unknown; - startedAt?: unknown; - endedAt?: unknown; - }; + payloads?: ReplyPayload[]; + meta?: LifecycleData; }; +function isTerminalPayload(payload: ReplyPayload): boolean { + if (payload.isError === true) { + return true; + } + if ( + payload.isCommentary === true || + payload.isReasoningSnapshot === true || + isReplyPayloadStatusNotice(payload) || + (payload as ReplyPayload & { visible?: unknown }).visible === false + ) { + return false; + } + return hasVisibleAgentPayload( + { payloads: [payload] }, + { + includeErrorPayloads: false, + includeReasoningPayloads: false, + includeSilentReplyPayloads: false, + }, + ); +} + +/** Return model-visible result text without leaking historical error payloads. */ +export function resolveOpenAiHttpResultText(result: unknown): string { + const payloads = (result as OpenAiHttpAgentResult | null | undefined)?.payloads; + return Array.isArray(payloads) + ? payloads + .filter((payload) => payload.isError !== true) + .map((payload) => (typeof payload.text === "string" ? payload.text : "")) + .filter(Boolean) + .join("\n\n") + : ""; +} + /** Preserve real provider failures even when the agent resolves its result. */ export function resolveOpenAiHttpAgentRunTerminalOutcome( result: unknown, @@ -38,32 +59,12 @@ export function resolveOpenAiHttpAgentRunTerminalOutcome( const meta = agentResult?.meta; // Completed tool calls can intentionally make a successful turn unsafe to // replay. Replay safety alone is not a provider or terminal-run failure. - // Recovery may retain a failed attempt before its final visible reply. - // Only the last visible/error payload owns the HTTP terminal result. - const terminalPayload = agentResult?.payloads?.findLast( - (payload) => - payload.isError === true || - (payload.isCommentary !== true && - payload.isCompactionNotice !== true && - payload.isFallbackNotice !== true && - payload.isReasoningSnapshot !== true && - payload.isStatusNotice !== true && - payload.visible !== false && - hasVisibleAgentPayload( - { payloads: [payload] }, - { - includeErrorPayloads: false, - includeReasoningPayloads: false, - includeSilentReplyPayloads: false, - }, - )), - ); - const resultFailed = meta?.error != null || terminalPayload?.isError === true; - + // Only the last real visible/error payload owns recovered fallback state. + const terminalPayload = agentResult?.payloads?.findLast(isTerminalPayload); return mergeAgentRunTerminalOutcome( previous, buildAgentRunTerminalOutcomeFromLifecycleEvent({ - phase: resultFailed ? "error" : "end", + phase: meta?.error != null || terminalPayload?.isError === true ? "error" : "end", data: meta, }), ); diff --git a/src/gateway/openai-http.test.ts b/src/gateway/openai-http.test.ts index a4a8dd17e00a..d22e3e03e008 100644 --- a/src/gateway/openai-http.test.ts +++ b/src/gateway/openai-http.test.ts @@ -24,7 +24,6 @@ import { isGatewaySubordinateWorkAdmissionClosed, } from "../process/gateway-work-admission.js"; import { withEnvAsync } from "../test-utils/env.js"; -import { resolveOpenAiHttpAgentRunTerminalOutcome } from "./openai-http-terminal-outcome.js"; import { buildAssistantDeltaResult } from "./test-helpers.agent-results.js"; import { agentCommandMock, @@ -37,8 +36,6 @@ import { installGatewayTestHooks({ scope: "suite" }); -const agentCommand = agentCommandMock; - let startGatewayServer: typeof import("./server.js").startGatewayServer; let enabledServer: Awaited>; let enabledPort: number; @@ -137,36 +134,6 @@ function parseSseDataLines(text: string): string[] { .map((line) => line.slice("data: ".length)); } -const PRESERVED_STREAM_FAILURE_CASES = [ - { - name: "an exhausted fallback result", - text: "Terminal tool summary", - lifecycle: { fallbackExhaustedFailure: true }, - error: { - kind: "incomplete_turn", - message: "raw exhausted provider detail should stay private", - fallbackSafe: true, - terminalPresentation: true, - }, - }, - { - name: "a non-replayable error result", - text: "Command may have changed state", - lifecycle: { replayInvalid: true }, - error: { - kind: "incomplete_turn", - message: "raw non-replayable provider detail should stay private", - fallbackSafe: false, - }, - }, -] as const; - -const PRESERVED_STREAM_LIFECYCLE_CASES = [ - { label: "after an error lifecycle", emitError: true, emitEnd: false }, - { label: "without an error lifecycle", emitError: false, emitEnd: false }, - { label: "after a superseded error lifecycle", emitError: true, emitEnd: true }, -] as const; - type FirstAgentCommandOptions = { clientTools?: Array<{ function?: { @@ -2030,167 +1997,25 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { expect(agentCommandMock).toHaveBeenCalledTimes(1); }); - it("classifies a bare aborted result as cancellation rather than timeout", () => { - expect(resolveOpenAiHttpAgentRunTerminalOutcome({ meta: { aborted: true } })).toMatchObject({ - reason: "aborted", - status: "error", - stopReason: "aborted", - }); - }); - - it("completes a successful non-replayable tool turn", async () => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: "FAKE_PLUGIN_OK fake_plugin_tool_17" }], - meta: { - agentMeta: { usage: { input: 128, output: 40, total: 168 } }, - livenessState: "working", - replayInvalid: true, - stopReason: "stop", - }, - } as never); - - const res = await postChatCompletions(enabledPort, { - model: "openclaw", - messages: [{ role: "user", content: "tool search qa check target=fake_plugin_tool_17" }], - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - const response = JSON.parse(body) as { - choices?: Array<{ finish_reason?: string; message?: { content?: string } }>; - usage?: { prompt_tokens: number; completion_tokens: number; total_tokens: number }; - }; - expect(response.choices?.[0]?.message?.content).toBe("FAKE_PLUGIN_OK fake_plugin_tool_17"); - expect(response.choices?.[0]?.finish_reason).toBe("stop"); - expect(response.usage).toEqual({ - prompt_tokens: 128, - completion_tokens: 40, - total_tokens: 168, - }); - expect(agentCommand).toHaveBeenCalledTimes(1); - }); - it.each([false, true])( - "completes a recovered chat after an earlier error payload with stream=%s", + "fails a resolved terminal error without exposing provider details with stream=%s", async (stream) => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [ - { text: "Historical failed provider attempt", isError: true }, - { text: "fallback recovered" }, - {}, - ], + const privateFailure = "private terminal payload"; + const privateError = "private provider detail"; + agentCommandMock.mockClear(); + agentCommandMock.mockResolvedValueOnce({ + payloads: [{ text: privateFailure, isError: true }], + meta: { + error: { kind: "incomplete_turn", message: privateError }, + agentMeta: { usage: { input: 7, output: 3, total: 10 } }, + }, } as never); const res = await postChatCompletions(enabledPort, { - stream, model: "openclaw", messages: [{ role: "user", content: "hi" }], - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - - if (stream) { - const data = parseSseDataLines(body); - const chunks = data - .filter((line) => line !== "[DONE]") - .map((line) => JSON.parse(line) as Record); - expect(chunks.filter((chunk) => "error" in chunk)).toHaveLength(0); - expect( - chunks - .flatMap( - (chunk) => - (chunk.choices as Array<{ finish_reason?: string | null }> | undefined) ?? [], - ) - .filter((choice) => choice.finish_reason === "stop"), - ).toHaveLength(1); - expect(data.at(-1)).toBe("[DONE]"); - expect(body).toContain("fallback recovered"); - } else { - const response = JSON.parse(body) as { - choices?: Array<{ finish_reason?: string; message?: { content?: string } }>; - }; - expect(response.choices?.[0]?.message?.content).toContain("fallback recovered"); - expect(response.choices?.[0]?.finish_reason).toBe("stop"); - } - - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "completes a recovered media-only chat after an earlier error payload with stream=%s", - async (stream) => { - const privateFailure = "Historical private provider failure"; - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [ - { text: privateFailure, isError: true }, - { - mediaUrl: "https://example.invalid/recovered-image.png", - mediaUrls: ["https://example.invalid/recovered-document.pdf"], - }, - ], - } as never); - - const res = await postChatCompletions(enabledPort, { stream, - model: "openclaw", - messages: [{ role: "user", content: "recover the generated attachment" }], - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - - if (stream) { - const data = parseSseDataLines(body); - const chunks = data - .filter((line) => line !== "[DONE]") - .map((line) => JSON.parse(line) as Record); - expect(chunks.filter((chunk) => "error" in chunk)).toHaveLength(0); - expect( - chunks - .flatMap( - (chunk) => - (chunk.choices as Array<{ finish_reason?: string | null }> | undefined) ?? [], - ) - .filter((choice) => choice.finish_reason === "stop"), - ).toHaveLength(1); - expect(data.filter((line) => line === "[DONE]")).toHaveLength(1); - expect(data.at(-1)).toBe("[DONE]"); - } else { - const response = JSON.parse(body) as { - choices?: Array<{ finish_reason?: string; message?: { content?: string } }>; - }; - expect(response.choices?.[0]?.finish_reason).toBe("stop"); - } - - expect(body).not.toContain("api_error"); - expect(body).not.toContain(privateFailure); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "preserves a failed chat when only transient notices follow with stream=%s", - async (stream) => { - const privateFailure = "Historical private provider failure"; - const notices = [ - { text: "Private commentary notice", isCommentary: true }, - { text: "Private compaction notice", isCompactionNotice: true }, - { text: "Private fallback notice", isFallbackNotice: true }, - { text: "Private reasoning snapshot", isReasoningSnapshot: true }, - { text: "Private status notice", isStatusNotice: true }, - { text: "Private hidden notice", visible: false }, - ]; - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: privateFailure, isError: true }, ...notices], - } as never); - - const res = await postChatCompletions(enabledPort, { - stream, - model: "openclaw", - messages: [{ role: "user", content: "finish the failed request" }], + ...(stream ? { stream_options: { include_usage: true } } : {}), }); const body = await res.text(); @@ -2203,15 +2028,7 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { expect(chunks.filter((chunk) => "error" in chunk)).toEqual([ { error: { message: "internal error", type: "api_error" } }, ]); - expect( - chunks - .flatMap( - (chunk) => - (chunk.choices as Array<{ finish_reason?: string | null }> | undefined) ?? [], - ) - .filter((choice) => choice.finish_reason === "stop"), - ).toHaveLength(0); - expect(data.filter((line) => line === "[DONE]")).toHaveLength(1); + expect(chunks.filter((chunk) => "usage" in chunk)).toHaveLength(1); expect(data.at(-1)).toBe("[DONE]"); } else { expect(res.status, body).toBe(502); @@ -2219,176 +2036,11 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { error: { message: "internal error", type: "api_error" }, }); } - expect(body).not.toContain(privateFailure); - for (const notice of notices) { - expect(body).not.toContain(notice.text); - } - expect(agentCommand).toHaveBeenCalledTimes(1); + expect(body).not.toContain(privateError); + expect(agentCommandMock).toHaveBeenCalledTimes(1); }, ); - - it.each([false, true])( - "preserves a failed chat when its final payload is whitespace with stream=%s", - async (stream) => { - const privateFailure = "Historical private provider failure"; - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: privateFailure, isError: true }, { text: " \t\n " }], - } as never); - - const res = await postChatCompletions(enabledPort, { - stream, - model: "openclaw", - messages: [{ role: "user", content: "hi" }], - }); - const body = await res.text(); - - if (stream) { - expect(res.status, body).toBe(200); - const data = parseSseDataLines(body); - const chunks = data - .filter((line) => line !== "[DONE]") - .map((line) => JSON.parse(line) as Record); - expect(chunks.filter((chunk) => "error" in chunk)).toEqual([ - { error: { message: "internal error", type: "api_error" } }, - ]); - expect( - chunks - .flatMap( - (chunk) => - (chunk.choices as Array<{ finish_reason?: string | null }> | undefined) ?? [], - ) - .filter((choice) => choice.finish_reason === "stop"), - ).toHaveLength(0); - expect(data.filter((line) => line === "[DONE]")).toHaveLength(1); - expect(data.at(-1)).toBe("[DONE]"); - } else { - expect(res.status, body).toBe(502); - expect(JSON.parse(body)).toEqual({ - error: { message: "internal error", type: "api_error" }, - }); - } - - expect(body).not.toContain(privateFailure); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each(PRESERVED_STREAM_FAILURE_CASES)( - "fails non-stream chat completions for $name without exposing provider details", - async ({ text, error }) => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text, isError: true }], - meta: { error }, - } as never); - - const res = await postChatCompletions(enabledPort, { - model: "openclaw", - messages: [{ role: "user", content: "hi" }], - }); - - expect(res.status).toBe(502); - const body = await res.text(); - expect(JSON.parse(body)).toEqual({ - error: { message: "internal error", type: "api_error" }, - }); - expect(body).not.toContain(text); - expect(body).not.toContain(error.message); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each( - PRESERVED_STREAM_FAILURE_CASES.flatMap((failure) => - PRESERVED_STREAM_LIFECYCLE_CASES.flatMap((lifecycleCase) => - [false, true].map((includeUsage) => ({ - name: failure.name, - text: failure.text, - lifecycle: failure.lifecycle, - error: failure.error, - includeUsage, - label: `${failure.name} ${includeUsage ? "with" : "without"} streamed usage`, - lifecycleLabel: lifecycleCase.label, - emitError: lifecycleCase.emitError, - emitEnd: lifecycleCase.emitEnd, - })), - ), - ), - )( - "fails the chat stream when $label resolves $lifecycleLabel", - async ({ text, lifecycle, error, includeUsage, emitError, emitEnd }) => { - const idleRootCount = getActiveGatewayRootWorkCount(); - 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 (emitError) { - emitAgentEvent({ - runId, - stream: "lifecycle", - data: { phase: "error", error: text, ...lifecycle }, - }); - } - if (emitEnd) { - emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end" } }); - } - return { - payloads: [{ text, isError: true }], - meta: { - stopReason: "end_turn", - error, - agentMeta: { usage: { input: 7, output: 3, total: 10 } }, - }, - }; - }) as never); - - const res = await postChatCompletions(enabledPort, { - stream: true, - ...(includeUsage ? { stream_options: { include_usage: true } } : {}), - model: "openclaw", - messages: [{ role: "user", content: "hi" }], - }); - expect(res.status).toBe(200); - - const body = await res.text(); - const data = parseSseDataLines(body); - expect(data.filter((line) => line === "[DONE]")).toHaveLength(1); - expect(data.at(-1)).toBe("[DONE]"); - - const chunks = data - .filter((line) => line !== "[DONE]") - .map((line) => JSON.parse(line) as Record); - expect(chunks.filter((chunk) => "error" in chunk)).toEqual([ - { error: { message: "internal error", type: "api_error" } }, - ]); - const finishReasons = chunks.flatMap( - (chunk) => (chunk.choices as Array<{ finish_reason?: string | null }> | undefined) ?? [], - ); - expect(finishReasons.some((choice) => choice.finish_reason === "stop")).toBe(false); - - const usageChunks = chunks.filter((chunk) => "usage" in chunk); - if (includeUsage) { - expect(usageChunks).toHaveLength(1); - expect(usageChunks[0]?.choices).toEqual([]); - expect(usageChunks[0]?.usage).toEqual({ - prompt_tokens: 7, - completion_tokens: 3, - total_tokens: 10, - }); - } else { - expect(usageChunks).toHaveLength(0); - } - expect(body).not.toContain(error.message); - expect(body).not.toContain(text); - expect(agentCommand).toHaveBeenCalledTimes(1); - await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(idleRootCount)); - }, - ); - it("forwards response_format into streamParams", async () => { const port = enabledPort; const mockAgentOnce = (payloads: Array<{ text: string }>) => { diff --git a/src/gateway/openai-http.ts b/src/gateway/openai-http.ts index dc258107a91a..f558189a7ef1 100644 --- a/src/gateway/openai-http.ts +++ b/src/gateway/openai-http.ts @@ -72,7 +72,10 @@ import { import { normalizeInputHostnameAllowlist } from "./input-allowlist.js"; import { resolveAgentRunUsage } from "./openai-agent-run-usage.js"; import { resolveOpenAiCompatError, validateOpenAiSamplingParams } from "./openai-compat-errors.js"; -import { resolveOpenAiHttpAgentRunTerminalOutcome } from "./openai-http-terminal-outcome.js"; +import { + resolveOpenAiHttpAgentRunTerminalOutcome, + resolveOpenAiHttpResultText, +} from "./openai-http-terminal-outcome.js"; import { isToolChoiceConstraintSatisfied, resolveUnsatisfiedToolChoiceMessage, @@ -742,33 +745,6 @@ function coerceRequest(val: unknown): OpenAiChatCompletionRequest { return val as OpenAiChatCompletionRequest; } -function resolveAgentResponseText(result: unknown): string { - const payloads = (result as { payloads?: Array<{ isError?: boolean; text?: string }> } | null) - ?.payloads; - if (!Array.isArray(payloads) || payloads.length === 0) { - return "No response from OpenClaw."; - } - const content = payloads - .filter((payload) => payload.isError !== true) - .map((p) => (typeof p.text === "string" ? p.text : "")) - .filter(Boolean) - .join("\n\n"); - return content || "No response from OpenClaw."; -} - -function resolveAgentResponseCommentary(result: unknown): string { - const payloads = (result as { payloads?: Array<{ isError?: boolean; text?: string }> } | null) - ?.payloads; - if (!Array.isArray(payloads) || payloads.length === 0) { - return ""; - } - return payloads - .filter((payload) => payload.isError !== true) - .map((p) => (typeof p.text === "string" ? p.text : "")) - .filter(Boolean) - .join("\n\n"); -} - type PendingToolCall = { id?: unknown; name?: unknown; @@ -1138,7 +1114,7 @@ export async function handleOpenAiHttpRequest( } if (stopReason === "tool_calls" && pendingToolCalls && pendingToolCalls.length > 0) { - const commentary = resolveAgentResponseCommentary(result); + const commentary = resolveOpenAiHttpResultText(result); sendJson(res, 200, { id: runId, object: "chat.completion", @@ -1163,7 +1139,7 @@ export async function handleOpenAiHttpRequest( }); return true; } - const content = resolveAgentResponseText(result); + const content = resolveOpenAiHttpResultText(result) || "No response from OpenClaw."; sendJson(res, 200, { id: runId, @@ -1213,15 +1189,9 @@ export async function handleOpenAiHttpRequest( let bufferedAssistantContent = ""; let bufferedReplaceableAssistantContent = ""; let finalUsage: OpenAiChatCompletionsUsage | undefined; - type StreamFinalization = - | { - status: "completed"; - finishReason: "stop" | "tool_calls"; - outcome?: AgentRunTerminalOutcome; - } - | { status: "failed"; outcome: AgentRunTerminalOutcome }; - let finalizeRequested: StreamFinalization | null = null; - const readFinalization = (): StreamFinalization | null => finalizeRequested; + let finalizeRequested = false; + let finalizeFinishReason: "stop" | "tool_calls" = "stop"; + let terminalOutcome: AgentRunTerminalOutcome | undefined; let finalizeScheduled = false; let resultResolved = false; let closed = false; @@ -1231,14 +1201,17 @@ export async function handleOpenAiHttpRequest( let unsubscribe = () => {}; let stopWatchingDisconnect = () => {}; - const finalizeFailedStream = (error: { message: string; type: string; code?: string }) => { + const finishStreamWithError = ( + error: { message: string; type: string; code?: string }, + includeUsage = false, + ) => { if (closed) { return; } closed = true; stopWatchingDisconnect(); unsubscribe(); - if (streamIncludeUsage && finalUsage) { + if (includeUsage && streamIncludeUsage && finalUsage) { writeUsageChunk(res, { runId, model, usage: finalUsage }); } writeSse(res, { error }); @@ -1251,8 +1224,8 @@ export async function handleOpenAiHttpRequest( return; } // Resolved preserved errors are failures, not successful assistant stops. - if (finalizeRequested.status === "failed") { - finalizeFailedStream({ message: "internal error", type: "api_error" }); + if (terminalOutcome?.reason && terminalOutcome.reason !== "completed") { + finishStreamWithError({ message: "internal error", type: "api_error" }, true); return; } if (streamIncludeUsage && !finalUsage) { @@ -1265,12 +1238,7 @@ export async function handleOpenAiHttpRequest( if (closed) { return; } - const finalization = finalizeRequested; - if (!finalization) { - finalizeScheduled = false; - return; - } - if (finalization.status === "failed") { + if (terminalOutcome?.reason && terminalOutcome.reason !== "completed") { finalizeScheduled = false; maybeFinalize(); return; @@ -1283,7 +1251,7 @@ export async function handleOpenAiHttpRequest( stopWatchingDisconnect(); unsubscribe(); if (!wroteStopChunk) { - writeAssistantFinishChunk(res, { runId, model, finishReason: finalization.finishReason }); + writeAssistantFinishChunk(res, { runId, model, finishReason: finalizeFinishReason }); wroteStopChunk = true; } if (streamIncludeUsage && finalUsage) { @@ -1299,22 +1267,17 @@ export async function handleOpenAiHttpRequest( outcome?: AgentRunTerminalOutcome, ) => { // Failed attempts remain provisional until a recovered fallback settles. - const previous = readFinalization(); - const preservedFinishReason = - previous?.status === "completed" && previous.finishReason === "tool_calls" - ? "tool_calls" - : finishReason; - const preservedOutcome = outcome ?? previous?.outcome; - finalizeRequested = { - status: "completed", - finishReason: preservedFinishReason, - ...(preservedOutcome ? { outcome: preservedOutcome } : {}), - }; + if (finishReason === "tool_calls") { + finalizeFinishReason = finishReason; + } + terminalOutcome = outcome ?? terminalOutcome; + finalizeRequested = true; maybeFinalize(); }; const requestFailedStream = (outcome: AgentRunTerminalOutcome) => { - finalizeRequested = { status: "failed", outcome }; + terminalOutcome = outcome; + finalizeRequested = true; maybeFinalize(); }; @@ -1405,7 +1368,7 @@ export async function handleOpenAiHttpRequest( phase, data: evt.data, }); - const outcome = mergeAgentRunTerminalOutcome(finalizeRequested?.outcome, incomingOutcome); + const outcome = mergeAgentRunTerminalOutcome(terminalOutcome, incomingOutcome); if (outcome.reason === "completed") { requestFinalize("stop", outcome); } else { @@ -1415,18 +1378,6 @@ export async function handleOpenAiHttpRequest( } }); - 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(); @@ -1458,18 +1409,12 @@ export async function handleOpenAiHttpRequest( } finalUsage = resolveChatCompletionUsage(result); - const resultOutcome = resolveOpenAiHttpAgentRunTerminalOutcome(result); - if (resultOutcome.reason !== "completed") { - requestFailedStream( - mergeAgentRunTerminalOutcome(readFinalization()?.outcome, resultOutcome), - ); - return; - } + const outcome = resolveOpenAiHttpAgentRunTerminalOutcome(result, terminalOutcome); + terminalOutcome = outcome; if (terminalStreamError) { finishStreamWithError(terminalStreamError); return; } - const outcome = resolveOpenAiHttpAgentRunTerminalOutcome(result, readFinalization()?.outcome); if (outcome.reason !== "completed") { requestFailedStream(outcome); return; @@ -1503,7 +1448,7 @@ export async function handleOpenAiHttpRequest( if (!sawAssistantDelta) { const commentary = bufferedAssistantContent || - resolveAgentResponseCommentary(result) || + resolveOpenAiHttpResultText(result) || bufferedReplaceableAssistantContent; if (commentary) { sawAssistantDelta = true; @@ -1530,9 +1475,8 @@ export async function handleOpenAiHttpRequest( } const content = - resolveAgentResponseCommentary(result) || + resolveOpenAiHttpResultText(result) || bufferedReplaceableAssistantContent || - resolveAgentResponseText(result) || "No response from OpenClaw."; sawAssistantDelta = true; diff --git a/src/gateway/openresponses-http.test.ts b/src/gateway/openresponses-http.test.ts index 5ca7b17e16d1..7b2c22a66899 100644 --- a/src/gateway/openresponses-http.test.ts +++ b/src/gateway/openresponses-http.test.ts @@ -20,7 +20,6 @@ import { } from "../process/gateway-work-admission.js"; import { withEnvAsync } from "../test-utils/env.js"; import { IMAGE_ONLY_USER_MESSAGE } from "./agent-prompt.js"; -import type { Usage } from "./open-responses.schema.js"; import { buildAssistantDeltaResult } from "./test-helpers.agent-results.js"; import { agentCommandMock, @@ -47,31 +46,6 @@ vi.mock("../infra/net/fetch-guard.js", async () => { installGatewayTestHooks({ scope: "suite" }); -const agentCommand = agentCommandMock; - -function expectedResponsesUsage( - inputTokens: number, - outputTokens: number, - totalTokens: number, - details: { - cachedTokens?: number; - cacheWriteTokens?: number; - reasoningTokens?: number; - } = {}, -): Usage { - const usage: OpenAI.Responses.ResponseUsage = { - input_tokens: inputTokens, - input_tokens_details: { - cached_tokens: details.cachedTokens ?? 0, - cache_write_tokens: details.cacheWriteTokens ?? 0, - }, - output_tokens: outputTokens, - output_tokens_details: { reasoning_tokens: details.reasoningTokens ?? 0 }, - total_tokens: totalTokens, - }; - return usage; -} - let enabledServer: Awaited>; let enabledPort: number; let openResponsesTesting: { @@ -289,36 +263,6 @@ const STREAM_FAILURE_CASES = [ }, ] as const; -const PRESERVED_STREAM_FAILURE_CASES = [ - { - name: "an exhausted fallback result", - text: "Terminal tool summary", - lifecycle: { fallbackExhaustedFailure: true }, - error: { - kind: "incomplete_turn", - message: "raw exhausted provider detail should stay private", - fallbackSafe: true, - terminalPresentation: true, - }, - }, - { - name: "a non-replayable error result", - text: "Command may have changed state", - lifecycle: { replayInvalid: true }, - error: { - kind: "incomplete_turn", - message: "raw non-replayable provider detail should stay private", - fallbackSafe: false, - }, - }, -] as const; - -const PRESERVED_STREAM_LIFECYCLE_CASES = [ - { label: "after an error lifecycle", emitError: true, emitEnd: false }, - { label: "without an error lifecycle", emitError: false, emitEnd: false }, - { label: "after a superseded error lifecycle", emitError: true, emitEnd: true }, -] as const; - function buildUrlInputMessage(params: { kind: "input_file" | "input_image"; url: string; @@ -1858,261 +1802,102 @@ describe("OpenResponses HTTP API (e2e)", () => { expect(agentCommandMock).toHaveBeenCalledTimes(1); }); - it.each([false, true])( - "replays canonical OpenAI SDK response output with stream=%s", - async (stream) => { - const reasoning: OpenAI.Responses.ResponseReasoningItem = { - type: "reasoning", - id: "rs_replay_1", - summary: [{ type: "summary_text", text: "Locate the current weather." }], - content: [{ type: "reasoning_text", text: "Use the weather tool." }], - encrypted_content: null, - status: "completed", - }; - const assistant: OpenAI.Responses.ResponseOutputMessage = { - type: "message", - id: "msg_replay_1", - role: "assistant", - phase: null, - status: "completed", - content: [ - { - type: "output_text", - text: "Checking the weather.", - annotations: [], - logprobs: [ - { - token: "Checking", - bytes: [67, 104, 101, 99, 107, 105, 110, 103], - logprob: -0.25, - top_logprobs: [{ token: "Checking", bytes: [67], logprob: -0.25 }], - }, - ], - }, - { - type: "output_text", - text: "Tool details retained.", - annotations: [], - logprobs: [], - }, - ], - }; - const functionCall: OpenAI.Responses.ResponseFunctionToolCall = { - type: "function_call", - id: "fc_replay_1", - call_id: "call_replay_1", - name: "get_weather", - arguments: '{"city":"Taipei"}', - caller: { type: "direct" }, - namespace: "weather", - status: "completed", - }; - const functionOutput: OpenAI.Responses.ResponseFunctionToolCallOutputItem = { - type: "function_call_output", - id: "fc_output_replay_1", - call_id: "call_replay_1", - output: '{"temperature":"72F"}', - caller: { type: "program", caller_id: "program_replay_1" }, - created_by: "weather-worker", - status: "completed", - }; + it("replays strict SDK output and preserves structured tool data without fetching it", async () => { + const structuredOutput = [ + { + type: "input_text", + text: "Structured weather: 72F.", + prompt_cache_breakpoint: { mode: "explicit" }, + }, + { + type: "input_image", + detail: "auto", + image_url: "https://example.invalid/sdk-tool-image.png", + }, + { + type: "input_file", + detail: "auto", + file_url: "https://example.invalid/sdk-tool-output.txt", + filename: "sdk-tool-output.txt", + }, + ] satisfies OpenAI.Responses.ResponseFunctionCallOutputItemList; - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ payloads: [{ text: "Taipei is 72F." }] } as never); - - const res = await postResponses(enabledPort, { - model: "openclaw", - stream, - input: [ - { type: "message", role: "user", content: "Check the weather." }, - reasoning, - assistant, - functionCall, - functionOutput, - { type: "message", role: "user", content: "Summarize the result." }, - ], - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - if (stream) { - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(1); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - } else { - expect((JSON.parse(body) as { status?: string }).status).toBe("completed"); - } - const prompt = firstAgentOpts().message; - expect(prompt).toContain("Checking the weather."); - expect(prompt).toContain('{"temperature":"72F"}'); - expect(prompt).toContain("Summarize the result."); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "replays strict OpenAI SDK program-call ownership metadata with stream=%s", - async (stream) => { - const functionCall: OpenAI.Responses.ResponseFunctionToolCallItem = { - type: "function_call", - id: "fc_owned_replay_1", - call_id: "call_owned_replay_1", - name: "get_weather", - arguments: '{"city":"Taipei"}', - caller: { type: "program", caller_id: "program_owned_replay_1" }, - namespace: "weather", - created_by: "weather-program", - status: "completed", - }; - const functionOutput: OpenAI.Responses.ResponseFunctionToolCallOutputItem = { - type: "function_call_output", - id: "fc_output_owned_replay_1", - call_id: "call_owned_replay_1", - output: '{"temperature":"72F"}', - caller: { type: "direct" }, - created_by: "weather-worker", - status: "completed", - }; - - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: "Owned replay accepted." }], - } as never); - const res = await postResponses(enabledPort, { - model: "openclaw", - stream, - input: [ - functionCall, - functionOutput, - { type: "message", role: "user", content: "Summarize the owned tool result." }, - ], - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - if (stream) { - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - } else { - expect((JSON.parse(body) as { status?: string }).status).toBe("completed"); - } - expect(firstAgentOpts().message).toContain('{"temperature":"72F"}'); - expect(firstAgentOpts().message).toContain("Summarize the owned tool result."); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "replays structured OpenAI SDK function output without fetching media with stream=%s", - async (stream) => { - const output = [ + agentCommandMock.mockClear(); + agentCommandMock.mockResolvedValueOnce({ payloads: [{ text: "Taipei is 72F." }] } as never); + const res = await postResponses(enabledPort, { + model: "openclaw", + input: [ { - type: "input_text", - text: "Structured weather: 72F.", - prompt_cache_breakpoint: { mode: "explicit" }, - }, + type: "reasoning", + id: "rs_replay_1", + summary: [{ type: "summary_text", text: "Locate the current weather." }], + content: [{ type: "reasoning_text", text: "Use the weather tool." }], + encrypted_content: null, + status: "completed", + } satisfies OpenAI.Responses.ResponseReasoningItem, { - type: "input_image", - detail: "auto", - image_url: "https://example.invalid/sdk-tool-image.png", - }, + type: "message", + id: "msg_replay_1", + role: "assistant", + phase: null, + status: "completed", + content: [ + { + type: "output_text", + text: "Checking the weather.", + annotations: [ + { + type: "url_citation", + start_index: 0, + end_index: 8, + title: "Weather", + url: "https://example.invalid/weather", + }, + ], + logprobs: [ + { + token: "Checking", + bytes: null, + logprob: -0.25, + top_logprobs: [{ token: "Checking", bytes: null, logprob: -0.25 }], + }, + ], + }, + ], + } satisfies OpenAI.Responses.ResponseOutputMessage, { - type: "input_file", - detail: "auto", - file_url: "https://example.invalid/sdk-tool-output.txt", - filename: "sdk-tool-output.txt", - }, - ] satisfies OpenAI.Responses.ResponseFunctionCallOutputItemList; - const structuredOutput: OpenAI.Responses.ResponseFunctionToolCallOutputItem = { - type: "function_call_output", - id: "fc_output_structured_1", - call_id: "call_structured_1", - output, - status: "completed", - }; - - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: "Structured replay accepted." }], - } as never); - - const res = await postResponses(enabledPort, { - model: "openclaw", - stream, - input: [ - structuredOutput, - { type: "message", role: "user", content: "Summarize the structured result." }, - ], - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - if (stream) { - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - } else { - expect((JSON.parse(body) as { status?: string }).status).toBe("completed"); - } - const opts = firstAgentOpts(); - expect(opts.message).toContain(JSON.stringify(output)); - expect(opts.message).toContain("Structured weather: 72F."); - expect(opts.message).toContain("https://example.invalid/sdk-tool-image.png"); - expect(opts.message).toContain("https://example.invalid/sdk-tool-output.txt"); - expect(opts.images ?? []).toEqual([]); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "replays nullable output-token byte vectors with stream=%s", - async (stream) => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ payloads: [{ text: "Replay accepted." }] } as never); - - const res = await postResponses(enabledPort, { - model: "openclaw", - stream, - input: [ - { - type: "message", - id: "msg_nullable_logprobs_1", - role: "assistant", - status: "completed", - content: [ - { - type: "output_text", - text: "Nullable byte replay.", - annotations: [], - logprobs: [ - { - token: "nullable", - bytes: null, - logprob: -0.5, - top_logprobs: [{ token: "nullable", bytes: null, logprob: -0.5 }], - }, - ], - }, - ], - }, - { type: "message", role: "user", content: "Continue safely." }, - ], - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - if (stream) { - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(1); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - } else { - expect((JSON.parse(body) as { status?: string }).status).toBe("completed"); - } - expect(firstAgentOpts().message).toContain("Nullable byte replay."); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); + type: "function_call", + id: "fc_replay_1", + call_id: "call_replay_1", + name: "get_weather", + arguments: '{"city":"Taipei"}', + caller: { type: "program", caller_id: "program_replay_1" }, + namespace: "weather", + created_by: "weather-program", + status: "completed", + } satisfies OpenAI.Responses.ResponseFunctionToolCallItem, + { + type: "function_call_output", + id: "fc_output_replay_1", + call_id: "call_replay_1", + output: structuredOutput, + caller: { type: "direct" }, + created_by: "weather-worker", + status: "completed", + } satisfies OpenAI.Responses.ResponseFunctionToolCallOutputItem, + { type: "message", role: "user", content: "Summarize the result." }, + ], + }); + const body = await res.text(); + expect(res.status, body).toBe(200); + expect((JSON.parse(body) as { status?: string }).status).toBe("completed"); + const opts = firstAgentOpts(); + expect(opts.message).toContain("Checking the weather."); + expect(opts.message).toContain(JSON.stringify(structuredOutput)); + expect(opts.message).toContain("Summarize the result."); + expect(opts.images ?? []).toEqual([]); + expect(agentCommandMock).toHaveBeenCalledTimes(1); + }); it.each([ { @@ -2123,42 +1908,24 @@ describe("OpenResponses HTTP API (e2e)", () => { summary: [{ type: "untrusted_summary", text: "reject this" }], }, }, - { - name: "an unknown function-output field", - item: { - type: "function_call_output", - id: "fc_output_invalid_1", - call_id: "call_invalid_1", - output: "ok", - status: "completed", - untrusted_extra: true, - }, - }, { name: "an unknown nested function-caller field", item: { type: "function_call", - id: "fc_caller_invalid_1", - call_id: "call_caller_invalid_1", + call_id: "call_invalid_1", name: "get_weather", - arguments: '{"city":"Taipei"}', - caller: { - type: "direct", - caller_id: "untrusted-program-escalation", - }, + arguments: "{}", + caller: { type: "direct", caller_id: "untrusted-program-escalation" }, }, }, { - name: "an unknown nested structured function-output field", + name: "an unknown nested structured-output field", item: { type: "function_call_output", - id: "fc_output_structured_invalid_1", - call_id: "call_structured_invalid_1", - status: "completed", + call_id: "call_invalid_1", output: [ { type: "input_image", - detail: "auto", image_url: "https://example.invalid/sdk-tool-image.png", untrusted_extra: true, }, @@ -2169,14 +1936,11 @@ describe("OpenResponses HTTP API (e2e)", () => { name: "an unknown nested token-logprob field", item: { type: "message", - id: "msg_invalid_logprobs_1", role: "assistant", - status: "completed", content: [ { type: "output_text", text: "Reject non-SDK token metadata.", - annotations: [], logprobs: [ { token: "reject", @@ -2191,446 +1955,72 @@ describe("OpenResponses HTTP API (e2e)", () => { }, }, ])("rejects non-SDK replay input containing $name", async ({ item }) => { - agentCommand.mockClear(); - + agentCommandMock.mockClear(); const res = await postResponses(enabledPort, { model: "openclaw", - input: [item, { type: "message", role: "user", content: "Reject the invalid replay." }], + input: [item, { type: "message", role: "user", content: "Reject invalid replay." }], }); const body = await res.text(); expect(res.status, body).toBe(400); expect((JSON.parse(body) as { error?: { type?: string } }).error?.type).toBe( "invalid_request_error", ); - expect(agentCommand).not.toHaveBeenCalled(); - }); - - it.each([ - { - name: "without lifecycle metadata", - result: { payloads: [{ text: "FAKE_PLUGIN_OK fake_plugin_tool_17" }] }, - usage: expectedResponsesUsage(0, 0, 0), - }, - { - name: "with usage-only metadata", - result: { - payloads: [{ text: "FAKE_PLUGIN_OK fake_plugin_tool_17" }], - meta: { agentMeta: { usage: { input: 128, output: 40, total: 168 } } }, - }, - usage: expectedResponsesUsage(128, 40, 168), - }, - { - name: "after a successfully completed non-replayable tool turn", - result: { - payloads: [{ text: "FAKE_PLUGIN_OK fake_plugin_tool_17" }], - meta: { - agentMeta: { usage: { input: 128, output: 40, total: 168 } }, - livenessState: "working", - replayInvalid: true, - stopReason: "stop", - }, - }, - usage: expectedResponsesUsage(128, 40, 168), - }, - ])("completes a successful QA tool-search result $name", async ({ result, usage }) => { - const previousAgentsConfig = testState.agentsConfig; - testState.agentsConfig = { list: [{ id: "main" }, { id: "qa" }] }; - resetConfigRuntimeState(); - try { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce(result as never); - - const res = await postResponses( - enabledPort, - { - model: "openclaw/qa", - input: [ - { - type: "message", - role: "user", - content: [ - { - type: "input_text", - text: "tool search qa check target=fake_plugin_tool_17", - }, - ], - }, - ], - max_output_tokens: 256, - stream: false, - }, - { - "x-openclaw-agent": "qa", - "x-openclaw-session-key": "tool-search-gateway-normal", - }, - ); - const body = await res.text(); - expect(res.status, body).toBe(200); - const response = JSON.parse(body) as { - status?: string; - output?: Array<{ type?: string; content?: Array<{ text?: string }> }>; - usage?: { input_tokens: number; output_tokens: number; total_tokens: number }; - }; - expect(response.status).toBe("completed"); - expect(response.output?.[0]?.type).toBe("message"); - expect(response.output?.[0]?.content?.[0]?.text).toBe("FAKE_PLUGIN_OK fake_plugin_tool_17"); - expect(response.usage).toEqual(usage); - expect(agentCommand).toHaveBeenCalledTimes(1); - } finally { - testState.agentsConfig = previousAgentsConfig; - resetConfigRuntimeState(); - } + expect(agentCommandMock).not.toHaveBeenCalled(); }); it.each([false, true])( - "completes a recovered response after an earlier error payload with stream=%s", + "fails a resolved terminal error without exposing provider details with stream=%s", async (stream) => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [ - { text: "Historical failed provider attempt", isError: true }, - { text: "fallback recovered" }, - {}, - ], - } as never); - - const res = await postResponses(enabledPort, { stream, model: "openclaw", input: "hi" }); - const body = await res.text(); - expect(res.status, body).toBe(200); - - if (stream) { - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(0); - expect(events.at(-1)?.data).toBe("[DONE]"); - expect(body).toContain("fallback recovered"); - } else { - const response = JSON.parse(body) as { - status?: string; - output?: Array<{ type?: string; content?: Array<{ text?: string }> }>; - }; - expect(response.status).toBe("completed"); - expect(response.output?.[0]?.type).toBe("message"); - expect(response.output?.[0]?.content?.[0]?.text).toContain("fallback recovered"); - } - - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "completes a recovered media-only response after an earlier error payload with stream=%s", - async (stream) => { - const privateFailure = "Historical private provider failure"; - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [ - { text: privateFailure, isError: true }, - { - mediaUrl: "https://example.invalid/recovered-image.png", - mediaUrls: ["https://example.invalid/recovered-document.pdf"], - }, - ], - } as never); - - const res = await postResponses(enabledPort, { - stream, - model: "openclaw", - input: "recover the generated attachment", - }); - const body = await res.text(); - expect(res.status, body).toBe(200); - - if (stream) { - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - } else { - const response = JSON.parse(body) as { status?: string; error?: unknown }; - expect(response.status).toBe("completed"); - expect(response.error).toBeUndefined(); - } - - expect(body).not.toContain("api_error"); - expect(body).not.toContain(privateFailure); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "preserves a failed response when only transient notices follow with stream=%s", - async (stream) => { - const privateFailure = "Historical private provider failure"; - const notices = [ - { text: "Private commentary notice", isCommentary: true }, - { text: "Private compaction notice", isCompactionNotice: true }, - { text: "Private fallback notice", isFallbackNotice: true }, - { text: "Private reasoning snapshot", isReasoningSnapshot: true }, - { text: "Private status notice", isStatusNotice: true }, - { text: "Private hidden notice", visible: false }, - ]; - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: privateFailure, isError: true }, ...notices], - } as never); - - const res = await postResponses(enabledPort, { - stream, - model: "openclaw", - input: "finish the failed request", - }); - const body = await res.text(); - - if (stream) { - expect(res.status, body).toBe(200); - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - const failedResponse = ( - parseSseData(findSseEvent(events, "response.failed")) as { - response?: { status?: string; error?: { code?: string; message?: string } }; - } - ).response; - expect(failedResponse?.status).toBe("failed"); - expect(failedResponse?.error).toEqual({ code: "api_error", message: "internal error" }); - } else { - expect(res.status, body).toBe(502); - const response = JSON.parse(body) as { - status?: string; - output?: unknown[]; - error?: { code?: string; message?: string }; - }; - expect(response.status).toBe("failed"); - expect(response.output).toEqual([]); - expect(response.error).toEqual({ code: "api_error", message: "internal error" }); - } - - expect(body).not.toContain(privateFailure); - for (const notice of notices) { - expect(body).not.toContain(notice.text); - } - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each([false, true])( - "preserves a failed response when its final payload is whitespace with stream=%s", - async (stream) => { - const privateFailure = "Historical private provider failure"; - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: privateFailure, isError: true }, { text: " \t\n " }], - } as never); - - const res = await postResponses(enabledPort, { stream, model: "openclaw", input: "hi" }); - const body = await res.text(); - - if (stream) { - expect(res.status, body).toBe(200); - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - const failedResponse = ( - parseSseData(findSseEvent(events, "response.failed")) as { - response?: { status?: string; error?: { code?: string; message?: string } }; - } - ).response; - expect(failedResponse?.status).toBe("failed"); - expect(failedResponse?.error).toEqual({ code: "api_error", message: "internal error" }); - } else { - expect(res.status, body).toBe(502); - const response = JSON.parse(body) as { - status?: string; - output?: unknown[]; - error?: { code?: string; message?: string }; - }; - expect(response.status).toBe("failed"); - expect(response.output).toEqual([]); - expect(response.error).toEqual({ code: "api_error", message: "internal error" }); - } - - expect(body).not.toContain(privateFailure); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each(PRESERVED_STREAM_FAILURE_CASES)( - "fails non-stream responses for $name without exposing provider details", - async ({ text, error }) => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text, isError: true }], + const privatePayload = "private terminal payload"; + const privateError = "private provider detail"; + agentCommandMock.mockClear(); + agentCommandMock.mockResolvedValueOnce({ + payloads: [{ text: privatePayload, isError: true }], meta: { - error, + error: { kind: "incomplete_turn", message: privateError }, agentMeta: { usage: { input: 7, output: 3, total: 10 } }, }, } as never); - const res = await postResponses(enabledPort, { model: "openclaw", input: "hi" }); - expect(res.status).toBe(502); + const res = await postResponses(enabledPort, { + model: "openclaw", + input: "hi", + stream, + }); const body = await res.text(); - const response = JSON.parse(body) as { - status?: string; - output?: unknown[]; - error?: { code?: string; message?: string }; - usage?: { input_tokens: number; output_tokens: number; total_tokens: number }; - }; - expect(response.status).toBe("failed"); - expect(response.output).toEqual([]); - expect(response.error).toEqual({ code: "api_error", message: "internal error" }); - expect(response.usage).toEqual(expectedResponsesUsage(7, 3, 10)); - expect(body).not.toContain(text); - expect(body).not.toContain(error.message); - expect(agentCommand).toHaveBeenCalledTimes(1); - }, - ); - - it.each( - STREAM_FAILURE_CASES.flatMap((failure) => - [false, true].map((emitErrorLifecycle) => ({ - name: failure.name, - createError: failure.createError, - tools: failure.tools, - expectedCode: failure.expectedCode, - expectedMessage: failure.expectedMessage, - emitErrorLifecycle, - label: `${failure.name} ${emitErrorLifecycle ? "after" : "without"} an error lifecycle`, - })), - ), - )( - "closes the response stream for $label without reporting completion", - async ({ createError, emitErrorLifecycle, expectedCode, expectedMessage, tools }) => { - const idleRootCount = getActiveGatewayRootWorkCount(); - agentCommandMock.mockClear(); - agentCommandMock.mockImplementationOnce((async (opts: unknown) => { - if (emitErrorLifecycle) { - const runId = (opts as { runId?: string }).runId; - if (!runId) { - throw new Error("expected a streaming response run ID"); - } - emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "error" } }); - } - throw createError(); - }) as never); - - const res = await postResponses( - enabledPort, - { - stream: true, - model: "openclaw", - input: "hi", - tools, - }, - undefined, - AbortSignal.timeout(5_000), - ); - expect(res.status).toBe(200); - - const events = parseSseEvents(await res.text()); - const failedEvents = events.filter((event) => event.event === "response.failed"); - expect(failedEvents).toHaveLength(1); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - - const failedResponse = ( - parseSseData(findSseEvent(events, "response.failed")) as { + if (stream) { + expect(res.status, body).toBe(200); + const events = parseSseEvents(body); + const failed = parseSseData(findSseEvent(events, "response.failed")) as { response?: { status?: string; error?: { code?: string; message?: string } }; - } - ).response; - expect(failedResponse?.status).toBe("failed"); - expect(failedResponse?.error?.code).toBe(expectedCode); - expect(failedResponse?.error?.message).toContain(expectedMessage); - expect(agentCommandMock).toHaveBeenCalledTimes(1); - await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(idleRootCount)); - }, - ); - - it.each( - PRESERVED_STREAM_FAILURE_CASES.flatMap((failure) => - PRESERVED_STREAM_LIFECYCLE_CASES.map((lifecycleCase) => ({ - name: failure.name, - text: failure.text, - lifecycle: failure.lifecycle, - error: failure.error, - lifecycleLabel: lifecycleCase.label, - emitError: lifecycleCase.emitError, - emitEnd: lifecycleCase.emitEnd, - })), - ), - )( - "fails the response stream when $name resolves $lifecycleLabel", - async ({ text, lifecycle, error, emitError, emitEnd }) => { - const idleRootCount = getActiveGatewayRootWorkCount(); - 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 (emitError) { - emitAgentEvent({ - runId, - stream: "lifecycle", - data: { phase: "error", error: text, ...lifecycle }, - }); - } - if (emitEnd) { - emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end" } }); - } - return { - payloads: [{ text, isError: true }], - meta: { - stopReason: "end_turn", - error, - agentMeta: { usage: { input: 7, output: 3, total: 10 } }, - }, }; - }) as never); - - const res = await postResponses( - enabledPort, - { stream: true, model: "openclaw", input: "hi" }, - undefined, - AbortSignal.timeout(5_000), - ); - expect(res.status).toBe(200); - - const body = await res.text(); - const events = parseSseEvents(body); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - - const failedResponse = ( - parseSseData(findSseEvent(events, "response.failed")) as { - response?: { - status?: string; - error?: { code?: string; message?: string }; - usage?: { input_tokens: number; output_tokens: number; total_tokens: number }; - }; - } - ).response; - expect(failedResponse?.status).toBe("failed"); - expect(failedResponse?.error).toEqual({ code: "api_error", message: "internal error" }); - expect(failedResponse?.usage).toEqual(expectedResponsesUsage(7, 3, 10)); - expect(body).not.toContain(error.message); - expect(body).not.toContain(text); - expect(agentCommand).toHaveBeenCalledTimes(1); - await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(idleRootCount)); + expect(failed.response?.status).toBe("failed"); + expect(failed.response?.error).toEqual({ + code: "api_error", + message: "internal error", + }); + expect(events.filter((event) => event.event === "response.completed")).toHaveLength(0); + expect(events.at(-1)?.data).toBe("[DONE]"); + } else { + expect(res.status, body).toBe(502); + const response = JSON.parse(body) as { + status?: string; + output?: unknown[]; + error?: { code?: string; message?: string }; + }; + expect(response.status).toBe("failed"); + expect(response.output).toEqual([]); + expect(response.error).toEqual({ code: "api_error", message: "internal error" }); + } + expect(body).not.toContain(privatePayload); + expect(body).not.toContain(privateError); + expect(agentCommandMock).toHaveBeenCalledTimes(1); }, ); - it("completes a response when a failed attempt recovers through model fallback", async () => { - agentCommand.mockClear(); - agentCommand.mockImplementationOnce((async (opts: unknown) => { + it("completes a stream when a failed attempt is superseded by fallback success", async () => { + agentCommandMock.mockClear(); + agentCommandMock.mockImplementationOnce((async (opts: unknown) => { const runId = (opts as { runId?: string }).runId; if (!runId) { throw new Error("expected a streaming response run ID"); @@ -2657,7 +2047,6 @@ describe("OpenResponses HTTP API (e2e)", () => { expect(body).toContain("fallback recovered"); expect(body).not.toContain("raw primary provider failure"); }); - it("preserves declared owner identity for streaming and non-streaming private callers", async () => { const port = enabledPort; for (const stream of [false, true]) { @@ -3625,150 +3014,52 @@ describe("OpenResponses HTTP API (e2e)", () => { await ensureResponseConsumed(secondResponse); }); - it("does not reuse another user's previous response under the same auth subject", async () => { - const port = enabledPort; - agentCommandMock.mockClear(); - agentCommandMock.mockResolvedValueOnce({ - payloads: [{ text: "First turn." }], - } as never); - - const firstResponse = await postResponses(port, { - stream: false, - model: "openclaw", - user: "alice", - input: "hello", - }); - expect(firstResponse.status).toBe(200); - const firstJson = (await firstResponse.json()) as { id?: string }; - const firstOpts = firstAgentOpts() as { sessionKey?: string } | undefined; - expect(firstOpts?.sessionKey ?? "").toContain("openresponses-user:alice"); - - agentCommandMock.mockResolvedValueOnce({ - payloads: [{ text: "Second turn." }], - } as never); - - const secondResponse = await postResponses(port, { - stream: false, - model: "openclaw", - user: "bob", - previous_response_id: firstJson.id, - input: "hello again", - }); - expect(secondResponse.status).toBe(200); - const secondOpts = firstAgentOpts(1) as { sessionKey?: string } | undefined; - expect(secondOpts?.sessionKey).not.toBe(firstOpts?.sessionKey); - expect(secondOpts?.sessionKey ?? "").toContain("openresponses-user:bob"); - await ensureResponseConsumed(secondResponse); - }); - - it.each( - [false, true].flatMap((stream) => - [false, true].map((namedUser) => ({ - stream, - namedUser, - label: `${namedUser ? "named" : "anonymous"} response with stream=${stream}`, - })), - ), - )( - "isolates a previous $label when the optional SDK user is omitted", - async ({ stream, namedUser }) => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: "First private turn." }], - } as never); - - const firstResponse = await postResponses(enabledPort, { - stream: false, - model: "openclaw", - ...(namedUser ? { user: " alice " } : {}), - input: "private first turn", - }); - const firstBody = await firstResponse.text(); - expect(firstResponse.status, firstBody).toBe(200); - const firstJson = JSON.parse(firstBody) as { id?: string }; - if (!firstJson.id) { - throw new Error("expected a previous response ID"); - } - const firstSessionKey = requireSessionKey( - (firstAgentOpts() as { sessionKey?: string }).sessionKey, - "first response", - ); - if (namedUser) { - expect(firstSessionKey).toContain("openresponses-user:alice"); - } - - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: "Optional-user continuation recovered." }], - } as never); - const secondRequest = { - model: "openclaw", - stream, - previous_response_id: firstJson.id, - input: "continue without optional user metadata", - } satisfies OpenAI.Responses.ResponseCreateParams; - const secondResponse = await postResponses(enabledPort, secondRequest); - const secondBody = await secondResponse.text(); - expect(secondResponse.status, secondBody).toBe(200); - const continuedSessionKey = requireSessionKey( - (firstAgentOpts(1) as { sessionKey?: string }).sessionKey, - "optional-user continuation", - ); - if (namedUser) { - expect(continuedSessionKey).not.toBe(firstSessionKey); - expect(continuedSessionKey).not.toContain("openresponses-user:alice"); - } else { - expect(continuedSessionKey).toBe(firstSessionKey); - } - - if (stream) { - const events = parseSseEvents(secondBody); - expect(events.filter((event) => event.event === "response.completed")).toHaveLength(1); - expect(events.filter((event) => event.event === "response.failed")).toHaveLength(0); - expect(events.filter((event) => event.data === "[DONE]")).toHaveLength(1); - expect(events.at(-1)?.data).toBe("[DONE]"); - } else { - const secondJson = JSON.parse(secondBody) as { status?: string }; - expect(secondJson.status).toBe("completed"); - } - expect(secondBody).toContain("Optional-user continuation recovered."); - expect(agentCommand).toHaveBeenCalledTimes(2); + it.each([ + { + name: "same normalized user", + firstUser: " alice ", + secondUser: "alice", + reuses: true, }, - ); - - it("reuses a previous response for the same normalized user", async () => { - agentCommand.mockClear(); - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: "First private turn." }], - } as never); - + { name: "different user", firstUser: "alice", secondUser: "bob", reuses: false }, + { name: "omitted named user", firstUser: "alice", secondUser: undefined, reuses: false }, + { name: "anonymous user", firstUser: undefined, secondUser: undefined, reuses: true }, + ])("scopes previous responses by $name", async ({ firstUser, secondUser, reuses }) => { + agentCommandMock.mockClear(); + agentCommandMock.mockResolvedValueOnce({ payloads: [{ text: "First turn." }] } as never); const firstResponse = await postResponses(enabledPort, { - stream: false, model: "openclaw", - user: " alice ", - input: "private first turn", + ...(firstUser === undefined ? {} : { user: firstUser }), + input: "first turn", }); expect(firstResponse.status).toBe(200); const firstJson = (await firstResponse.json()) as { id?: string }; - const firstOpts = firstAgentOpts() as { sessionKey?: string } | undefined; - expect(firstOpts?.sessionKey ?? "").toContain("openresponses-user:alice"); - - agentCommand.mockResolvedValueOnce({ - payloads: [{ text: "Second private turn." }], - } as never); + if (!firstJson.id) { + throw new Error("expected a previous response ID"); + } + const firstSessionKey = requireSessionKey( + (firstAgentOpts() as { sessionKey?: string }).sessionKey, + "first response", + ); + agentCommandMock.mockResolvedValueOnce({ payloads: [{ text: "Second turn." }] } as never); const secondResponse = await postResponses(enabledPort, { - stream: false, model: "openclaw", - user: "alice", + ...(secondUser === undefined ? {} : { user: secondUser }), previous_response_id: firstJson.id, - input: "private follow-up", + input: "second turn", }); expect(secondResponse.status).toBe(200); - const secondOpts = firstAgentOpts(1) as { sessionKey?: string } | undefined; - expect(secondOpts?.sessionKey).toBe(firstOpts?.sessionKey); + const secondSessionKey = requireSessionKey( + (firstAgentOpts(1) as { sessionKey?: string }).sessionKey, + "second response", + ); + expect(secondSessionKey === firstSessionKey).toBe(reuses); + if (secondUser) { + expect(secondSessionKey).toContain(`openresponses-user:${secondUser}`); + } await ensureResponseConsumed(secondResponse); }); - it("stores response session mappings when the response is emitted", async () => { const port = enabledPort; agentCommandMock.mockClear(); @@ -3814,60 +3105,6 @@ describe("OpenResponses HTTP API (e2e)", () => { expect(openResponsesTesting.lookupResponseSessionAt("resp_504", 505)).toBe("session_504"); }); - it("keeps cached previous responses scoped to their normalized user", () => { - openResponsesTesting.storeResponseSessionAt("resp_alice", "session_alice", 100, { - authSubject: "subject:a", - agentId: "main", - user: " alice ", - }); - - expect( - openResponsesTesting.lookupResponseSessionAt("resp_alice", 101, { - authSubject: "subject:a", - agentId: "main", - user: "alice", - }), - ).toBe("session_alice"); - expect( - openResponsesTesting.lookupResponseSessionAt("resp_alice", 101, { - authSubject: "subject:a", - agentId: "main", - user: "bob", - }), - ).toBeUndefined(); - expect( - openResponsesTesting.lookupResponseSessionAt("resp_alice", 101, { - authSubject: "subject:a", - agentId: "main", - }), - ).toBeUndefined(); - - openResponsesTesting.storeResponseSessionAt("resp_anonymous", "session_anonymous", 102, { - authSubject: "subject:a", - agentId: "main", - }); - expect( - openResponsesTesting.lookupResponseSessionAt("resp_anonymous", 103, { - authSubject: "subject:a", - agentId: "main", - }), - ).toBe("session_anonymous"); - expect( - openResponsesTesting.lookupResponseSessionAt("resp_anonymous", 103, { - authSubject: "subject:a", - agentId: "main", - user: " ", - }), - ).toBe("session_anonymous"); - expect( - openResponsesTesting.lookupResponseSessionAt("resp_anonymous", 103, { - authSubject: "subject:a", - agentId: "main", - user: "alice", - }), - ).toBeUndefined(); - }); - it("does not reuse cached sessions when the auth subject changes", () => { openResponsesTesting.storeResponseSessionAt("resp_1", "session_1", 100, { authSubject: "subject:a", diff --git a/src/gateway/openresponses-http.ts b/src/gateway/openresponses-http.ts index 61053fe5aa86..b200b3c50804 100644 --- a/src/gateway/openresponses-http.ts +++ b/src/gateway/openresponses-http.ts @@ -84,7 +84,10 @@ import { } from "./open-responses.schema.js"; import { resolveAgentRunUsage } from "./openai-agent-run-usage.js"; import { resolveOpenAiCompatError } from "./openai-compat-errors.js"; -import { resolveOpenAiHttpAgentRunTerminalOutcome } from "./openai-http-terminal-outcome.js"; +import { + resolveOpenAiHttpAgentRunTerminalOutcome, + resolveOpenAiHttpResultText, +} from "./openai-http-terminal-outcome.js"; import { isToolChoiceConstraintSatisfied, resolveUnsatisfiedToolChoiceMessage, @@ -750,11 +753,6 @@ export async function handleOpenResponsesHttpRequest( return true; } - const payloads = ( - result as { - payloads?: Array<{ isError?: boolean; text?: string }>; - } | null - )?.payloads; const meta = (result as { meta?: unknown } | null)?.meta; const { stopReason, pendingToolCalls } = resolveStopReasonAndPendingToolCalls(meta); @@ -787,14 +785,7 @@ export async function handleOpenResponsesHttpRequest( // pending call was emitted, so multi-tool turns lost every call but // the leading one. if (stopReason === "tool_calls" && pendingToolCalls && pendingToolCalls.length > 0) { - const assistantText = - Array.isArray(payloads) && payloads.length > 0 - ? payloads - .filter((replyPayload) => replyPayload.isError !== true) - .map((p) => (typeof p.text === "string" ? p.text : "")) - .filter(Boolean) - .join("\n\n") - : ""; + const assistantText = resolveOpenAiHttpResultText(result); const output: OutputItem[] = []; if (assistantText) { @@ -833,14 +824,7 @@ export async function handleOpenResponsesHttpRequest( return true; } - const content = - Array.isArray(payloads) && payloads.length > 0 - ? payloads - .filter((replyPayload) => replyPayload.isError !== true) - .map((p) => (typeof p.text === "string" ? p.text : "")) - .filter(Boolean) - .join("\n\n") - : "No response from OpenClaw."; + const content = resolveOpenAiHttpResultText(result) || "No response from OpenClaw."; const response = createResponseResource({ id: responseId, @@ -921,11 +905,8 @@ export async function handleOpenResponsesHttpRequest( let unsubscribe = () => {}; let stopWatchingDisconnect = () => {}; let finalUsage: Usage | undefined; - type StreamFinalization = - | { status: "completed"; text: string; outcome?: AgentRunTerminalOutcome } - | { status: "failed"; outcome: AgentRunTerminalOutcome }; - let finalizeRequested: StreamFinalization | null = null; - const readFinalization = (): StreamFinalization | null => finalizeRequested; + let finalizeRequested: { status: "completed" | "failed"; text: string } | null = null; + let terminalOutcome: AgentRunTerminalOutcome | undefined; let finalizeScheduled = false; let terminalLifecyclePhase: "end" | "error" = "end"; let terminalStreamError: string | undefined; @@ -960,15 +941,14 @@ export async function handleOpenResponsesHttpRequest( finalizeUnrepresentableAssistantReplacement(); return; } - const completedFinalization = finalizeRequested; - if (completedFinalization.status !== "completed") { + if (finalizeRequested.status !== "completed") { finalizeScheduled = false; maybeFinalize(); return; } const usage = finalUsage; const finalText = - accumulatedText || bufferedReplaceableAssistantContent || completedFinalization.text; + accumulatedText || bufferedReplaceableAssistantContent || finalizeRequested.text; closed = true; stopWatchingDisconnect(); @@ -1018,9 +998,9 @@ export async function handleOpenResponsesHttpRequest( }); }; - const requestFinalize = (terminal: StreamFinalization) => { + const requestFinalize = (status: "completed" | "failed", text = "") => { // Attempt errors stay provisional while a successful fallback can recover. - finalizeRequested = terminal; + finalizeRequested = { status, text }; maybeFinalize(); }; @@ -1176,18 +1156,15 @@ export async function handleOpenResponsesHttpRequest( phase, data: evt.data, }); - const outcome = mergeAgentRunTerminalOutcome(finalizeRequested?.outcome, incomingOutcome); + const outcome = mergeAgentRunTerminalOutcome(terminalOutcome, incomingOutcome); + terminalOutcome = outcome; if (outcome.reason !== "completed") { - requestFinalize({ status: "failed", outcome }); + requestFinalize("failed"); } else { - requestFinalize({ - status: "completed", - text: - accumulatedText || - bufferedReplaceableAssistantContent || - "No response from OpenClaw.", - outcome, - }); + requestFinalize( + "completed", + accumulatedText || bufferedReplaceableAssistantContent || "No response from OpenClaw.", + ); } } } @@ -1232,15 +1209,14 @@ export async function handleOpenResponsesHttpRequest( return; } finalUsage = extractUsageFromResult(result); - const resultOutcome = resolveOpenAiHttpAgentRunTerminalOutcome(result); - if (resultOutcome.reason !== "completed") { - requestFinalize({ - status: "failed", - outcome: mergeAgentRunTerminalOutcome(readFinalization()?.outcome, resultOutcome), - }); + const priorFinalization = finalizeRequested; + const outcome = resolveOpenAiHttpAgentRunTerminalOutcome(result, terminalOutcome); + terminalOutcome = outcome; + if (outcome.reason !== "completed") { + requestFinalize("failed"); return; } - if (readFinalization()?.status === "failed" && terminalStreamError) { + if (priorFinalization?.status === "failed" && terminalStreamError) { const failedResponse = createResponseResource({ id: responseId, model, @@ -1253,11 +1229,6 @@ export async function handleOpenResponsesHttpRequest( finalizeFailedResponse(failedResponse); return; } - const outcome = resolveOpenAiHttpAgentRunTerminalOutcome(result, readFinalization()?.outcome); - if (outcome.reason !== "completed") { - requestFinalize({ status: "failed", outcome }); - return; - } if (unrepresentableAssistantReplacement) { finalizeUnrepresentableAssistantReplacement(); @@ -1266,17 +1237,8 @@ export async function handleOpenResponsesHttpRequest( // Check for pending client tool calls BEFORE maybeFinalize() because the // lifecycle:end event may already have requested finalization. - const resultAny = result as { - payloads?: Array<{ isError?: boolean; text?: string }>; - meta?: unknown; - }; - const resultPayloadText = Array.isArray(resultAny.payloads) - ? resultAny.payloads - .filter((replyPayload) => replyPayload.isError !== true) - .map((p) => (typeof p.text === "string" ? p.text : "")) - .filter(Boolean) - .join("\n\n") - : ""; + const resultAny = result as { meta?: unknown }; + const resultPayloadText = resolveOpenAiHttpResultText(result); const meta = resultAny.meta; const { stopReason, pendingToolCalls } = resolveStopReasonAndPendingToolCalls(meta); @@ -1413,9 +1375,8 @@ export async function handleOpenResponsesHttpRequest( accumulatedText = content; sawAssistantDelta = true; - const finalization = readFinalization(); - if (finalization?.status === "completed") { - finalizeRequested = { ...finalization, text: content }; + if (finalizeRequested?.status === "completed") { + finalizeRequested = { ...finalizeRequested, text: content }; } writeSseEvent(res, {