fix stream handler error

src/index.ts

@@ -168,10 +168,11 @@ async function run(options: RunOptions = {}) {
       await router(req, reply, config);
     }
   });
-  server.addHook("onSend", async (req, reply, payload) => {
+  server.addHook("onSend", (req, reply, payload, done) => {
     if (req.sessionId && req.url.startsWith("/v1/messages")) {
       if (payload instanceof ReadableStream) {
         if (req.agents) {
+          const abortController = new AbortController();
           const eventStream = payload.pipeThrough(new SSEParserTransform())
           let currentAgent: undefined | IAgent;
           let currentToolIndex = -1
@@ -181,7 +182,8 @@ async function run(options: RunOptions = {}) {
           const toolMessages: any[] = []
           const assistantMessages: any[] = []
           // 存储Anthropic格式的消息体，区分文本和工具类型
-          return rewriteStream(eventStream, async (data, controller) => {
+          return done(null, rewriteStream(eventStream, async (data, controller) => {
+            try {
               // 检测工具调用开始
               if (data.event === 'content_block_start' && data?.data?.content_block?.name) {
                 const agent = req.agents.find((name: string) => agentsManager.getAgent(name)?.tools.get(data.data.content_block.name))
@@ -254,6 +256,7 @@ async function run(options: RunOptions = {}) {
                 const stream = response.body!.pipeThrough(new SSEParserTransform())
                 const reader = stream.getReader()
                 while (true) {
+                  try {
                     const {value, done} = await reader.read();
                     if (done) {
                       break;
@@ -261,17 +264,47 @@ async function run(options: RunOptions = {}) {
                     if (['message_start', 'message_stop'].includes(value.event)) {
                       continue
                     }
+
+                    // 检查流是否仍然可写
+                    if (!controller.desiredSize) {
+                      console.log('Stream backpressure detected');
+                      break;
+                    }
+
                     controller.enqueue(value)
+                  }catch (readError: any) {
+                    if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') {
+                      console.log('Stream reading aborted due to client disconnect');
+                      abortController.abort(); // 中止所有相关操作
+                      break;
+                    }
+                    throw readError;
+                  }
+
                 }
                 return undefined
               }
               return data
-          }).pipeThrough(new SSESerializerTransform())
+            }catch (error: any) {
+              console.error('Unexpected error in stream processing:', error);
+
+              // 处理流提前关闭的错误
+              if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') {
+                console.log('Stream prematurely closed, aborting operations');
+                abortController.abort();
+                return undefined;
+              }
+
+              // 其他错误仍然抛出
+              throw error;
+            }
+          }).pipeThrough(new SSESerializerTransform()))
         }
 
         const [originalStream, clonedStream] = payload.tee();
         const read = async (stream: ReadableStream) => {
           const reader = stream.getReader();
+          try {
             while (true) {
               const { done, value } = await reader.read();
               if (done) break;
@@ -286,13 +319,25 @@ async function run(options: RunOptions = {}) {
                 sessionUsageCache.put(req.sessionId, message.usage);
               } catch {}
             }
+          } catch (readError: any) {
+            if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') {
+              console.log('Background read stream closed prematurely');
+            } else {
+              console.error('Error in background stream reading:', readError);
+            }
+          } finally {
+            reader.releaseLock();
+          }
         }
         read(clonedStream);
-        return originalStream
+        return done(null, originalStream)
       }
       sessionUsageCache.put(req.sessionId, payload.usage);
     }
-    return payload;
+    if (typeof payload ==='object' && payload.error) {
+      done(payload.error, null)
+    }
+    done(null, payload)
   });
 
   server.start();