5362b4a796
Attachments copied into a local computer's workspace were read whole from the artifact store first, so a 500 MiB video sat in memory before it was written to disk. attach_file did the same in the other direction: a large file on a local computer was read whole before it was stored. SandboxProvider gains optional readFileStream/writeFileStream. The desktop (host workspace) provider implements them with the same path confinement as its byte methods, one chunk in memory at a time, and removes a file whose write fails or is aborted, including one that replaced an earlier copy. The host-aware wrapper routes them, and providers without them (remote computers, still capped at 12 MiB) receive the collected bytes. Current-turn staging and attach_file now go through these helpers, and unread streams are cancelled so the stored file is released. Media job inputs stop copying when the job is cancelled. The stream helpers move from the local artifact store into a shared module so both use one implementation. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u
92 lines
2.8 KiB
TypeScript
92 lines
2.8 KiB
TypeScript
import type { FileHandle } from "node:fs/promises";
|
|
|
|
type ByteSource = ReadableStream<Uint8Array> | AsyncIterable<Uint8Array>;
|
|
|
|
const READ_CHUNK_BYTES = 256 * 1024;
|
|
|
|
export async function* byteChunks(source: ByteSource): AsyncIterable<Uint8Array> {
|
|
if (!("getReader" in source)) {
|
|
yield* source;
|
|
return;
|
|
}
|
|
const reader = source.getReader();
|
|
try {
|
|
for (;;) {
|
|
const { done, value } = await reader.read();
|
|
if (done) return;
|
|
yield value;
|
|
}
|
|
} finally {
|
|
// Stop the producer (e.g. an upload body) when the consumer gives up early, without waiting
|
|
// on a producer that may never confirm.
|
|
void reader.cancel().catch(() => undefined);
|
|
reader.releaseLock();
|
|
}
|
|
}
|
|
|
|
/** Releases a stream nobody will read, so its producer (e.g. an open file) is closed. */
|
|
export function discardByteStream(source: ByteSource): void {
|
|
if ("cancel" in source) void source.cancel().catch(() => undefined);
|
|
}
|
|
|
|
/** Collects a stream into one buffer, for destinations that only accept bytes. */
|
|
export async function collectBytes(source: ByteSource): Promise<Uint8Array> {
|
|
const chunks: Uint8Array[] = [];
|
|
let size = 0;
|
|
for await (const chunk of byteChunks(source)) {
|
|
chunks.push(chunk);
|
|
size += chunk.byteLength;
|
|
}
|
|
const bytes = new Uint8Array(size);
|
|
let offset = 0;
|
|
for (const chunk of chunks) {
|
|
bytes.set(chunk, offset);
|
|
offset += chunk.byteLength;
|
|
}
|
|
return bytes;
|
|
}
|
|
|
|
export async function writeFully(handle: FileHandle, chunk: Uint8Array) {
|
|
let offset = 0;
|
|
while (offset < chunk.byteLength) {
|
|
const { bytesWritten } = await handle.write(chunk, offset, chunk.byteLength - offset);
|
|
offset += bytesWritten;
|
|
}
|
|
}
|
|
|
|
/** Streams [start, end] of an open file and closes it when done, failed or cancelled. */
|
|
export function fileByteStream(handle: FileHandle, start: number, end: number) {
|
|
let position = start;
|
|
let closed = false;
|
|
const close = async () => {
|
|
if (closed) return;
|
|
closed = true;
|
|
await handle.close().catch(() => undefined);
|
|
};
|
|
return new ReadableStream<Uint8Array>({
|
|
async pull(controller) {
|
|
try {
|
|
const length = Math.min(end - position + 1, READ_CHUNK_BYTES);
|
|
if (length <= 0) {
|
|
await close();
|
|
controller.close();
|
|
return;
|
|
}
|
|
const buffer = new Uint8Array(length);
|
|
const { bytesRead } = await handle.read(buffer, 0, length, position);
|
|
if (bytesRead === 0) throw new Error("File ended before the requested range");
|
|
position += bytesRead;
|
|
controller.enqueue(bytesRead === length ? buffer : buffer.subarray(0, bytesRead));
|
|
if (position > end) {
|
|
await close();
|
|
controller.close();
|
|
}
|
|
} catch (error) {
|
|
await close();
|
|
controller.error(error);
|
|
}
|
|
},
|
|
cancel: close,
|
|
});
|
|
}
|