perf(devin): batch consumed stream chunk removal

scripts/test-layout/layout.json

@@ -898,7 +898,7 @@
     "devin-cli-authmode-migration.test.ts": "providers",
     "devin-effort-ladder.test.ts": "providers",
     "devin-family-resolution.test.ts": "providers",
-    "devin-hardening.test.ts": "providers",
+    "devin-hardening.test.ts": "providers", "devin-stream-queue.test.ts": "providers",
     "devin-image-passthrough.test.ts": "providers",
     "devin-live-models.test.ts": "providers",
     "devin-login.test.ts": "providers",

src/adapters/devin/cloud-direct/chat.ts

@@ -1392,8 +1392,8 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator<C
   // chunk])` per chunk — O(n²) over a long stream because every chunk copies
   // every buffered byte again. Now we keep a queue of arriving chunks with a
   // running offset; we only `Buffer.concat` when a frame straddles a chunk
-  // boundary, and we slice/drop fully-consumed chunks immediately. For
-  // typical 50-200KB responses this is ~5x faster and produces zero waste.
+  // boundary. A drain clears the queue or removes its consumed prefix once,
+  // instead of shifting the array separately for every consumed chunk.
   const chunkQueue: Buffer[] = [];
   let queuedBytes = 0;
   // Bun + Node ReadableStream readers diverge on the type-level shape
@@ -1430,17 +1430,26 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator<C
   /** Drop the first `n` bytes from the chunk queue. */
   function drop(n: number): void {
     queuedBytes -= n;
+    if (queuedBytes === 0) {
+      chunkQueue.length = 0;
+      return;
+    }
     let remaining = n;
-    while (remaining > 0 && chunkQueue.length > 0) {
-      const head = chunkQueue[0];
+    let consumed = 0;
+    while (remaining > 0 && consumed < chunkQueue.length) {
+      const head = chunkQueue[consumed];
       if (head.length <= remaining) {
-        chunkQueue.shift();
+        consumed++;
         remaining -= head.length;
       } else {
-        chunkQueue[0] = head.slice(remaining);
+        chunkQueue[consumed] = head.slice(remaining);
         remaining = 0;
       }
     }
+    if (consumed > 0) {
+      chunkQueue.copyWithin(0, consumed);
+      chunkQueue.length -= consumed;
+    }
   }
 
   // Track the idle timer at outer scope so the finally block can clear it

structure/transports/byte-accounting.md

@@ -151,6 +151,13 @@ The turn limit and transport-read accounting remain in force. `tests/providers/c
 covers exact-limit Unicode records across whole/one-byte reads and newline/EOF termination,
 one-byte overflow, another retained owner, consumer return and upstream cancellation.
 
+Devin Connect-RPC in `src/adapters/devin/cloud-direct/chat.ts` compacts surviving chunks in place
+and truncates the queue, or clears it when no bytes remain. No deleted-chunk
+result array is allocated. Partial head slices and later
+frames retain their order without one `shift()` per chunk. `tests/providers/devin-stream-queue.test.ts`
+bounds prefix-removal calls and discarded-array slots on fragmented frames and covers split/coalesced headers and payloads,
+Unicode, empty frames and gzip. Existing frame/inflation caps, EOS handling and teardown remain in force.
+
 Native Ollama NDJSON in `src/adapters/ollama-native.ts` measures new decoded fragments without
 allocating UTF-8 arrays. A fragment without a newline extends the cached residual byte count;
 only a fragment containing a newline triggers splitting of the accumulated record. Complete

tests/fixtures/test-layout-expected.json

@@ -831,7 +831,7 @@
   "devin-adapter-reset-wait.test.ts": "providers", "devin-adapter.test.ts": "providers",
   "devin-chat-wire-fixes.test.ts": "providers", "devin-cli-authmode-migration.test.ts": "providers",
   "devin-effort-ladder.test.ts": "providers", "devin-family-resolution.test.ts": "providers",
-  "devin-hardening.test.ts": "providers", "devin-image-passthrough.test.ts": "providers",
+  "devin-hardening.test.ts": "providers", "devin-stream-queue.test.ts": "providers", "devin-image-passthrough.test.ts": "providers",
   "devin-live-models.test.ts": "providers", "devin-login.test.ts": "providers",
   "devin-quota.test.ts": "providers", "devin-output-budget.test.ts": "providers",
   "devin-prompt-cache.test.ts": "providers", "devin-trajectory.test.ts": "providers",

tests/providers/devin-stream-queue.test.ts

@@ -0,0 +1,76 @@
+import { expect, spyOn, test } from "bun:test";
+import { streamChatEvents, type CloudChatEvent } from "../../src/adapters/devin/cloud-direct/chat";
+import { encodeString, frameConnectStream } from "../../src/adapters/devin/cloud-direct/wire";
+
+function trailer() {
+  return Buffer.from([2, 0, 0, 0, 2, 123, 125]);
+}
+
+async function collect(wire: Buffer, chunkBytes: number) {
+  let offset = 0;
+  const body = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (offset === wire.length) { controller.close(); return; }
+      const end = Math.min(wire.length, offset + chunkBytes);
+      controller.enqueue(wire.subarray(offset, end));
+      offset = end;
+    },
+  }, { highWaterMark: 0 });
+  const before = process.env.OPENCODEX_DEVIN_SEND_USER_JWT;
+  process.env.OPENCODEX_DEVIN_SEND_USER_JWT = "0";
+  try {
+    const events: CloudChatEvent[] = [];
+    for await (const event of streamChatEvents({
+      apiKey: "devin-session-token$header.payload.signature",
+      modelUid: "swe-1-7", messages: [], catalog: null,
+      executor: (async () => new Response(body)) as typeof fetch,
+    })) events.push(event);
+    return events;
+  } finally {
+    if (before === undefined) delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT;
+    else process.env.OPENCODEX_DEVIN_SEND_USER_JWT = before;
+  }
+}
+
+test("Devin fragmented frame drains use bounded queue-prefix removals", async () => {
+  const text = "x".repeat(64 * 1024);
+  const wire = Buffer.concat([frameConnectStream(encodeString(3, text), false), trailer()]);
+  const shift = Array.prototype.shift;
+  const splice = Array.prototype.splice;
+  const copyWithin = Array.prototype.copyWithin;
+  let removals = 0;
+  let discardedChunkSlots = 0;
+  const shifting = spyOn(Array.prototype, "shift").mockImplementation(function () {
+    if (Buffer.isBuffer(this[0])) removals++;
+    return Reflect.apply(shift, this, []);
+  });
+  const splicing = spyOn(Array.prototype, "splice").mockImplementation(function (...args: Parameters<typeof splice>) {
+    const prefix = args[0] === 0 && args[1] && Buffer.isBuffer(this[0]);
+    const removed = Reflect.apply(splice, this, args);
+    if (prefix) { removals++; discardedChunkSlots += removed.length; }
+    return removed;
+  });
+  const copying = spyOn(Array.prototype, "copyWithin").mockImplementation(function (...args: Parameters<typeof copyWithin>) {
+    if (args[0] === 0 && args[1] > 0 && Buffer.isBuffer(this[0])) removals++;
+    return Reflect.apply(copyWithin, this, args);
+  });
+  let events: CloudChatEvent[];
+  try { events = await collect(wire, 4); }
+  finally { shifting.mockRestore(); splicing.mockRestore(); copying.mockRestore(); }
+  expect(events.filter(event => event.kind === "text").map(event => event.text)).toEqual([text]);
+  expect(removals).toBeLessThanOrEqual(8);
+  expect(discardedChunkSlots).toBe(0);
+});
+
+for (const chunkBytes of [1, 3, 5, 7, 17, 65536]) {
+  test(`Devin queue retains frame order and partial heads across ${chunkBytes}-byte reads`, async () => {
+    const wire = Buffer.concat([
+      frameConnectStream(encodeString(3, "한글 😀"), false),
+      frameConnectStream(Buffer.alloc(0), false),
+      frameConnectStream(encodeString(3, "compressed 끝"), true),
+      trailer(),
+    ]);
+    const events = await collect(wire, chunkBytes);
+    expect(events.filter(event => event.kind === "text").map(event => event.text)).toEqual(["한글 😀", "compressed 끝"]);
+  });
+}