From 24119a360e7e288ea364a5b9d2bd19e1db145193 Mon Sep 17 00:00:00 2001 From: 0disoft Date: Sat, 31 Oct 2026 19:24:29 +0900 Subject: [PATCH 1/2] perf(devin): batch consumed stream chunk removal --- scripts/test-layout/layout.json | 2 +- src/adapters/devin/cloud-direct/chat.ts | 18 ++++-- structure/transports/byte-accounting.md | 6 ++ tests/fixtures/test-layout-expected.json | 2 +- tests/providers/devin-stream-queue.test.ts | 67 ++++++++++++++++++++++ 5 files changed, 87 insertions(+), 8 deletions(-) create mode 100644 tests/providers/devin-stream-queue.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 3232f870b0f..58c19d45c55 100644 --- a/scripts/test-layout/layout.json +++ b/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", diff --git a/src/adapters/devin/cloud-direct/chat.ts b/src/adapters/devin/cloud-direct/chat.ts index 435bfca98b0..5be38656faf 100644 --- a/src/adapters/devin/cloud-direct/chat.ts +++ b/src/adapters/devin/cloud-direct/chat.ts @@ -1392,8 +1392,8 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator 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.splice(0, consumed); } // Track the idle timer at outer scope so the finally block can clear it diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index ff55305299f..42ba1b7019b 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -140,6 +140,12 @@ envelopes, and real HTTP streams; `tests/server/inbound-body-budget.test.ts` cov ## Stream-buffer accounting +Devin Connect-RPC in `src/adapters/devin/cloud-direct/chat.ts` drains consumed chunk prefixes +in one array operation, or clears the queue when no bytes remain. 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 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. + `src/lib/pinned-http.ts` pauses its socket response when the receiving Web Stream queue fills and resumes when the reader asks for more. For gzip and deflate, the final decoded queue also gates subsequent socket input even when the decoder would accept more coded bytes. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index b97dd7e3e9c..93b4723a3c5 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -810,7 +810,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", diff --git a/tests/providers/devin-stream-queue.test.ts b/tests/providers/devin-stream-queue.test.ts new file mode 100644 index 00000000000..1b3ef24e918 --- /dev/null +++ b/tests/providers/devin-stream-queue.test.ts @@ -0,0 +1,67 @@ +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({ + 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; + let removals = 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) { + if (args[0] === 0 && args[1] && Buffer.isBuffer(this[0])) removals++; + return Reflect.apply(splice, this, args); + }); + let events: CloudChatEvent[]; + try { events = await collect(wire, 4); } + finally { shifting.mockRestore(); splicing.mockRestore(); } + expect(events.filter(event => event.kind === "text").map(event => event.text)).toEqual([text]); + expect(removals).toBeLessThanOrEqual(8); +}); + +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 끝"]); + }); +} From 36f795edc054c4f34df77ee50b0de4c84da21474 Mon Sep 17 00:00:00 2001 From: 0disoft Date: Sat, 31 Oct 2026 19:32:11 +0900 Subject: [PATCH 2/2] perf(devin): compact stream chunks without a discarded array --- src/adapters/devin/cloud-direct/chat.ts | 5 ++++- structure/transports/byte-accounting.md | 7 ++++--- tests/providers/devin-stream-queue.test.ts | 15 ++++++++++++--- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/src/adapters/devin/cloud-direct/chat.ts b/src/adapters/devin/cloud-direct/chat.ts index 5be38656faf..05a02c58bfb 100644 --- a/src/adapters/devin/cloud-direct/chat.ts +++ b/src/adapters/devin/cloud-direct/chat.ts @@ -1446,7 +1446,10 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator 0) chunkQueue.splice(0, consumed); + if (consumed > 0) { + chunkQueue.copyWithin(0, consumed); + chunkQueue.length -= consumed; + } } // Track the idle timer at outer scope so the finally block can clear it diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 42ba1b7019b..db5f1ddb462 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -140,10 +140,11 @@ envelopes, and real HTTP streams; `tests/server/inbound-body-budget.test.ts` cov ## Stream-buffer accounting -Devin Connect-RPC in `src/adapters/devin/cloud-direct/chat.ts` drains consumed chunk prefixes -in one array operation, or clears the queue when no bytes remain. Partial head slices and later +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 on fragmented frames and covers split/coalesced headers and payloads, +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. `src/lib/pinned-http.ts` pauses its socket response when the receiving Web Stream queue fills diff --git a/tests/providers/devin-stream-queue.test.ts b/tests/providers/devin-stream-queue.test.ts index 1b3ef24e918..f318b3a1801 100644 --- a/tests/providers/devin-stream-queue.test.ts +++ b/tests/providers/devin-stream-queue.test.ts @@ -37,20 +37,29 @@ test("Devin fragmented frame drains use bounded queue-prefix removals", async () 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) { - if (args[0] === 0 && args[1] && Buffer.isBuffer(this[0])) removals++; - return Reflect.apply(splice, this, args); + 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) { + 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(); } + 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]) {