[WRONG BRANCH] fix(claude): preserve pre-output keepalives

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;

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<Uint8Array>({
+      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"));