Use stream reader instead of async iteration for reading messages

This makes the TS library not dependent on node types so it can also run
in the browser.

It is likely the previous code would be ok, but this method is supported
in more environments

Signed-off-by: Ben Brandt <benjamin.j.brandt@gmail.com>

Author: benbrandt

typescript/acp.test.ts

@@ -1,5 +1,4 @@
 import { describe, it, expect, beforeEach } from "vitest";
-import { TransformStream } from "node:stream/web";
 import {
   Agent,
   ClientSideConnection,

typescript/acp.ts

@@ -2,8 +2,6 @@ import { z } from "zod";
 import * as schema from "./schema.js";
 export * from "./schema.js";
 
-import { WritableStream, ReadableStream } from "node:stream/web";
-
 /**
  * An agent-side connection to a client.
  *
@@ -605,39 +603,51 @@ class Connection {
   async #receive(output: ReadableStream<Uint8Array>) {
     let content = "";
     const decoder = new TextDecoder();
-    for await (const chunk of output) {
-      content += decoder.decode(chunk, { stream: true });
-      const lines = content.split("\n");
-      content = lines.pop() || "";
-
-      for (const line of lines) {
-        const trimmedLine = line.trim();
-
-        if (trimmedLine) {
-          let id;
-          try {
-            const message = JSON.parse(trimmedLine);
-            id = message.id;
-            this.#processMessage(message);
-          } catch (err) {
-            console.error(
-              "Unexpected error during message processing:",
-              trimmedLine,
-              err,
-            );
-            if (id) {
-              this.#sendMessage({
-                jsonrpc: "2.0",
-                id,
-                error: {
-                  code: -32700,
-                  message: "Parse error",
-                },
-              });
+    const reader = output.getReader();
+    try {
+      while (true) {
+        const { value, done } = await reader.read();
+        if (done) {
+          break;
+        }
+        if (!value) {
+          continue;
+        }
+        content += decoder.decode(value, { stream: true });
+        const lines = content.split("\n");
+        content = lines.pop() || "";
+
+        for (const line of lines) {
+          const trimmedLine = line.trim();
+
+          if (trimmedLine) {
+            let id;
+            try {
+              const message = JSON.parse(trimmedLine);
+              id = message.id;
+              this.#processMessage(message);
+            } catch (err) {
+              console.error(
+                "Unexpected error during message processing:",
+                trimmedLine,
+                err,
+              );
+              if (id) {
+                this.#sendMessage({
+                  jsonrpc: "2.0",
+                  id,
+                  error: {
+                    code: -32700,
+                    message: "Parse error",
+                  },
+                });
+              }
             }
           }
         }
       }
+    } finally {
+      reader.releaseLock();
     }
   }
 

typescript/examples/agent.ts

@@ -2,7 +2,6 @@
 
 import { AgentSideConnection, Agent, PROTOCOL_VERSION } from "../acp.js";
 import * as schema from "../schema.js";
-import { WritableStream, ReadableStream } from "node:stream/web";
 import { Readable, Writable } from "node:stream";
 
 interface AgentSession {

typescript/examples/client.ts

@@ -3,7 +3,6 @@
 import { spawn } from "node:child_process";
 import { fileURLToPath } from "node:url";
 import { dirname, join } from "node:path";
-import { WritableStream, ReadableStream } from "node:stream/web";
 import { Writable, Readable } from "node:stream";
 import readline from "node:readline/promises";