From 390a5badbd4de30672fa13140ff718cf3a045a95 Mon Sep 17 00:00:00 2001 From: luvs01 Date: Tue, 11 Aug 2026 09:42:15 +0900 Subject: [PATCH] fix(claude): preserve pre-output keepalives --- src/claude/outbound.ts | 10 +++++----- tests/claude-outbound.test.ts | 28 ++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 5 deletions(-) diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index 095d332d49..ec04fcf8ca 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -267,12 +267,12 @@ export function responsesSseToAnthropicSse( emit("message_start", { type: "message_start", message: messageSnapshot(model) }); emit("ping", { type: "ping" }); }; - // Once a semantic Anthropic message has started, keepalive pings protect remote - // deployments behind LB/NAT idle timeouts. Transport-only Responses prelude frames - // must not manufacture a message before a possible initial error. + // Keepalive pings protect remote deployments behind LB/NAT idle timeouts, including + // while waiting for the first semantic frame. A ping is transport-only, so it does + // not manufacture an Anthropic message before a possible initial error. if (pingIntervalMs > 0) { pingTimer = setInterval(() => { - if (terminated || !started) return; + if (terminated) return; try { emit("ping", { type: "ping" }); } catch { /* controller torn down; the read loop is ending anyway */ } @@ -366,7 +366,7 @@ export function responsesSseToAnthropicSse( // Transport prelude only. Start Anthropic framing on semantic output or completion. break; case "response.heartbeat": - if (started) emit("ping", { type: "ping" }); + emit("ping", { type: "ping" }); break; case "response.output_text.delta": { if (typeof data.delta !== "string" || data.delta.length === 0) break; diff --git a/tests/claude-outbound.test.ts b/tests/claude-outbound.test.ts index f7e36a67ab..1281fa4d68 100644 --- a/tests/claude-outbound.test.ts +++ b/tests/claude-outbound.test.ts @@ -617,6 +617,34 @@ describe("claude outbound SSE", () => { expect(events.at(-1)!.name).toBe("message_stop"); }); + test("idle keepalive pings flow before semantic output during upstream silence", async () => { + const PING_INTERVAL_MS = 25; + const SILENCE_MS = 300; + const encoder = new TextEncoder(); + const upstream = new ReadableStream({ + async start(controller) { + controller.enqueue(encoder.encode(sse("response.created", { response: {} }))); + await new Promise(r => setTimeout(r, SILENCE_MS)); + controller.enqueue(encoder.encode(sse("response.completed", { response: { status: "completed" } }))); + controller.close(); + }, + }); + const events = await collectEvents(responsesSseToAnthropicSse(upstream, "m", { pingIntervalMs: PING_INTERVAL_MS })); + const messageStartIndex = events.findIndex(e => e.name === "message_start"); + expect(events.slice(0, messageStartIndex).filter(e => e.name === "ping").length).toBeGreaterThanOrEqual(2); + expect(events.at(-1)!.name).toBe("message_stop"); + }); + + test("upstream heartbeats remain transport-only before an initial error", async () => { + const upstream = [ + sse("response.created", { response: {} }), + sse("response.heartbeat", {}), + sse("response.failed", { response: { status: "failed", error: { status: 500, message: "failed" } } }), + ].join(""); + const events = await collectEvents(responsesSseToAnthropicSse(streamFrom(upstream), "m", { pingIntervalMs: 0 })); + expect(events.map(e => e.name)).toEqual(["ping", "error"]); + }); + test("no-output completed still emits a valid empty message", async () => { const upstream = sse("response.created", { response: {} }) + sse("response.completed", { response: { status: "completed" } }); const events = await collectEvents(responsesSseToAnthropicSse(streamFrom(upstream), "m"));