diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index ed25a471a2d..55bae0134d2 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..05a02c58bfb 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.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 a93cd509622..8a18875c1be 100644 --- a/structure/transports/byte-accounting.md +++ b/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 diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 79fd72e9af0..d6cc69371a2 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/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", diff --git a/tests/providers/devin-stream-queue.test.ts b/tests/providers/devin-stream-queue.test.ts new file mode 100644 index 00000000000..f318b3a1801 --- /dev/null +++ b/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({ + 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) { + 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(); 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 끝"]); + }); +}