From 12f72dac8129883265dceca62ac7a7a5cae069df Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 21 Aug 2026 17:28:01 -0700 Subject: [PATCH] fix(gateway): fail streaming responses when agent runs fail (#127662) * fix(gateway): fail streaming responses when agent runs fail * test(gateway): preserve typed streaming failure fixtures --- src/gateway/openai-compat-errors.ts | 7 ++ src/gateway/openai-http.test.ts | 106 +++++++++++++++++++++- src/gateway/openai-http.ts | 14 ++- src/gateway/openresponses-http.test.ts | 121 ++++++++++++++++++++++++- src/gateway/openresponses-http.ts | 21 ++++- 5 files changed, 263 insertions(+), 6 deletions(-) diff --git a/src/gateway/openai-compat-errors.ts b/src/gateway/openai-compat-errors.ts index 455e670a4800..15e3d18e64b3 100644 --- a/src/gateway/openai-compat-errors.ts +++ b/src/gateway/openai-compat-errors.ts @@ -1,3 +1,4 @@ +import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce"; import { describeFailoverError, resolveFailoverStatus } from "../agents/failover-error.js"; // OpenAI-compatible error helpers. // Converts OpenClaw failover/sampling errors to OpenAI-style HTTP responses. @@ -31,6 +32,12 @@ const ERROR_TYPE_BY_REASON = { unknown: undefined, } satisfies Record; +/** Resolved agent failures must not become successful OpenAI HTTP responses. */ +export function isFailedOpenAiAgentRun(result: unknown): boolean { + const metadata = asOptionalRecord(asOptionalRecord(result)?.meta); + return Boolean(metadata?.error) || metadata?.stopReason === "error"; +} + function statusForReason(reason: FailoverReason, status: number | undefined): number { if (reason === "server_error") { return status && status >= 400 && status < 500 ? status : 502; diff --git a/src/gateway/openai-http.test.ts b/src/gateway/openai-http.test.ts index f27e4804027c..64096acd5100 100644 --- a/src/gateway/openai-http.test.ts +++ b/src/gateway/openai-http.test.ts @@ -6,7 +6,12 @@ import path from "node:path"; import OpenAI from "openai"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { createDeferred } from "../../test/helpers/promise.js"; +import { + buildAgentRunTerminalOutcome, + buildAgentRunTerminalOutcomeFromLifecycleEvent, +} from "../agents/agent-run-terminal-outcome.js"; import { createClientToolNameConflictError } from "../agents/agent-tool-definition-adapter.js"; +import { createAgentCommandLifecycle } from "../agents/command/lifecycle.js"; import { createStubSessionHarness, emitAssistantTextDelta, @@ -17,7 +22,11 @@ import { HISTORY_CONTEXT_MARKER } from "../auto-reply/reply/history.js"; import { CURRENT_MESSAGE_MARKER } from "../auto-reply/reply/mentions.js"; import { resetConfigRuntimeState } from "../config/config.js"; import { upsertSessionEntryCore } from "../config/sessions/session-accessor.js"; -import { emitAgentEvent, onAgentEvent } from "../infra/agent-events.js"; +import { + emitAgentEvent, + getAgentEventLifecycleGeneration, + onAgentEvent, +} from "../infra/agent-events.js"; import { enqueueCommandInLane } from "../process/command-queue.js"; import { getActiveGatewayRootWorkCount, @@ -2031,6 +2040,101 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { expect(res.status).toBe(500); }); + it.each( + [ + { + label: "terminal metadata", + meta: { error: { kind: "incomplete_turn" as const, message: "private provider failure" } }, + expectedPhase: "error" as const, + }, + { + label: "an error stop reason", + meta: { stopReason: "error" }, + expectedPhase: "end" as const, + }, + ].flatMap((failure) => + [false, true].map((producerTerminal) => ({ + meta: failure.meta, + expectedPhase: failure.expectedPhase, + producerTerminal, + label: `${failure.label} ${producerTerminal ? "after" : "without"} a producer terminal`, + })), + ), + )( + "rejects resolved streaming agent failures from $label", + async ({ meta, expectedPhase, producerTerminal }) => { + let runId: string | undefined; + const terminals: Array<{ phase: "end" | "error"; status: string }> = []; + const unsubscribe = onAgentEvent((event) => { + if (event.runId === runId && event.stream === "lifecycle") { + const phase = event.data?.phase; + if (phase === "end" || phase === "error") { + terminals.push({ + phase, + status: buildAgentRunTerminalOutcomeFromLifecycleEvent({ phase, data: event.data }) + .status, + }); + } + } + }); + agentCommandMock.mockClear(); + agentCommandMock.mockImplementationOnce((async (options: unknown) => { + runId = (options as { runId?: string }).runId; + if (!runId) { + throw new Error("expected a streaming chat-completion run ID"); + } + const result = { + payloads: [{ text: "Command may have changed state", isError: true }], + meta: { durationMs: 0, ...meta }, + }; + if (producerTerminal) { + const lifecycle = createAgentCommandLifecycle({ + runId, + lifecycleGeneration: getAgentEventLifecycleGeneration, + startedAt: Date.now(), + state: { + currentTurnUserMessagePersisted: true, + lifecycleFinishing: false, + lifecycleEnded: false, + }, + }); + const terminal = { + metadata: {}, + outcome: buildAgentRunTerminalOutcome({ status: "error", stopReason: "error" }), + }; + if (lifecycle.resolveResultError(result, false)) { + lifecycle.emitResultError(result, false, terminal); + } else { + lifecycle.emitEnd(terminal); + } + } + return result; + }) as never); + + try { + const stream = await createOpenAiChatClient(enabledPort).chat.completions.create({ + model: "openclaw", + messages: [{ role: "user", content: "hi" }], + stream: true, + }); + const finishReasons: Array = []; + await expect(async () => { + for await (const chunk of stream) { + finishReasons.push(...chunk.choices.map((choice) => choice.finish_reason)); + } + }).rejects.toMatchObject({ + error: { message: "internal error", type: "api_error" }, + }); + expect(finishReasons).not.toContain("stop"); + expect(terminals).toEqual([ + { phase: producerTerminal ? expectedPhase : "error", status: "error" }, + ]); + } finally { + unsubscribe(); + } + }, + ); + 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 da81e71e925e..1596ae05ab54 100644 --- a/src/gateway/openai-http.ts +++ b/src/gateway/openai-http.ts @@ -66,7 +66,11 @@ import { } from "./http-utils.js"; import { normalizeInputHostnameAllowlist } from "./input-allowlist.js"; import { resolveAgentRunUsage } from "./openai-agent-run-usage.js"; -import { resolveOpenAiCompatError, validateOpenAiSamplingParams } from "./openai-compat-errors.js"; +import { + isFailedOpenAiAgentRun, + resolveOpenAiCompatError, + validateOpenAiSamplingParams, +} from "./openai-compat-errors.js"; import { isToolChoiceConstraintSatisfied, resolveUnsatisfiedToolChoiceMessage, @@ -1098,7 +1102,7 @@ export async function handleOpenAiHttpRequest( } const meta = (result as { meta?: { error?: unknown; stopReason?: unknown } } | null)?.meta; - if (meta?.error || meta?.stopReason === "error") { + if (isFailedOpenAiAgentRun(result)) { throw new Error("agent run failed"); } const usage = resolveChatCompletionUsage(result); @@ -1383,6 +1387,12 @@ export async function handleOpenAiHttpRequest( return; } + if (isFailedOpenAiAgentRun(result)) { + terminalLifecyclePhase = "error"; + finishStreamWithError({ message: "internal error", type: "api_error" }); + return; + } + if (terminalStreamError) { finishStreamWithError(terminalStreamError); return; diff --git a/src/gateway/openresponses-http.test.ts b/src/gateway/openresponses-http.test.ts index b916122af35a..4a52068b4b44 100644 --- a/src/gateway/openresponses-http.test.ts +++ b/src/gateway/openresponses-http.test.ts @@ -6,13 +6,22 @@ import path from "node:path"; import OpenAI from "openai"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { createDeferred } from "../../test/helpers/promise.js"; +import { + buildAgentRunTerminalOutcome, + buildAgentRunTerminalOutcomeFromLifecycleEvent, +} from "../agents/agent-run-terminal-outcome.js"; import { createClientToolNameConflictError } from "../agents/agent-tool-definition-adapter.js"; +import { createAgentCommandLifecycle } from "../agents/command/lifecycle.js"; import { FailoverError } from "../agents/failover-error.js"; import { HISTORY_CONTEXT_MARKER } from "../auto-reply/reply/history.js"; import { CURRENT_MESSAGE_MARKER } from "../auto-reply/reply/mentions.js"; import { resetConfigRuntimeState } from "../config/config.js"; import { upsertSessionEntryCore } from "../config/sessions/session-accessor.js"; -import { emitAgentEvent, onAgentEvent } from "../infra/agent-events.js"; +import { + emitAgentEvent, + getAgentEventLifecycleGeneration, + onAgentEvent, +} from "../infra/agent-events.js"; import { enqueueCommandInLane } from "../process/command-queue.js"; import { getActiveGatewayRootWorkCount, @@ -1832,6 +1841,116 @@ describe("OpenResponses HTTP API (e2e)", () => { expect(res.status).toBe(500); }); + it.each( + [ + { + label: "terminal metadata", + meta: { error: { kind: "incomplete_turn" as const, message: "private provider failure" } }, + expectedPhase: "error" as const, + }, + { + label: "an error stop reason", + meta: { stopReason: "error" }, + expectedPhase: "end" as const, + }, + ].flatMap((failure) => + [false, true].map((producerTerminal) => ({ + meta: failure.meta, + expectedPhase: failure.expectedPhase, + producerTerminal, + label: `${failure.label} ${producerTerminal ? "after" : "without"} a producer terminal`, + })), + ), + )( + "fails resolved streaming agent failures from $label", + async ({ meta, expectedPhase, producerTerminal }) => { + let runId: string | undefined; + const terminals: Array<{ phase: "end" | "error"; status: string }> = []; + const unsubscribe = onAgentEvent((event) => { + if (event.runId === runId && event.stream === "lifecycle") { + const phase = event.data?.phase; + if (phase === "end" || phase === "error") { + terminals.push({ + phase, + status: buildAgentRunTerminalOutcomeFromLifecycleEvent({ phase, data: event.data }) + .status, + }); + } + } + }); + agentCommandMock.mockClear(); + agentCommandMock.mockImplementationOnce((async (options: unknown) => { + runId = (options as { runId?: string }).runId; + if (!runId) { + throw new Error("expected a streaming response run ID"); + } + const result = { + payloads: [{ text: "Command may have changed state", isError: true }], + meta: { + durationMs: 0, + agentMeta: { + sessionId: "failed-stream-session", + provider: "openai", + model: "test-model", + usage: { input: 11, output: 7, total: 18 }, + }, + ...meta, + }, + }; + if (producerTerminal) { + const lifecycle = createAgentCommandLifecycle({ + runId, + lifecycleGeneration: getAgentEventLifecycleGeneration, + startedAt: Date.now(), + state: { + currentTurnUserMessagePersisted: true, + lifecycleFinishing: false, + lifecycleEnded: false, + }, + }); + const terminal = { + metadata: {}, + outcome: buildAgentRunTerminalOutcome({ status: "error", stopReason: "error" }), + }; + if (lifecycle.resolveResultError(result, false)) { + lifecycle.emitResultError(result, false, terminal); + } else { + lifecycle.emitEnd(terminal); + } + } + return result; + }) as never); + + try { + const client = new OpenAI({ + apiKey: "test", + baseURL: `http://127.0.0.1:${enabledPort}/v1`, + defaultHeaders: { "x-openclaw-scopes": "operator.write" }, + maxRetries: 0, + }); + const stream = client.responses.stream({ model: "openclaw", input: "hi" }); + const terminalEvents: string[] = []; + stream.on("response.completed", () => terminalEvents.push("response.completed")); + stream.on("response.failed", () => terminalEvents.push("response.failed")); + + const response = await stream.finalResponse(); + expect(response.status).toBe("failed"); + expect(response.error).toEqual({ code: "api_error", message: "internal error" }); + expect(response.usage).toMatchObject({ + input_tokens: 11, + output_tokens: 7, + total_tokens: 18, + }); + expect(terminalEvents).toEqual(["response.failed"]); + expect(terminals).toEqual([ + { phase: producerTerminal ? expectedPhase : "error", status: "error" }, + ]); + } finally { + unsubscribe(); + } + }, + ); + it.each( STREAM_FAILURE_CASES.flatMap((failure) => [false, true].map((emitErrorLifecycle) => ({ diff --git a/src/gateway/openresponses-http.ts b/src/gateway/openresponses-http.ts index 37b9ebf9fbc0..1d0c1ea60172 100644 --- a/src/gateway/openresponses-http.ts +++ b/src/gateway/openresponses-http.ts @@ -77,7 +77,7 @@ import { type Usage, } from "./open-responses.schema.js"; import { resolveAgentRunUsage } from "./openai-agent-run-usage.js"; -import { resolveOpenAiCompatError } from "./openai-compat-errors.js"; +import { isFailedOpenAiAgentRun, resolveOpenAiCompatError } from "./openai-compat-errors.js"; import { isToolChoiceConstraintSatisfied, resolveUnsatisfiedToolChoiceMessage, @@ -720,7 +720,7 @@ export async function handleOpenResponsesHttpRequest( } const meta = (result as { meta?: { error?: unknown; stopReason?: unknown } } | null)?.meta; - if (meta?.error || meta?.stopReason === "error") { + if (isFailedOpenAiAgentRun(result)) { throw new Error("agent run failed"); } const payloads = (result as { payloads?: Array<{ text?: string }> } | null)?.payloads; @@ -1184,6 +1184,23 @@ export async function handleOpenResponsesHttpRequest( if (closed) { return; } + + if (isFailedOpenAiAgentRun(result)) { + terminalLifecyclePhase = "error"; + rememberResponseSession(); + finalizeFailedResponse( + createResponseResource({ + id: responseId, + model, + status: "failed", + output: [], + error: { code: "api_error", message: "internal error" }, + usage: extractUsageFromResult(result), + }), + ); + return; + } + finalUsage = extractUsageFromResult(result); if (unrepresentableAssistantReplacement) {