refactor(process): share decoded NDJSON framing

src/ghidra/GhidraClient.ts

@@ -20,6 +20,7 @@ import { err, ok, type Result } from "../domain/result.js";
 import { silentLogger } from "../logger.js";
 import type { Logger } from "pino";
 import { PendingOperations } from "../process/PendingOperations.js";
+import { DecodedNdjsonLineBuffer } from "../process/DecodedNdjsonLineBuffer.js";
 import { ABORTED, waitForAbortable } from "../process/AbortablePromise.js";
 import {
   SharedStartupAttempt,
@@ -50,7 +51,6 @@ import {
   GHIDRA_PROCESS_DIAGNOSTIC_BYTES,
 } from "./GhidraDiagnostics.js";
 import { GhidraLaunchError, type GhidraLaunch } from "./GhidraLauncher.js";
-import { GhidraResponseBuffer } from "./GhidraResponseBuffer.js";
 import { GhidraResponseRouter } from "./GhidraResponseRouter.js";
 import { GhidraRequestQueue } from "./GhidraRequestQueue.js";
 import {
@@ -166,11 +166,11 @@ export class GhidraClient {
       }),
     protocolFailure: (message, cause) => this.#abortProtocol(message, cause),
   });
-  readonly #responseBuffer = new GhidraResponseBuffer({
-    onLine: (line) => this.#responseRouter.route(line),
-  });
-  readonly #onSocketData = (chunk: string): void =>
-    this.#responseBuffer.push(chunk);
+  readonly #responseLines = new DecodedNdjsonLineBuffer();
+  readonly #onSocketData = (chunk: string): void => {
+    for (const line of this.#responseLines.push(chunk))
+      this.#responseRouter.route(line);
+  };
   readonly #onSocketError = (): void => {
     this.#failAll(this.#failure("process", "Ghidra bridge socket failed"));
   };
@@ -974,7 +974,7 @@ export class GhidraClient {
       this.#detachSocket = undefined;
       this.#socket?.destroy();
       this.#socket = undefined;
-      this.#responseBuffer.reset();
+      this.#responseLines.reset();
     }
   }
 
@@ -1119,7 +1119,7 @@ export class GhidraClient {
       this.#endpointPath = undefined;
       this.#token = undefined;
       this.#runId = undefined;
-      this.#responseBuffer.reset();
+      this.#responseLines.reset();
       if (forcedStopFailure !== undefined) throw forcedStopFailure.cause;
       for (const failure of this.#cleanupFailures) failure.resolveCleanup();
       this.#cleanupFailures.clear();

src/ghidra/GhidraResponseBuffer.ts

@@ -1,37 +0,0 @@
-/** Callbacks for one newline-delimited Ghidra response stream. */
-export interface GhidraResponseBufferOptions {
-  readonly onLine: (line: string) => void;
-}
-
-/** Splits fragmented UTF-8 socket data into response lines. */
-export class GhidraResponseBuffer {
-  readonly #options: GhidraResponseBufferOptions;
-  #buffer = "";
-  #fragments: string[] = [];
-
-  constructor(options: GhidraResponseBufferOptions) {
-    this.#options = options;
-  }
-
-  /** Consume one decoded socket chunk. */
-  push(chunk: string): void {
-    this.#fragments.push(chunk);
-    if (!chunk.includes("\n")) return;
-    this.#buffer += this.#fragments.join("");
-    this.#fragments = [];
-    let newline = this.#buffer.indexOf("\n");
-    while (newline >= 0) {
-      const encodedLine = this.#buffer.slice(0, newline);
-      this.#buffer = this.#buffer.slice(newline + 1);
-      const line = encodedLine.trim();
-      if (line.length > 0) this.#options.onLine(line);
-      newline = this.#buffer.indexOf("\n");
-    }
-  }
-
-  /** Drop any incomplete line when a session closes. */
-  reset(): void {
-    this.#buffer = "";
-    this.#fragments = [];
-  }
-}

src/ghidra/GhidraResponseRouter.test.ts

@@ -0,0 +1,49 @@
+import { expect, it, vi } from "vitest";
+
+import type { JsonValue } from "../domain/jsonValue.js";
+import { err, ok, type Result } from "../domain/result.js";
+import { DecodedNdjsonLineBuffer } from "../process/DecodedNdjsonLineBuffer.js";
+import { PendingOperations } from "../process/PendingOperations.js";
+import { GhidraResponseRouter } from "./GhidraResponseRouter.js";
+import { GhidraSessionError } from "./GhidraSessionError.js";
+
+it("routes later Ghidra response lines after an unknown response id", async () => {
+  const pending = new PendingOperations<
+    number,
+    Result<JsonValue, GhidraSessionError>
+  >();
+  const protocolFailure = vi.fn();
+  const router = new GhidraResponseRouter({
+    pending,
+    nextId: () => 2,
+    remoteFailure: (failure) =>
+      new GhidraSessionError(
+        "remote",
+        failure.message,
+        {},
+        {
+          remoteCode: failure.code,
+        },
+      ),
+    protocolFailure: (message, cause) =>
+      protocolFailure(
+        new GhidraSessionError("protocol", message, {}, { cause }),
+      ),
+  });
+  const response = pending.wait(1, {
+    cancelledValue: () =>
+      err(new GhidraSessionError("cancelled", "request cancelled")),
+  });
+  const lines = new DecodedNdjsonLineBuffer();
+
+  for (const line of lines.push(
+    '{"id":2,"ok":true,"result":null}\n{"id":1,"ok":true,"result":"settled"}\n',
+  ))
+    router.route(line);
+
+  expect(protocolFailure).toHaveBeenCalledOnce();
+  expect(protocolFailure).toHaveBeenCalledWith(
+    expect.objectContaining({ kind: "protocol" }),
+  );
+  await expect(response).resolves.toEqual(ok("settled"));
+});

src/hopper/HopperResponseStream.test.ts

@@ -0,0 +1,70 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { HopperResponseStream } from "./HopperResponseStream.js";
+
+describe("Hopper response stream", () => {
+  it("stops on malformed input and resumes with buffered response lines", () => {
+    const accepted: number[] = [];
+    const abort = vi.fn();
+    const stream = new HopperResponseStream({
+      accept: (message) => {
+        accepted.push(message.id);
+        return true;
+      },
+      hasQueued: () => false,
+      nextRequestId: () => 4,
+      abort,
+    });
+
+    stream.push('not json\n{"id":2,"result":null}\n');
+    expect(accepted).toEqual([]);
+    expect(abort).toHaveBeenCalledOnce();
+
+    stream.push('{"id":3,"result":null}\n');
+    expect(accepted).toEqual([2, 3]);
+    expect(abort).toHaveBeenCalledOnce();
+  });
+
+  it("stops on a future response id and retains later complete lines", () => {
+    const accepted: number[] = [];
+    const abort = vi.fn();
+    const stream = new HopperResponseStream({
+      accept: (message) => {
+        if (message.id >= 3) return false;
+        accepted.push(message.id);
+        return true;
+      },
+      hasQueued: () => false,
+      nextRequestId: () => 3,
+      abort,
+    });
+
+    stream.push('{"id":3,"result":null}\n{"id":1,"result":null}\n');
+    expect(accepted).toEqual([]);
+    expect(abort).toHaveBeenCalledWith(
+      "Hopper returned an unknown response id",
+    );
+
+    stream.push('{"id":2,"result":null}\n');
+    expect(accepted).toEqual([1, 2]);
+    expect(abort).toHaveBeenCalledOnce();
+  });
+
+  it("discards incomplete lines on reset", () => {
+    const accepted: number[] = [];
+    const stream = new HopperResponseStream({
+      accept: (message) => {
+        accepted.push(message.id);
+        return true;
+      },
+      hasQueued: () => false,
+      nextRequestId: () => 2,
+      abort: vi.fn(),
+    });
+
+    stream.push('{"id":1,"result":null}');
+    stream.reset();
+    stream.push('{"id":1,"result":null}\n');
+    expect(accepted).toEqual([1]);
+  });
+});

src/hopper/HopperResponseStream.ts

@@ -2,6 +2,7 @@ import {
   parseBridgeMessageLine,
   type HopperBridgeMessage,
 } from "./protocol.js";
+import { DecodedNdjsonLineBuffer } from "../process/DecodedNdjsonLineBuffer.js";
 
 export interface HopperResponseStreamOptions {
   readonly accept: (message: HopperBridgeMessage) => boolean;
@@ -12,28 +13,17 @@ export interface HopperResponseStreamOptions {
 
 /** Incrementally validates Hopper NDJSON response frames. */
 export class HopperResponseStream {
-  #buffer = "";
-  #fragments: string[] = [];
+  readonly #lines = new DecodedNdjsonLineBuffer();
 
   constructor(readonly options: HopperResponseStreamOptions) {}
 
   push(chunk: string): void {
-    this.#fragments.push(chunk);
-    if (!chunk.includes("\n")) return;
-    this.#buffer += this.#fragments.join("");
-    this.#fragments = [];
-    let newline = this.#buffer.indexOf("\n");
-    while (newline >= 0) {
-      const line = this.#buffer.slice(0, newline).trim();
-      this.#buffer = this.#buffer.slice(newline + 1);
-      if (line.length > 0 && !this.#acceptLine(line)) return;
-      newline = this.#buffer.indexOf("\n");
-    }
+    for (const line of this.#lines.push(chunk))
+      if (!this.#acceptLine(line)) break;
   }
 
   reset(): void {
-    this.#buffer = "";
-    this.#fragments = [];
+    this.#lines.reset();
   }
 
   #acceptLine(line: string): boolean {

src/process/DecodedNdjsonLineBuffer.test.ts

@@ -0,0 +1,33 @@
+import { describe, expect, it } from "vitest";
+
+import { DecodedNdjsonLineBuffer } from "./DecodedNdjsonLineBuffer.js";
+
+describe("decoded NDJSON line buffer", () => {
+  it("joins fragments and yields trimmed nonblank complete lines", () => {
+    const lines = new DecodedNdjsonLineBuffer();
+
+    expect([...lines.push("  first ")]).toEqual([]);
+    expect([...lines.push("\n \t\n second \n unfinished")]).toEqual([
+      "first",
+      "second",
+    ]);
+    expect([...lines.push(" line\n")]).toEqual(["unfinished line"]);
+  });
+
+  it("retains unconsumed complete lines and reset discards them", () => {
+    const lines = new DecodedNdjsonLineBuffer();
+    const iterator = lines.push('invalid\n{"id":2}\n');
+
+    expect(iterator.next()).toEqual({ value: "invalid", done: false });
+    iterator.return(undefined);
+    expect([...lines.push('{"id":3}\n')]).toEqual(['{"id":2}', '{"id":3}']);
+
+    const nextIterator = lines.push(
+      "discard me\nwill also be discarded\nkeep partial",
+    );
+    expect(nextIterator.next()).toEqual({ value: "discard me", done: false });
+    nextIterator.return(undefined);
+    lines.reset();
+    expect([...lines.push("after reset\n")]).toEqual(["after reset"]);
+  });
+});

src/process/DecodedNdjsonLineBuffer.ts

@@ -0,0 +1,31 @@
+/** Incrementally splits decoded newline-delimited JSON socket data into lines. */
+export class DecodedNdjsonLineBuffer {
+  #buffer = "";
+  #fragments: string[] = [];
+
+  /**
+   * Consume a chunk and yield each complete, trimmed, nonblank line.
+   * Unconsumed complete lines remain buffered if iteration stops early.
+   */
+  *push(chunk: string): Generator<string> {
+    this.#fragments.push(chunk);
+    if (!chunk.includes("\n")) return;
+    this.#buffer += this.#fragments.join("");
+    this.#fragments = [];
+
+    let newline = this.#buffer.indexOf("\n");
+    while (newline >= 0) {
+      const encodedLine = this.#buffer.slice(0, newline);
+      this.#buffer = this.#buffer.slice(newline + 1);
+      const line = encodedLine.trim();
+      if (line.length > 0) yield line;
+      newline = this.#buffer.indexOf("\n");
+    }
+  }
+
+  /** Drop incomplete and unconsumed lines when a session closes. */
+  reset(): void {
+    this.#buffer = "";
+    this.#fragments = [];
+  }
+}