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

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).
This commit is contained in:
openperf
2026-06-05 23:31:24 +08:00
parent 17203dc6ac
commit 399a243c64
2 changed files with 65 additions and 1 deletions
@@ -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");
});
});
+5 -1
View File
@@ -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") {