From 47ab8d416174a24b38fce525e1f5d1b18bc3f1ff Mon Sep 17 00:00:00 2001 From: maphew <486200+maphew@users.noreply.github.com> Date: Sat, 5 Sep 2026 06:07:37 +0000 Subject: [PATCH 1/2] fix(cloud-agent-next): retry attachment downloads and hide raw stream errors Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com> --- .../wrapper/src/session-bootstrap.test.ts | 118 +++++++++++- .../wrapper/src/session-bootstrap.ts | 179 +++++++++++++----- 2 files changed, 251 insertions(+), 46 deletions(-) diff --git a/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts b/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts index 11801d4119..ebc9e813c7 100644 --- a/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts +++ b/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts @@ -3067,7 +3067,7 @@ describe('prepareWrapperBootstrapWorkspace', () => { { type: 'text', text: 'Process this file' }, { type: 'text', - text: 'attachment interrupted.bin could not be retrieved (stream reset)', + text: 'attachment interrupted.bin could not be retrieved (download failed after 3 attempts)', }, ]); expect(fs.existsSync(localPath)).toBe(false); @@ -3131,7 +3131,7 @@ describe('prepareWrapperBootstrapWorkspace', () => { expect(await fsp.readFile(okPath, 'utf8')).toBe('png-bytes'); }); - it('converts a network/timeout failure into an explanatory text part and continues', async () => { + it('converts an exhausted retry into an explanatory text part and continues', async () => { const okPath = path.join(tmpDir, 'ok.json'); const failPath = path.join(tmpDir, 'bad.json'); const prompt: WrapperPromptRequest = { @@ -3162,21 +3162,24 @@ describe('prepareWrapperBootstrapWorkspace', () => { }, }; + let badRequests = 0; const result = await materializePromptAttachments(prompt, { fetch: asFetch(async input => { const url = typeof input === 'string' ? input : (input as Request).url; if (url === 'https://r2.example.com/bad.json') { + badRequests += 1; throw new Error('socket hang up'); } return new Response('{"ok":true}', { status: 200 }); }), }); + expect(badRequests).toBe(3); expect(result.message.parts).toEqual([ { type: 'text', text: 'Read both' }, { type: 'text', - text: 'attachment bad.json could not be retrieved (socket hang up)', + text: 'attachment bad.json could not be retrieved (download failed after 3 attempts)', }, { type: 'file', @@ -3189,6 +3192,115 @@ describe('prepareWrapperBootstrapWorkspace', () => { expect(await fsp.readFile(okPath, 'utf8')).toBe('{"ok":true}'); }); + it('retries a transient fetch failure and materializes the attachment on the retry', async () => { + const localPath = path.join(tmpDir, 'retry.png'); + const prompt: WrapperPromptRequest = { + message: { + id: 'msg_retry', + prompt: 'Look at this image', + attachments: [ + { + filename: 'retry.png', + mime: 'image/png', + signedUrl: 'https://r2.example.com/retry.png', + localPath, + }, + ], + }, + session: { + ingestUrl: 'wss://worker.example.com/sessions/user/agent/ingest', + workerAuthToken: 'token', + wrapperRunId: 'wr_test', + wrapperGeneration: 1, + wrapperConnectionId: 'conn_test', + }, + }; + + let attempts = 0; + const result = await materializePromptAttachments(prompt, { + fetch: asFetch(async () => { + attempts += 1; + if (attempts === 1) { + throw new Error('socket hang up'); + } + return new Response('png-bytes', { + status: 200, + headers: { 'content-length': '9' }, + }); + }), + }); + + expect(attempts).toBe(2); + expect(result.message.parts).toEqual([ + { type: 'text', text: 'Look at this image' }, + { + type: 'file', + mime: 'image/png', + url: `file://${localPath}`, + filename: 'retry.png', + }, + ]); + expect(await fsp.readFile(localPath, 'utf8')).toBe('png-bytes'); + }); + + it('falls back to a buffered read when the streaming read fails mid-transfer', async () => { + const localPath = path.join(tmpDir, 'fallback.png'); + const prompt: WrapperPromptRequest = { + message: { + id: 'msg_fallback', + prompt: 'Look at this image', + attachments: [ + { + filename: 'fallback.png', + mime: 'image/png', + signedUrl: 'https://r2.example.com/fallback.png', + localPath, + }, + ], + }, + session: { + ingestUrl: 'wss://worker.example.com/sessions/user/agent/ingest', + workerAuthToken: 'token', + wrapperRunId: 'wr_test', + wrapperGeneration: 1, + wrapperConnectionId: 'conn_test', + }, + }; + + let attempts = 0; + const result = await materializePromptAttachments(prompt, { + fetch: asFetch(async () => { + attempts += 1; + if (attempts === 1) { + const chunk = new Uint8Array(64 * 1024); + const body = new ReadableStream({ + pull(controller) { + controller.enqueue(chunk); + controller.error(new Error('stream reset')); + }, + }); + return new Response(body, { status: 200 }); + } + return new Response('png-bytes', { + status: 200, + headers: { 'content-length': '9' }, + }); + }), + }); + + expect(attempts).toBe(2); + expect(result.message.parts).toEqual([ + { type: 'text', text: 'Look at this image' }, + { + type: 'file', + mime: 'image/png', + url: `file://${localPath}`, + filename: 'fallback.png', + }, + ]); + expect(await fsp.readFile(localPath, 'utf8')).toBe('png-bytes'); + }); + it('materializes a generic binary attachment as a text part describing the saved file', async () => { const localPath = path.join(tmpDir, 'payload.zip'); const prompt: WrapperPromptRequest = { diff --git a/services/cloud-agent-next/wrapper/src/session-bootstrap.ts b/services/cloud-agent-next/wrapper/src/session-bootstrap.ts index 7efd837940..491bc325a3 100644 --- a/services/cloud-agent-next/wrapper/src/session-bootstrap.ts +++ b/services/cloud-agent-next/wrapper/src/session-bootstrap.ts @@ -42,6 +42,11 @@ const SETUP_COMMAND_DIAGNOSTIC_MAX_BYTES = 1_024; const GIT_BOOTSTRAP_MARKER = 'kilo-bootstrap-complete'; const MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024; const MAX_ATTACHMENT_DOWNLOAD_BYTES = MAX_ATTACHMENT_BYTES + 1; +const ATTACHMENT_DOWNLOAD_TIMEOUT_MS = 120_000; +const MAX_ATTACHMENT_DOWNLOAD_ATTEMPTS = 3; +// Backoff before each retry attempt, keyed by attempt number: attempt 2 waits +// 250ms and attempt 3 waits 750ms. +const ATTACHMENT_RETRY_BACKOFF_MS: Record = { 1: 250, 2: 750 }; function cleanTerminalOutput(text: string): string { return stripAnsi(text) @@ -1023,6 +1028,17 @@ async function safeUnlink(filePath: string): Promise { } } +/** + * Permanent per-attachment failure: the body exceeded the size cap. Unlike + * transient network/stream errors it must never be retried. + */ +class AttachmentTooLargeError extends Error { + constructor(message: string) { + super(message); + this.name = 'AttachmentTooLargeError'; + } +} + /** * Bounded streaming read. We never trust the server's `content-length` header * alone: the response body is pulled at most `MAX_ATTACHMENT_DOWNLOAD_BYTES` @@ -1083,7 +1099,7 @@ async function downloadBounded( if (overflowed) { await safeUnlink(filePath); - throw new Error( + throw new AttachmentTooLargeError( `Attachment too large: bytes exceeded the ${MAX_ATTACHMENT_BYTES / (1024 * 1024)} MiB cap` ); } @@ -1093,18 +1109,72 @@ async function downloadBounded( export type DownloadResult = | { kind: 'ok'; part: WrapperPromptPart; bytesWritten: number } - | { kind: 'failed'; part: WrapperPromptPart }; + | { kind: 'failed'; message: string; retryable: boolean }; + +type AttachmentReadStrategy = 'stream' | 'buffer'; + +async function sleep(ms: number): Promise { + await new Promise(resolve => setTimeout(resolve, ms)); +} + +/** + * Buffered read used when retrying a failed attachment download. The whole + * body is pulled with `response.arrayBuffer()`, which bypasses the Web + * Streams reader that intermittently fails inside the sandbox Bun runtime + * (`TypeError: undefined is not a function`) when the body arrives while the + * stream is being read. Only used when `content-length` is present and within + * the cap so memory stays bounded; otherwise the bounded streaming read is + * used instead. + */ +async function downloadBuffered( + filePath: string, + response: Response, + signal: AbortSignal +): Promise<{ bytesWritten: number }> { + signal.throwIfAborted(); + const contentLengthHeader = response.headers.get('content-length'); + const contentLength = contentLengthHeader === null ? undefined : Number(contentLengthHeader); + if ( + contentLength === undefined || + Number.isNaN(contentLength) || + contentLength > MAX_ATTACHMENT_BYTES + ) { + throw new Error('Attachment download failed: unbounded body'); + } + + const bytes = new Uint8Array(await response.arrayBuffer()); + signal.throwIfAborted(); + if (bytes.byteLength > MAX_ATTACHMENT_BYTES) { + throw new AttachmentTooLargeError( + `Attachment too large: bytes exceeded the ${MAX_ATTACHMENT_BYTES / (1024 * 1024)} MiB cap` + ); + } + + const handle = await fs.open( + filePath, + fs.constants.O_WRONLY | fs.constants.O_CREAT | fs.constants.O_TRUNC | fs.constants.O_NOFOLLOW, + 0o600 + ); + try { + await handle.write(bytes); + } finally { + await handle.close(); + } + return { bytesWritten: bytes.byteLength }; +} /** * Download a single attachment. Per-file failure (non-2xx response, - * read/timeout error, overflow) is converted to an explanatory text part so - * the rest of the prompt can still proceed; the whole-message abort path is + * read/timeout error, overflow) is converted to a structured failure so the + * caller can retry transient errors or surface an explanatory text part and + * let the rest of the prompt proceed; the whole-message abort path is * reserved for non-attachment failures. */ async function downloadAndMaterializeAttachment( attachment: WrapperBootstrapAttachment, fetchImpl: typeof fetch, - signal: AbortSignal + signal: AbortSignal, + strategy: AttachmentReadStrategy = 'stream' ): Promise { signal.throwIfAborted(); await fs.mkdir(path.dirname(attachment.localPath), { recursive: true }); @@ -1115,39 +1185,26 @@ async function downloadAndMaterializeAttachment( response = await fetchImpl(attachment.signedUrl, { signal }); } catch (error) { const message = redactSecrets(error instanceof Error ? error.message : String(error)); - return { - kind: 'failed', - part: { - type: 'text', - text: `attachment ${attachment.filename} could not be retrieved (${message})`, - }, - }; + return { kind: 'failed', message, retryable: true }; } if (!response.ok) { void response.body?.cancel().catch(() => {}); - return { - kind: 'failed', - part: { - type: 'text', - text: `attachment ${attachment.filename} could not be retrieved (HTTP ${response.status})`, - }, - }; + const retryable = response.status === 429 || response.status >= 500; + return { kind: 'failed', message: `HTTP ${response.status}`, retryable }; } let result: { bytesWritten: number }; try { - result = await downloadBounded(attachment.localPath, response, signal); + result = + strategy === 'buffer' + ? await downloadBuffered(attachment.localPath, response, signal) + : await downloadBounded(attachment.localPath, response, signal); } catch (error) { void response.body?.cancel().catch(() => {}); const message = redactSecrets(error instanceof Error ? error.message : String(error)); - return { - kind: 'failed', - part: { - type: 'text', - text: `attachment ${attachment.filename} could not be retrieved (${message})`, - }, - }; + const retryable = !(error instanceof AttachmentTooLargeError); + return { kind: 'failed', message, retryable }; } if (isPromptFileMime(attachment.mime)) { @@ -1176,6 +1233,57 @@ async function downloadAndMaterializeAttachment( }; } +/** + * Download and materialize a single attachment, retrying transient failures. + * The first attempt uses the bounded streaming read; later attempts use the + * buffered read, which re-fetches and bypasses the flaky Web Streams reader. + * Returns the prompt part: a `file://` part on success, or an explanatory + * text part when retries are exhausted or the failure is permanent. + */ +async function materializeAttachment( + attachment: WrapperBootstrapAttachment, + fetchImpl: typeof fetch, + externalSignal?: AbortSignal +): Promise { + for (let attempt = 1; attempt <= MAX_ATTACHMENT_DOWNLOAD_ATTEMPTS; attempt++) { + const abortController = new AbortController(); + const timeout = setTimeout( + () => abortController.abort(new Error('attachment download timeout')), + ATTACHMENT_DOWNLOAD_TIMEOUT_MS + ); + const signal = externalSignal + ? AbortSignal.any([abortController.signal, externalSignal]) + : abortController.signal; + try { + const result = await downloadAndMaterializeAttachment( + attachment, + fetchImpl, + signal, + attempt === 1 ? 'stream' : 'buffer' + ); + if (result.kind === 'ok') return result.part; + if (!result.retryable) { + return { + type: 'text', + text: `attachment ${attachment.filename} could not be retrieved (${result.message})`, + }; + } + logToFile( + `attachment download attempt ${attempt}/${MAX_ATTACHMENT_DOWNLOAD_ATTEMPTS} failed filename=${attachment.filename} reason=${result.message}` + ); + } finally { + clearTimeout(timeout); + } + if (attempt < MAX_ATTACHMENT_DOWNLOAD_ATTEMPTS) { + await sleep(ATTACHMENT_RETRY_BACKOFF_MS[attempt] ?? 0); + } + } + return { + type: 'text', + text: `attachment ${attachment.filename} could not be retrieved (download failed after ${MAX_ATTACHMENT_DOWNLOAD_ATTEMPTS} attempts)`, + }; +} + export type MaterializeDeps = { fetch?: typeof fetch; signal?: AbortSignal; @@ -1192,22 +1300,7 @@ export async function materializeMessageAttachments( const parts: WrapperPromptPart[] = []; for (const attachment of message.attachments) { deps.signal?.throwIfAborted(); - const abortController = new AbortController(); - const timeout = setTimeout( - () => abortController.abort(new Error('attachment download timeout')), - 120_000 - ); - const signal = deps.signal - ? AbortSignal.any([abortController.signal, deps.signal]) - : abortController.signal; - let result: DownloadResult; - try { - result = await downloadAndMaterializeAttachment(attachment, fetchImpl, signal); - deps.signal?.throwIfAborted(); - } finally { - clearTimeout(timeout); - } - parts.push(result.part); + parts.push(await materializeAttachment(attachment, fetchImpl, deps.signal)); } return { From 27b4ed9b9dce77a770dc4bb8cd413858b4769bdc Mon Sep 17 00:00:00 2001 From: maphew <486200+maphew@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:46:33 +0000 Subject: [PATCH 2/2] fix(cloud-agent-next): fall back to bounded reads and honor aborts in attachment retries --- .../wrapper/src/session-bootstrap.test.ts | 96 +++++++++++++++++++ .../wrapper/src/session-bootstrap.ts | 3 +- 2 files changed, 98 insertions(+), 1 deletion(-) diff --git a/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts b/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts index ebc9e813c7..9c623f8abd 100644 --- a/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts +++ b/services/cloud-agent-next/wrapper/src/session-bootstrap.test.ts @@ -3301,6 +3301,102 @@ describe('prepareWrapperBootstrapWorkspace', () => { expect(await fsp.readFile(localPath, 'utf8')).toBe('png-bytes'); }); + it('falls back to the bounded streaming read when a buffered retry has no content-length', async () => { + const localPath = path.join(tmpDir, 'no-content-length.png'); + const prompt: WrapperPromptRequest = { + message: { + id: 'msg_no_content_length', + prompt: 'Look at this image', + attachments: [ + { + filename: 'no-content-length.png', + mime: 'image/png', + signedUrl: 'https://r2.example.com/no-content-length.png', + localPath, + }, + ], + }, + session: { + ingestUrl: 'wss://worker.example.com/sessions/user/agent/ingest', + workerAuthToken: 'token', + wrapperRunId: 'wr_test', + wrapperGeneration: 1, + wrapperConnectionId: 'conn_test', + }, + }; + + let attempts = 0; + const result = await materializePromptAttachments(prompt, { + fetch: asFetch(async () => { + attempts += 1; + if (attempts === 1) { + throw new Error('socket hang up'); + } + const response = new Response('png-bytes', { status: 200 }); + response.headers.delete('content-length'); + return response; + }), + }); + + expect(attempts).toBe(2); + expect(result.message.parts).toEqual([ + { type: 'text', text: 'Look at this image' }, + { + type: 'file', + mime: 'image/png', + url: `file://${localPath}`, + filename: 'no-content-length.png', + }, + ]); + expect(await fsp.readFile(localPath, 'utf8')).toBe('png-bytes'); + }); + + it('propagates a caller abort instead of retrying or masking it as an exhausted attachment', async () => { + const localPath = path.join(tmpDir, 'aborted.png'); + const prompt: WrapperPromptRequest = { + message: { + id: 'msg_aborted', + prompt: 'Look at this image', + attachments: [ + { + filename: 'aborted.png', + mime: 'image/png', + signedUrl: 'https://r2.example.com/aborted.png', + localPath, + }, + ], + }, + session: { + ingestUrl: 'wss://worker.example.com/sessions/user/agent/ingest', + workerAuthToken: 'token', + wrapperRunId: 'wr_test', + wrapperGeneration: 1, + wrapperConnectionId: 'conn_test', + }, + }; + + const controller = new AbortController(); + let attempts = 0; + const result = materializePromptAttachments(prompt, { + fetch: asFetch(async () => { + attempts += 1; + if (attempts < 3) { + throw new Error('socket hang up'); + } + controller.abort(); + return new Response('png-bytes', { + status: 200, + headers: { 'content-length': '9' }, + }); + }), + signal: controller.signal, + }); + + await expect(result).rejects.toThrow(); + expect(attempts).toBe(3); + expect(fs.existsSync(localPath)).toBe(false); + }); + it('materializes a generic binary attachment as a text part describing the saved file', async () => { const localPath = path.join(tmpDir, 'payload.zip'); const prompt: WrapperPromptRequest = { diff --git a/services/cloud-agent-next/wrapper/src/session-bootstrap.ts b/services/cloud-agent-next/wrapper/src/session-bootstrap.ts index 491bc325a3..281b3e2c92 100644 --- a/services/cloud-agent-next/wrapper/src/session-bootstrap.ts +++ b/services/cloud-agent-next/wrapper/src/session-bootstrap.ts @@ -1139,7 +1139,7 @@ async function downloadBuffered( Number.isNaN(contentLength) || contentLength > MAX_ATTACHMENT_BYTES ) { - throw new Error('Attachment download failed: unbounded body'); + return downloadBounded(filePath, response, signal); } const bytes = new Uint8Array(await response.arrayBuffer()); @@ -1274,6 +1274,7 @@ async function materializeAttachment( } finally { clearTimeout(timeout); } + externalSignal?.throwIfAborted(); if (attempt < MAX_ATTACHMENT_DOWNLOAD_ATTEMPTS) { await sleep(ATTACHMENT_RETRY_BACKOFF_MS[attempt] ?? 0); }