fix(gateway): terminate failed chat streams with API errors (#118428)

* fix(gateway): terminate failed chat streams with API errors

* style(gateway): format streaming regression coverage

* fix(gateway): preserve terminal errors when agent execution rejects

src/gateway/openai-http.test.ts

@@ -2166,34 +2166,234 @@ describe("OpenAI-compatible HTTP API (e2e)", () => {
     expect(finishChoice?.finish_reason).toBe("stop");
   });
 
+  it.each([
+    { name: "resolved", reject: false },
+    { name: "rejected", reject: true },
+  ])(
+    "fails an official SDK stream when an error lifecycle precedes a $name run",
+    async ({ reject }) => {
+      const idleRootCount = getActiveGatewayRootWorkCount();
+      const wireResponse = createDeferred<string>();
+      agentCommand.mockClear();
+      agentCommand.mockImplementationOnce((async (opts: unknown) => {
+        const runId = (opts as { runId?: string }).runId;
+        if (!runId) {
+          throw new Error("expected a streaming chat-completion run ID");
+        }
+        emitAgentEvent({ runId, stream: "assistant", data: { delta: "partial answer" } });
+        emitAgentEvent({
+          runId,
+          stream: "lifecycle",
+          data: { phase: "error", error: "All model fallback candidates failed" },
+        });
+        emitAgentEvent({
+          runId,
+          stream: "lifecycle",
+          data: { phase: "error", error: "A later lifecycle event must not replace the failure" },
+        });
+        if (reject) {
+          throw new Error("private upstream failure");
+        }
+        return {
+          payloads: [{ text: "partial answer" }],
+          meta: { agentMeta: { usage: { input: 11, output: 7, total: 18 } } },
+        };
+      }) as never);
+
+      const client = new OpenAI({
+        apiKey: "test",
+        baseURL: `http://127.0.0.1:${enabledPort}/v1`,
+        defaultHeaders: { "x-openclaw-scopes": "operator.write" },
+        maxRetries: 0,
+        fetch: async (input, init) => {
+          const response = await fetch(input, init);
+          void response.clone().text().then(wireResponse.resolve, wireResponse.reject);
+          return response;
+        },
+      });
+      const stream = await client.chat.completions.create({
+        model: "openclaw",
+        messages: [{ role: "user", content: "Report the provider failure." }],
+        stream: true,
+      });
+      const deliveredContent: string[] = [];
+      const deliveredFinishReasons: Array<string | null> = [];
+
+      await expect(async () => {
+        for await (const chunk of stream) {
+          for (const choice of chunk.choices) {
+            if (typeof choice.delta.content === "string") {
+              deliveredContent.push(choice.delta.content);
+            }
+            deliveredFinishReasons.push(choice.finish_reason);
+          }
+        }
+      }).rejects.toMatchObject({
+        message: "All model fallback candidates failed",
+        type: "api_error",
+      });
+
+      const data = parseSseDataLines(await wireResponse.promise);
+      const chunks = data
+        .filter((line) => line !== "[DONE]")
+        .map((line) => JSON.parse(line) as Record<string, unknown>);
+      expect(deliveredContent).toEqual(["partial answer"]);
+      expect(deliveredFinishReasons.every((reason) => reason === null)).toBe(true);
+      expect(chunks.filter((chunk) => "error" in chunk)).toEqual([
+        { error: { message: "All model fallback candidates failed", type: "api_error" } },
+      ]);
+      expect(data.at(-1)).toBe("[DONE]");
+      expect(agentCommand).toHaveBeenCalledTimes(1);
+      await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(idleRootCount));
+    },
+  );
+
+  it.each([
+    { name: "rewritten", replacementText: "final answer" },
+    { name: "shortened", replacementText: "dra" },
+    { name: "cleared", replacementText: "" },
+  ])("fails an official SDK stream when streamed text is $name", async ({ replacementText }) => {
+    agentCommand.mockClear();
+    agentCommand.mockImplementationOnce((async (opts: unknown) => {
+      const runId = (opts as { runId?: string }).runId;
+      if (!runId) {
+        throw new Error("expected a streaming chat-completion run ID");
+      }
+      emitAgentEvent({
+        runId,
+        stream: "assistant",
+        data: { text: "draft answer", delta: "draft answer" },
+      });
+      emitAgentEvent({
+        runId,
+        stream: "assistant",
+        data: { text: replacementText, delta: "", replace: true, phase: "commentary" },
+      });
+      emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end" } });
+      return { payloads: [{ text: replacementText }] };
+    }) as never);
+
+    const stream = await createOpenAiChatClient(enabledPort).chat.completions.create({
+      model: "openclaw",
+      messages: [{ role: "user", content: "Reject an incompatible replacement snapshot." }],
+      stream: true,
+    });
+    const deliveredContent: string[] = [];
+    await expect(async () => {
+      for await (const chunk of stream) {
+        for (const choice of chunk.choices) {
+          if (typeof choice.delta.content === "string") {
+            deliveredContent.push(choice.delta.content);
+          }
+        }
+      }
+    }).rejects.toMatchObject({
+      message: "Assistant output cannot be represented as an append-only response stream.",
+      type: "api_error",
+    });
+    expect(deliveredContent).toEqual(["draft answer"]);
+    expect(agentCommand).toHaveBeenCalledTimes(1);
+  });
+
+  it.each([
+    {
+      name: "a producer replacement snapshot without a delta",
+      previousDelta: undefined,
+      replacementDelta: undefined,
+    },
+    {
+      name: "a producer replacement snapshot with its own delta",
+      previousDelta: undefined,
+      replacementDelta: "final answer",
+    },
+    {
+      name: "an append-compatible replacement after streamed partial text",
+      previousDelta: "final ",
+      replacementDelta: "answer",
+    },
+  ])(
+    "keeps official SDK text consistent for $name",
+    async ({ previousDelta, replacementDelta }) => {
+      agentCommand.mockClear();
+      agentCommand.mockImplementationOnce((async (opts: unknown) => {
+        const runId = (opts as { runId?: string }).runId;
+        if (!runId) {
+          throw new Error("expected a streaming chat-completion run ID");
+        }
+        if (previousDelta) {
+          emitAgentEvent({
+            runId,
+            stream: "assistant",
+            data: { text: previousDelta, delta: previousDelta },
+          });
+        }
+        emitAgentEvent({
+          runId,
+          stream: "assistant",
+          data: {
+            text: "final answer",
+            replace: true,
+            phase: "commentary",
+            ...(replacementDelta === undefined ? {} : { delta: replacementDelta }),
+          },
+        });
+        emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end" } });
+        return { payloads: [{ text: "final answer" }] };
+      }) as never);
+
+      const stream = await createOpenAiChatClient(enabledPort).chat.completions.create({
+        model: "openclaw",
+        messages: [{ role: "user", content: "Preserve an append-compatible replacement." }],
+        stream: true,
+      });
+      const deliveredContent: string[] = [];
+      const finishReasons: Array<string | null> = [];
+      for await (const chunk of stream) {
+        for (const choice of chunk.choices) {
+          if (typeof choice.delta.content === "string") {
+            deliveredContent.push(choice.delta.content);
+          }
+          finishReasons.push(choice.finish_reason);
+        }
+      }
+      expect(deliveredContent.join("")).toBe("final answer");
+      expect(finishReasons.at(-1)).toBe("stop");
+      expect(agentCommand).toHaveBeenCalledTimes(1);
+    },
+  );
+
   it.each([
     {
       name: "successful completion without a provider terminal",
       fail: false,
       providerTerminal: false,
       expected: "hello",
+      protocolError: false,
     },
     {
       name: "successful completion with a provider terminal",
       fail: false,
       providerTerminal: true,
       expected: "hello",
+      protocolError: false,
     },
     {
       name: "internal agent error without a provider terminal",
       fail: true,
       providerTerminal: false,
       expected: "Error: internal error",
+      protocolError: false,
     },
     {
       name: "internal agent error with a provider terminal",
       fail: true,
       providerTerminal: true,
-      expected: "Error: internal error",
+      expected: "Agent run failed",
+      protocolError: true,
     },
   ])(
     "separates streamed content from the terminal finish for an official SDK $name",
-    async ({ fail, providerTerminal, expected }) => {
+    async ({ fail, providerTerminal, expected, protocolError }) => {
       const idleRootCount = getActiveGatewayRootWorkCount();
       const terminalAdmission = createDeferred<{
         active: number;
@@ -2271,8 +2471,18 @@ describe("OpenAI-compatible HTTP API (e2e)", () => {
           delta: { content?: string | null };
           finish_reason: string | null;
         }> = [];
-        for await (const chunk of stream) {
-          choices.push(...chunk.choices);
+        const consumeStream = async () => {
+          for await (const chunk of stream) {
+            choices.push(...chunk.choices);
+          }
+        };
+        if (protocolError) {
+          await expect(consumeStream()).rejects.toMatchObject({
+            message: expected,
+            type: "api_error",
+          });
+        } else {
+          await consumeStream();
         }
 
         const [admission, wire] = await Promise.all([
@@ -2285,13 +2495,17 @@ describe("OpenAI-compatible HTTP API (e2e)", () => {
         expect(lifecycleTerminals).toEqual([fail ? "error" : "end"]);
 
         const contentChoices = choices.filter((choice) => typeof choice.delta.content === "string");
-        expect(contentChoices.map((choice) => choice.delta.content).join("")).toBe(expected);
+        expect(contentChoices.map((choice) => choice.delta.content).join("")).toBe(
+          protocolError ? "" : expected,
+        );
         expect(contentChoices.every((choice) => choice.finish_reason === null)).toBe(true);
 
         const terminalChoices = choices.filter((choice) => choice.finish_reason === "stop");
-        expect(terminalChoices).toHaveLength(1);
-        expect(terminalChoices[0]?.delta).toEqual({});
-        expect(choices.at(-1)).toEqual(terminalChoices[0]);
+        expect(terminalChoices).toHaveLength(protocolError ? 0 : 1);
+        if (!protocolError) {
+          expect(terminalChoices[0]?.delta).toEqual({});
+          expect(choices.at(-1)).toEqual(terminalChoices[0]);
+        }
         await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(idleRootCount));
       } finally {
         continueAgent.resolve();

src/gateway/openai-http.ts

@@ -1191,6 +1191,7 @@ export async function handleOpenAiHttpRequest(
   let resultResolved = false;
   let closed = false;
   let observedTerminalLifecycle = false;
+  let terminalStreamError: { message: string; type: string; code?: string } | undefined;
   let terminalLifecyclePhase: "end" | "error" = "end";
   let stopWatchingDisconnect = () => {};
 
@@ -1211,6 +1212,10 @@ export async function handleOpenAiHttpRequest(
       if (closed) {
         return;
       }
+      if (terminalStreamError) {
+        finishStreamWithError(terminalStreamError);
+        return;
+      }
       closed = true;
       stopWatchingDisconnect();
       unsubscribe();
@@ -1257,6 +1262,20 @@ export async function handleOpenAiHttpRequest(
         return;
       }
 
+      // SSE deltas cannot retract bytes already delivered to the OpenAI client.
+      if (
+        replace &&
+        typeof text === "string" &&
+        !toolChoiceConstraint &&
+        !text.startsWith(streamedAssistantText)
+      ) {
+        terminalStreamError ??= {
+          message: "Assistant output cannot be represented as an append-only response stream.",
+          type: "api_error",
+        };
+        return;
+      }
+
       // Snapshots include prefixes held during tag-boundary filtering; the raw
       // delta alone can omit a literal leading less-than.
       const content =
@@ -1297,11 +1316,29 @@ export async function handleOpenAiHttpRequest(
       }
       if (phase === "end" || phase === "error") {
         observedTerminalLifecycle = true;
+        if (phase === "error" && terminalLifecyclePhase !== "error") {
+          terminalStreamError ??= {
+            message: normalizeOptionalString(evt.data?.error) ?? "Agent run failed",
+            type: "api_error",
+          };
+        }
         requestFinalize();
       }
     }
   });
 
+  const finishStreamWithError = (error: { message: string; type: string; code?: string }) => {
+    if (closed) {
+      return;
+    }
+    closed = true;
+    stopWatchingDisconnect();
+    unsubscribe();
+    writeSse(res, { error });
+    writeDone(res);
+    res.end();
+  };
+
   // Agent cleanup and deferred SSE delivery have independent lifetimes;
   // shutdown must wait until both have settled, whichever finishes last.
   const releaseAgentRootWork = retainGatewayRootWorkAdmissionContinuation();
@@ -1332,6 +1369,11 @@ export async function handleOpenAiHttpRequest(
         return;
       }
 
+      if (terminalStreamError) {
+        finishStreamWithError(terminalStreamError);
+        return;
+      }
+
       finalUsage = resolveChatCompletionUsage(result);
       const meta = (result as { meta?: unknown } | null)?.meta;
       const { stopReason, pendingToolCalls } = resolveStopReasonAndPendingToolCalls(meta);
@@ -1346,17 +1388,10 @@ export async function handleOpenAiHttpRequest(
           pendingToolCalls,
         })
       ) {
-        closed = true;
-        stopWatchingDisconnect();
-        unsubscribe();
-        writeSse(res, {
-          error: {
-            message: resolveUnsatisfiedToolChoiceMessage(toolChoiceConstraint),
-            type: "api_error",
-          },
+        finishStreamWithError({
+          message: resolveUnsatisfiedToolChoiceMessage(toolChoiceConstraint),
+          type: "api_error",
         });
-        writeDone(res);
-        res.end();
         return;
       }
 
@@ -1416,26 +1451,22 @@ export async function handleOpenAiHttpRequest(
       terminalLifecyclePhase = "error";
       logWarn(`openai-compat: streaming chat completion failed: ${String(err)}`);
       if (isClientToolNameConflictError(err)) {
-        closed = true;
-        stopWatchingDisconnect();
-        unsubscribe();
-        writeSse(res, {
-          error: { message: "invalid tool configuration", type: "invalid_request_error" },
+        finishStreamWithError({
+          message: "invalid tool configuration",
+          type: "invalid_request_error",
         });
-        writeDone(res);
-        res.end();
         return;
       }
       const mapped = resolveOpenAiCompatError(err);
       if (mapped) {
-        closed = true;
-        stopWatchingDisconnect();
-        unsubscribe();
-        writeSse(res, { error: mapped.error });
-        writeDone(res);
-        res.end();
+        finishStreamWithError(mapped.error);
         return;
       }
+      if (terminalStreamError) {
+        finishStreamWithError(terminalStreamError);
+        return;
+      }
+      // Runs without a producer-owned terminal retain the visible-error fallback.
       const content = "Error: internal error";
       writeAssistantContentChunk(res, {
         runId,