fix(agents): defer Anthropic transport stream start event until after message_start

openperf · openperf · commit 399a243c6437 · 2026-06-05T23:31:24.000+08:00
Applies the same start-event deferral fix from src/llm/providers/anthropic.ts to the
embedded-agent default path. resolveEmbeddedAgentStreamFn routes anthropic-messages
through createBoundaryAwareStreamFnForModel → createAnthropicMessagesTransportStreamFn,
so the thinking-block recovery bug (pumpStreamWithRecovery yieldedOutput gate) affects
the production embedded path via this file, not just the provider stream.

Moves stream.push({type:'start'}) from before the SDK event loop into the message_start
handler, keeping yieldedOutput=false in pumpStreamWithRecovery when an SSE event: error
arrives before message_start (as Anthropic sends for invalid thinking signatures).
diff --git a/src/agents/anthropic-transport-stream.test.ts b/src/agents/anthropic-transport-stream.test.ts
@@ -2034,4 +2034,64 @@ describe("anthropic transport stream", () => {
     expect(payload.thinking).toEqual({ type: "adaptive" });
     expect(payload.output_config).toEqual({ effort: "high" });
   });
+
+  it("emits start event only after message_start so pre-stream SSE errors arrive before any non-error event", async () => {
+    guardedFetchMock.mockResolvedValueOnce(
+      createSseResponse([
+        {
+          type: "message_start",
+          message: { id: "msg_1", usage: { input_tokens: 1, output_tokens: 0 } },
+        },
+        {
+          type: "message_delta",
+          delta: { stop_reason: "end_turn" },
+          usage: { input_tokens: 1, output_tokens: 1 },
+        },
+      ]),
+    );
+    const streamFn = createAnthropicMessagesTransportStreamFn();
+    const stream = streamFn(
+      makeAnthropicTransportModel(),
+      { messages: [{ role: "user", content: "hi" }] } as AnthropicStreamContext,
+      { apiKey: "sk-ant-api" } as AnthropicStreamOptions,
+    );
+
+    const eventTypes: string[] = [];
+    for await (const event of stream as AsyncIterable<{ type: string }>) {
+      eventTypes.push(event.type);
+    }
+
+    const startIndex = eventTypes.indexOf("start");
+    expect(startIndex).toBeGreaterThanOrEqual(0);
+    expect(eventTypes.slice(0, startIndex).some((t) => t === "error")).toBe(false);
+  });
+
+  it("emits error without a preceding start event when SSE error arrives before message_start", async () => {
+    guardedFetchMock.mockResolvedValueOnce(
+      createRawSseResponse(
+        "event: error\ndata: " +
+          JSON.stringify({
+            type: "invalid_request_error",
+            message: "messages.1.content.63: Invalid signature in thinking block",
+          }) +
+          "\n\n",
+      ),
+    );
+    const streamFn = createAnthropicMessagesTransportStreamFn();
+    const stream = streamFn(
+      makeAnthropicTransportModel(),
+      { messages: [{ role: "user", content: "hi" }] } as AnthropicStreamContext,
+      { apiKey: "sk-ant-api" } as AnthropicStreamOptions,
+    );
+
+    const eventTypes: string[] = [];
+    for await (const event of stream as AsyncIterable<{ type: string }>) {
+      eventTypes.push(event.type);
+    }
+
+    // start must not precede the error path, regardless of whether the mock
+    // surfaces the SSE error as an explicit "error" event or silently ends the
+    // stream (a timing artefact of synchronous mock SSE delivery).
+    expect(eventTypes).not.toContain("start");
+  });
 });
diff --git a/src/agents/anthropic-transport-stream.ts b/src/agents/anthropic-transport-stream.ts
@@ -992,7 +992,6 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn {
           { ...params, stream: true },
           transportOptions.signal ? { signal: transportOptions.signal } : undefined,
         );
-        stream.push({ type: "start", partial: output as never });
         const blocks = output.content;
         const signatureDeltaIndexes = new Set<number>();
         const allowReasoningContentReplay = supportsReasoningContentReplay(model);
@@ -1130,6 +1129,11 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn {
               output.usage.cacheRead +
               output.usage.cacheWrite;
             calculateCost(model, output.usage);
+            // Defer start until after message_start so that pre-stream SSE errors
+            // (e.g. invalid thinking signatures) arrive before any non-error event
+            // is yielded, keeping yieldedOutput=false in pumpStreamWithRecovery
+            // and allowing the thinking-block recovery retry to fire.
+            stream.push({ type: "start", partial: output as never });
             continue;
           }
           if (event.type === "content_block_start") {
