Fix _fetch_next_chunk for StreamProcessor due to cutting off output (#42)

Author: jamie-retool

lib/resources/abstraction/sandbox.ts

@@ -571,7 +571,30 @@ export class SandboxProcessStream {
     const output =
       typeof this.fetch_fn === "function" ? await this.fetch_fn() : "";
     if (output === this._last_output) return "";
-    const newOutput = output.slice(this._last_output.length);
+
+    let newOutput: string;
+    if (output.startsWith(this._last_output)) {
+      // Happy path: server buffer is append-only
+      newOutput = output.slice(this._last_output.length);
+    } else {
+      // Buffer rotation detected: find longest suffix of _last_output
+      // that matches a prefix of output (overlap detection).
+      // NOTE: This is a best-effort heuristic. If new data coincidentally
+      // matches a suffix of _last_output (e.g., repeated lines), the
+      // overlap may be falsely detected and that data silently skipped.
+      // Without server-side sequence numbers this is unavoidable; we
+      // prefer this over the alternative (returning the entire buffer
+      // and producing large duplicates on every rotation).
+      newOutput = output;
+      for (let i = 1; i < this._last_output.length; i++) {
+        const suffix = this._last_output.slice(i);
+        if (output.startsWith(suffix)) {
+          newOutput = output.slice(suffix.length);
+          break;
+        }
+      }
+    }
+
     this._last_output = output;
     return newOutput;
   }

tests/sandbox-stream.test.ts

@@ -91,4 +91,93 @@ describe("SandboxProcessStream async iterator", () => {
 
     expect(lines).toEqual(["only-line\n"]);
   });
+
+  test("recovers new data when server buffer rotates (drops old data from front)", async () => {
+    // Simulate a server with a fixed-size stdout buffer of 15 chars.
+    // Once output exceeds that, old data is dropped from the front.
+    const BUFFER_SIZE = 15;
+    const allLines = [
+      "AAAA\n", // cumulative: 5  chars
+      "BBBB\n", // cumulative: 10 chars
+      "CCCC\n", // cumulative: 15 chars — buffer full
+      "DDDD\n", // cumulative: 20 chars — "AAAA\n" dropped from front
+      "EEEE\n", // cumulative: 25 chars — "BBBB\n" dropped from front
+    ];
+
+    // Build the sequence of server responses, trimming the front at BUFFER_SIZE
+    let fullOutput = "";
+    const serverResponses: string[] = [];
+    for (const line of allLines) {
+      fullOutput += line;
+      if (fullOutput.length > BUFFER_SIZE) {
+        fullOutput = fullOutput.slice(fullOutput.length - BUFFER_SIZE);
+      }
+      serverResponses.push(fullOutput);
+    }
+    // serverResponses:
+    // [0] "AAAA\n"              (5 chars)
+    // [1] "AAAA\nBBBB\n"        (10 chars)
+    // [2] "AAAA\nBBBB\nCCCC\n"  (15 chars) — buffer full
+    // [3] "BBBB\nCCCC\nDDDD\n"  (15 chars) — "AAAA\n" dropped
+    // [4] "CCCC\nDDDD\nEEEE\n"  (15 chars) — "BBBB\n" dropped
+
+    let fetchIndex = 0;
+    const stream = new SandboxProcessStream(
+      makeMockProcess(0),
+      () => {
+        if (fetchIndex < serverResponses.length) {
+          return serverResponses[fetchIndex++];
+        }
+        return serverResponses[serverResponses.length - 1];
+      }
+    );
+
+    const lines: string[] = [];
+    for await (const line of stream) {
+      lines.push(line);
+    }
+
+    expect(lines).toEqual([
+      "AAAA\n",
+      "BBBB\n",
+      "CCCC\n",
+      "DDDD\n",
+      "EEEE\n",
+    ]);
+  });
+
+  test("returns entire output when rotation leaves no overlap with previous buffer", async () => {
+    // Simulate a server where the buffer is completely replaced between
+    // two consecutive fetches — zero characters in common.
+    const serverResponses = [
+      "AAAA\nBBBB\n",   // fetch 0: initial data
+      "CCCC\nDDDD\n",   // fetch 1: completely different content (no overlap)
+    ];
+
+    let fetchIndex = 0;
+    const stream = new SandboxProcessStream(
+      makeMockProcess(0),
+      () => {
+        if (fetchIndex < serverResponses.length) {
+          return serverResponses[fetchIndex++];
+        }
+        return serverResponses[serverResponses.length - 1];
+      }
+    );
+
+    const lines: string[] = [];
+    for await (const line of stream) {
+      lines.push(line);
+    }
+
+    // With no overlap the fallback returns the entire new buffer,
+    // which may duplicate data already yielded — but duplicates are
+    // preferable to silent data loss.
+    expect(lines).toEqual([
+      "AAAA\n",
+      "BBBB\n",
+      "CCCC\n",
+      "DDDD\n",
+    ]);
+  });
 });