fix stream event idle for 10s, claude code will retry the request with stream=false

Author: caozhiyuan

src/lib/utils.ts

@@ -24,3 +24,25 @@ export const cacheVSCodeVersion = async () => {
 
   consola.info(`Using VSCode version: ${response}`)
 }
+
+export const setupPingInterval = (
+  stream: {
+    writeSSE: (data: { event: string; data: string }) => Promise<void>
+  },
+  intervalMs: number = 3000,
+) => {
+  const pingInterval = setInterval(async () => {
+    try {
+      await stream.writeSSE({
+        event: "ping",
+        data: "",
+      })
+      consola.debug("Sent ping")
+    } catch (error) {
+      consola.warn("Failed to send ping:", error)
+      clearInterval(pingInterval)
+    }
+  }, intervalMs)
+
+  return pingInterval
+}

src/routes/chat-completions/handler.ts

@@ -7,6 +7,7 @@ import { awaitApproval } from "~/lib/approval"
 import { checkRateLimit } from "~/lib/rate-limit"
 import { state } from "~/lib/state"
 import { getTokenCount } from "~/lib/tokenizer"
+import { setupPingInterval } from "~/lib/utils"
 import { isNullish } from "~/lib/utils"
 import {
   createChatCompletions,
@@ -56,9 +57,15 @@ export async function handleCompletion(c: Context) {
 
   consola.debug("Streaming response")
   return streamSSE(c, async (stream) => {
-    for await (const chunk of response) {
-      consola.debug("Streaming chunk:", JSON.stringify(chunk))
-      await stream.writeSSE(chunk as SSEMessage)
+    const pingInterval = setupPingInterval(stream)
+
+    try {
+      for await (const chunk of response) {
+        consola.debug("Streaming chunk:", JSON.stringify(chunk))
+        await stream.writeSSE(chunk as SSEMessage)
+      }
+    } finally {
+      clearInterval(pingInterval)
     }
   })
 }

src/routes/messages/handler.ts

@@ -6,6 +6,7 @@ import { streamSSE } from "hono/streaming"
 import { awaitApproval } from "~/lib/approval"
 import { checkRateLimit } from "~/lib/rate-limit"
 import { state } from "~/lib/state"
+import { setupPingInterval } from "~/lib/utils"
 import {
   createResponsesStreamState,
   translateResponsesStreamEvent,
@@ -84,33 +85,39 @@ const handleWithChatCompletions = async (
 
   consola.debug("Streaming response from Copilot")
   return streamSSE(c, async (stream) => {
+    const pingInterval = setupPingInterval(stream)
+
     const streamState: AnthropicStreamState = {
       messageStartSent: false,
       contentBlockIndex: 0,
       contentBlockOpen: false,
       toolCalls: {},
     }
 
-    for await (const rawEvent of response) {
-      consola.debug("Copilot raw stream event:", JSON.stringify(rawEvent))
-      if (rawEvent.data === "[DONE]") {
-        break
-      }
+    try {
+      for await (const rawEvent of response) {
+        consola.debug("Copilot raw stream event:", JSON.stringify(rawEvent))
+        if (rawEvent.data === "[DONE]") {
+          break
+        }
 
-      if (!rawEvent.data) {
-        continue
-      }
+        if (!rawEvent.data) {
+          continue
+        }
 
-      const chunk = JSON.parse(rawEvent.data) as ChatCompletionChunk
-      const events = translateChunkToAnthropicEvents(chunk, streamState)
+        const chunk = JSON.parse(rawEvent.data) as ChatCompletionChunk
+        const events = translateChunkToAnthropicEvents(chunk, streamState)
 
-      for (const event of events) {
-        consola.debug("Translated Anthropic event:", JSON.stringify(event))
-        await stream.writeSSE({
-          event: event.type,
-          data: JSON.stringify(event),
-        })
+        for (const event of events) {
+          consola.debug("Translated Anthropic event:", JSON.stringify(event))
+          await stream.writeSSE({
+            event: event.type,
+            data: JSON.stringify(event),
+          })
+        }
       }
+    } finally {
+      clearInterval(pingInterval)
     }
   })
 }
@@ -135,45 +142,51 @@ const handleWithResponsesApi = async (
   if (responsesPayload.stream && isAsyncIterable(response)) {
     consola.debug("Streaming response from Copilot (Responses API)")
     return streamSSE(c, async (stream) => {
-      const streamState = createResponsesStreamState()
+      const pingInterval = setupPingInterval(stream)
 
-      for await (const chunk of response) {
-        const eventName = chunk.event
-        if (eventName === "ping") {
-          await stream.writeSSE({ event: "ping", data: "" })
-          continue
-        }
+      const streamState = createResponsesStreamState()
 
-        const data = chunk.data
-        if (!data) {
-          continue
+      try {
+        for await (const chunk of response) {
+          const eventName = chunk.event
+          if (eventName === "ping") {
+            await stream.writeSSE({ event: "ping", data: "" })
+            continue
+          }
+
+          const data = chunk.data
+          if (!data) {
+            continue
+          }
+
+          consola.debug("Responses raw stream event:", data)
+
+          const events = translateResponsesStreamEvent(
+            JSON.parse(data) as ResponseStreamEvent,
+            streamState,
+          )
+          for (const event of events) {
+            const eventData = JSON.stringify(event)
+            consola.debug("Translated Anthropic event:", eventData)
+            await stream.writeSSE({
+              event: event.type,
+              data: eventData,
+            })
+          }
         }
 
-        consola.debug("Responses raw stream event:", data)
-
-        const events = translateResponsesStreamEvent(
-          JSON.parse(data) as ResponseStreamEvent,
-          streamState,
-        )
-        for (const event of events) {
-          const eventData = JSON.stringify(event)
-          consola.debug("Translated Anthropic event:", eventData)
+        if (!streamState.messageCompleted) {
+          consola.warn(
+            "Responses stream ended without completion; sending fallback message_stop",
+          )
+          const fallback = { type: "message_stop" as const }
           await stream.writeSSE({
-            event: event.type,
-            data: eventData,
+            event: fallback.type,
+            data: JSON.stringify(fallback),
           })
         }
-      }
-
-      if (!streamState.messageCompleted) {
-        consola.warn(
-          "Responses stream ended without completion; sending fallback message_stop",
-        )
-        const fallback = { type: "message_stop" as const }
-        await stream.writeSSE({
-          event: fallback.type,
-          data: JSON.stringify(fallback),
-        })
+      } finally {
+        clearInterval(pingInterval)
       }
     })
   }

src/routes/responses/handler.ts

@@ -6,6 +6,7 @@ import { streamSSE } from "hono/streaming"
 import { awaitApproval } from "~/lib/approval"
 import { checkRateLimit } from "~/lib/rate-limit"
 import { state } from "~/lib/state"
+import { setupPingInterval } from "~/lib/utils"
 import {
   createResponses,
   type ResponsesPayload,
@@ -52,17 +53,7 @@ export const handleResponses = async (c: Context) => {
   if (isStreamingRequested(payload) && isAsyncIterable(response)) {
     consola.debug("Forwarding native Responses stream")
     return streamSSE(c, async (stream) => {
-      const pingInterval = setInterval(async () => {
-        try {
-          await stream.writeSSE({
-            event: "ping",
-            data: JSON.stringify({ timestamp: Date.now() }),
-          })
-        } catch (error) {
-          consola.warn("Failed to send ping:", error)
-          clearInterval(pingInterval)
-        }
-      }, 3000)
+      const pingInterval = setupPingInterval(stream)
 
       try {
         for await (const chunk of response) {