refactor(worker-utils): share bounded stream reader

packages/worker-utils/package.json

@@ -49,7 +49,8 @@
     "./r2-client": "./src/r2-client.ts",
     "./direct-byok-model": "./src/direct-byok-model.ts",
     "./user-id-encoding": "./src/user-id-encoding.ts",
-    "./github-authorization": "./src/github-authorization.ts"
+    "./github-authorization": "./src/github-authorization.ts",
+    "./bounded-stream-reader": "./src/bounded-stream-reader.ts"
   },
   "scripts": {
     "test": "vitest run",

packages/worker-utils/src/bounded-stream-reader.ts

@@ -0,0 +1,45 @@
+export type BoundedStreamReadResult =
+  | { ok: true; bytes: Uint8Array }
+  | { ok: false; reason: 'too_large' };
+
+export async function readBoundedStream(
+  stream: ReadableStream<Uint8Array>,
+  declaredBytes: number
+): Promise<BoundedStreamReadResult> {
+  const reader = stream.getReader();
+
+  try {
+    const chunks: Uint8Array[] = [];
+    let totalBytes = 0;
+
+    while (true) {
+      const result = await reader.read();
+      if (result.done) {
+        const bytes = new Uint8Array(totalBytes);
+        let offset = 0;
+        for (const chunk of chunks) {
+          bytes.set(chunk, offset);
+          offset += chunk.byteLength;
+        }
+        return { ok: true, bytes };
+      }
+
+      totalBytes += result.value.byteLength;
+      if (totalBytes > declaredBytes) {
+        await cancelReader(reader);
+        return { ok: false, reason: 'too_large' };
+      }
+      chunks.push(result.value);
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}
+
+async function cancelReader(reader: ReadableStreamDefaultReader<Uint8Array>): Promise<void> {
+  try {
+    await reader.cancel('stream exceeded ingest byte limit');
+  } catch {
+    // The typed overflow result remains authoritative if cancellation itself fails.
+  }
+}

services/session-ingest/src/ingest/bounded-stream-reader.ts

@@ -1,45 +1,4 @@
-export type BoundedStreamReadResult =
-  | { ok: true; bytes: Uint8Array }
-  | { ok: false; reason: 'too_large' };
-
-export async function readBoundedStream(
-  stream: ReadableStream<Uint8Array>,
-  declaredBytes: number
-): Promise<BoundedStreamReadResult> {
-  const reader = stream.getReader();
-
-  try {
-    const chunks: Uint8Array[] = [];
-    let totalBytes = 0;
-
-    while (true) {
-      const result = await reader.read();
-      if (result.done) {
-        const bytes = new Uint8Array(totalBytes);
-        let offset = 0;
-        for (const chunk of chunks) {
-          bytes.set(chunk, offset);
-          offset += chunk.byteLength;
-        }
-        return { ok: true, bytes };
-      }
-
-      totalBytes += result.value.byteLength;
-      if (totalBytes > declaredBytes) {
-        await cancelReader(reader);
-        return { ok: false, reason: 'too_large' };
-      }
-      chunks.push(result.value);
-    }
-  } finally {
-    reader.releaseLock();
-  }
-}
-
-async function cancelReader(reader: ReadableStreamDefaultReader<Uint8Array>): Promise<void> {
-  try {
-    await reader.cancel('stream exceeded ingest byte limit');
-  } catch {
-    // The typed overflow result remains authoritative if cancellation itself fails.
-  }
-}
+export {
+  readBoundedStream,
+  type BoundedStreamReadResult,
+} from '@kilocode/worker-utils/bounded-stream-reader';