diff --git a/apps/sim/lib/execution/remote-sandbox/conformance.test.ts b/apps/sim/lib/execution/remote-sandbox/conformance.test.ts index 2f71e789a87..d59504d28c8 100644 --- a/apps/sim/lib/execution/remote-sandbox/conformance.test.ts +++ b/apps/sim/lib/execution/remote-sandbox/conformance.test.ts @@ -119,6 +119,7 @@ import { E2B_MAX_SANDBOX_LIFETIME_MS, e2bProvider } from '@/lib/execution/remote import { MAX_SANDBOX_OUTPUT_BYTES, MAX_SANDBOX_PROCESS_OUTPUT_BYTES, + MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES, } from '@/lib/execution/remote-sandbox/output-limits' import { PI_SANDBOX_MIN_LIFETIME_MS, @@ -293,6 +294,9 @@ beforeEach(() => { mockGetSessionCommand.mockResolvedValue({ exitCode: 0 }) }) +/** Headroom for the note `tailStreamedSandboxOutput` prepends when it truncates. */ +const TRUNCATION_NOTE_ALLOWANCE_BYTES = 256 + describe.each(PROVIDERS)('sandbox conformance [%s]', (provider) => { beforeEach(() => useProvider(provider)) @@ -373,6 +377,101 @@ describe.each(PROVIDERS)('sandbox conformance [%s]', (provider) => { expect(provider === 'e2b' ? mockE2BKill : mockDelete).toHaveBeenCalledTimes(1) }) + it('exempts a caller-consumed stream from the budget and keeps a diagnostic tail', async () => { + // A Pi agent turn streams one JSONL event per step and routinely passes the retention budget + // while producing no oversized result — the caller parses every chunk and keeps none of it. + const oversized = 'x'.repeat(MAX_SANDBOX_PROCESS_OUTPUT_BYTES + 1024) + if (provider === 'e2b') { + mockE2BCommandsRun.mockImplementationOnce(async (_cmd, options) => { + options.onStdout(`${oversized}TAIL_MARKER`) + return { stdout: `${oversized}TAIL_MARKER`, stderr: '', exitCode: 0 } + }) + } else { + mockGetSessionCommandLogs.mockImplementationOnce( + async (_sessionId: string, _commandId: string, onStdout: (chunk: string) => void) => { + onStdout(`${oversized}TAIL_MARKER`) + } + ) + mockGetSessionCommand.mockResolvedValue({ exitCode: 0 }) + } + + let streamedBytes = 0 + const result = await withPiSandbox({}, (runner) => + runner.run('pi run', { + timeoutMs: 1000, + onStdout: (chunk) => { + streamedBytes += chunk.length + }, + }) + ) + + // Delivered in full to the caller... + expect(streamedBytes).toBeGreaterThan(MAX_SANDBOX_PROCESS_OUTPUT_BYTES) + expect(result.exitCode).toBe(0) + // ...but only the tail is retained. Reaching exitCode 0 at all is the point: before the + // exemption this threw `sandbox_output_limit_exceeded` and killed the sandbox mid-run. + expect(result.stdout).toContain('TAIL_MARKER') + // The tail plus its truncation note, NOT a multiple of it. A looser bound here passes on both + // providers even when one returns twice as much as the other, which is the divergence this + // pair exists to prevent. + expect(Buffer.byteLength(result.stdout)).toBeLessThanOrEqual( + MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES + TRUNCATION_NOTE_ALLOWANCE_BYTES + ) + }) + + it('cuts the retained tail identically when a stream ends between one and two tails', async () => { + // The Daytona appender only collapses once the accumulator passes twice the tail, so a stream + // finishing inside that band is the case where the two providers can disagree. + const midBand = 'y'.repeat(Math.floor(MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES * 1.5)) + if (provider === 'e2b') { + mockE2BCommandsRun.mockImplementationOnce(async (_cmd, options) => { + options.onStdout?.(midBand) + return { stdout: midBand, stderr: '', exitCode: 0 } + }) + } else { + mockGetSessionCommandLogs.mockImplementationOnce( + async (_sessionId: string, _commandId: string, onStdout: (chunk: string) => void) => { + onStdout(midBand) + } + ) + mockGetSessionCommand.mockResolvedValue({ exitCode: 0 }) + } + + const result = await withPiSandbox({}, (runner) => + runner.run('pi run', { timeoutMs: 1000, onStdout: () => {} }) + ) + + expect(Buffer.byteLength(result.stdout)).toBeLessThanOrEqual( + MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES + TRUNCATION_NOTE_ALLOWANCE_BYTES + ) + expect(Buffer.byteLength(result.stdout)).toBeLessThan(Buffer.byteLength(midBand)) + }) + + it('still bounds a stream the caller does not consume', async () => { + const oversized = 'x'.repeat(MAX_SANDBOX_PROCESS_OUTPUT_BYTES + 1) + if (provider === 'e2b') { + mockE2BCommandsRun.mockImplementationOnce(async (_cmd, options) => { + options.onStderr(oversized) + }) + } else { + mockGetSessionCommandLogs.mockImplementationOnce( + async ( + _sessionId: string, + _commandId: string, + _onStdout: (chunk: string) => void, + onStderr: (chunk: string) => void + ) => { + onStderr(oversized) + } + ) + } + + // stdout is streamed, stderr is not — the exemption is per stream, not per command. + await expect( + withPiSandbox({}, (runner) => runner.run('pi run', { timeoutMs: 1000, onStdout: () => {} })) + ).rejects.toMatchObject({ code: 'sandbox_output_limit_exceeded', outputKind: 'process' }) + }) + it('normalizes execution errors to the same shape', async () => { stubShellCommand(provider, '', 'Traceback...\nValueError: boom', 1) diff --git a/apps/sim/lib/execution/remote-sandbox/daytona.ts b/apps/sim/lib/execution/remote-sandbox/daytona.ts index 6a1b9aedbf5..9a5fce81963 100644 --- a/apps/sim/lib/execution/remote-sandbox/daytona.ts +++ b/apps/sim/lib/execution/remote-sandbox/daytona.ts @@ -13,12 +13,14 @@ import { } from '@/lib/core/utils/stream-limits' import { CodeLanguage } from '@/lib/execution/languages' import { + appendStreamedSandboxOutput, isSandboxOutputLimitError, MAX_SANDBOX_OUTPUT_BYTES, MAX_SANDBOX_PROCESS_OUTPUT_BYTES, SandboxOutputFileError, SandboxOutputLimitError, SandboxProcessOutputBudget, + tailStreamedSandboxOutput, } from '@/lib/execution/remote-sandbox/output-limits' import type { CreateSandboxOptions, @@ -246,6 +248,18 @@ class DaytonaSandboxHandle implements SandboxHandle { const outputBudget = new SandboxProcessOutputBudget( options.maxOutputBytes ?? MAX_SANDBOX_PROCESS_OUTPUT_BYTES ) + // Matches the E2B adapter: the budget bounds what Sim retains, so a stream the caller consumes + // itself is exempt and only a diagnostic tail is kept. Per stream, so a caller that streams + // stdout but not stderr still has stderr fully bounded. The failover must not change behavior. + const retainStdout = options.onStdout === undefined + const retainStderr = options.onStderr === undefined + // The appender keeps the accumulator under twice the tail so it is not re-cut on every chunk, + // which leaves it anywhere in that band when the stream ends. E2B tails the value it returns, + // so the final cut has to happen here too or a stream finishing between one and two tails comes + // back longer on Daytona than on E2B — a failover divergence, which is what this adapter pair + // must never have. + const finalStdout = () => (retainStdout ? stdout : tailStreamedSandboxOutput(stdout)) + const finalStderr = () => (retainStderr ? stderr : tailStreamedSandboxOutput(stderr)) try { await this.sandbox.process.createSession(sessionId) sessionCreated = true @@ -282,14 +296,17 @@ class DaytonaSandboxHandle implements SandboxHandle { const appendOutput = ( chunk: string, append: (value: string) => void, + retain: boolean, callback?: (value: string) => void ) => { - try { - outputBudget.add(chunk) - } catch { - void this.kill().catch(() => {}) - resolveOutputLimit() - return + if (retain) { + try { + outputBudget.add(chunk) + } catch { + void this.kill().catch(() => {}) + resolveOutputLimit() + return + } } append(chunk) callback?.(chunk) @@ -302,8 +319,9 @@ class DaytonaSandboxHandle implements SandboxHandle { appendOutput( chunk, (value) => { - stdout += value + stdout = retainStdout ? stdout + value : appendStreamedSandboxOutput(stdout, value) }, + retainStdout, options.onStdout ) }, @@ -311,8 +329,9 @@ class DaytonaSandboxHandle implements SandboxHandle { appendOutput( chunk, (value) => { - stderr += value + stderr = retainStderr ? stderr + value : appendStreamedSandboxOutput(stderr, value) }, + retainStderr, options.onStderr ) } @@ -355,8 +374,8 @@ class DaytonaSandboxHandle implements SandboxHandle { if (outcome === 'output-limit' || outputBudget.error) throw outputBudget.error if (outcome === 'timeout') { return { - stdout, - stderr: stderr || `Command timed out after ${options.timeoutMs}ms`, + stdout: finalStdout(), + stderr: finalStderr() || `Command timed out after ${options.timeoutMs}ms`, exitCode: 124, timedOut: true, } @@ -365,12 +384,17 @@ class DaytonaSandboxHandle implements SandboxHandle { if (outputBudget.error) throw outputBudget.error const timedOut = isDaytonaExecutionTimeout(streamError) if (!timedOut) throw streamError - return { stdout, stderr: stderr || getErrorMessage(streamError), exitCode: 124, timedOut } + return { + stdout: finalStdout(), + stderr: finalStderr() || getErrorMessage(streamError), + exitCode: 124, + timedOut, + } } const finished = await this.sandbox.process.getSessionCommand(sessionId, commandId) const exitCode = finished.exitCode ?? 0 - return { stdout, stderr, exitCode } + return { stdout: finalStdout(), stderr: finalStderr(), exitCode } } catch (error) { if (isSandboxOutputLimitError(error)) { void this.kill().catch(() => {}) @@ -382,10 +406,15 @@ class DaytonaSandboxHandle implements SandboxHandle { : new DOMException('Execution cancelled', 'AbortError') } if (isDaytonaExecutionTimeout(error)) { - return { stdout, stderr: stderr || getErrorMessage(error), exitCode: 124, timedOut: true } + return { + stdout: finalStdout(), + stderr: finalStderr() || getErrorMessage(error), + exitCode: 124, + timedOut: true, + } } if (operation === 'code') throw error - return { stdout, stderr: stderr || getErrorMessage(error), exitCode: 1 } + return { stdout: finalStdout(), stderr: finalStderr() || getErrorMessage(error), exitCode: 1 } } finally { if (sessionCreated) { try { diff --git a/apps/sim/lib/execution/remote-sandbox/e2b.ts b/apps/sim/lib/execution/remote-sandbox/e2b.ts index e0fd61802e2..c7adace21d8 100644 --- a/apps/sim/lib/execution/remote-sandbox/e2b.ts +++ b/apps/sim/lib/execution/remote-sandbox/e2b.ts @@ -36,6 +36,7 @@ import { SandboxOutputFileError, SandboxOutputLimitError, SandboxProcessOutputBudget, + tailStreamedSandboxOutput, } from '@/lib/execution/remote-sandbox/output-limits' import { quoteDependency, @@ -389,12 +390,19 @@ class E2BSandboxHandle implements SandboxHandle { const outputBudget = new SandboxProcessOutputBudget( options.maxOutputBytes ?? MAX_SANDBOX_PROCESS_OUTPUT_BYTES ) - const guardOutput = (value: string, callback?: (chunk: string) => void) => { - try { - outputBudget.add(value) - } catch (error) { - void this.kill().catch(() => {}) - throw error + // The budget bounds what Sim RETAINS, so a stream the caller consumes itself is exempt: it has + // already been delivered chunk by chunk, and only a diagnostic tail is kept. Per stream, not + // per command — a caller that streams stdout but not stderr still has stderr fully bounded. + const retainStdout = options.onStdout === undefined + const retainStderr = options.onStderr === undefined + const guardOutput = (value: string, retain: boolean, callback?: (chunk: string) => void) => { + if (retain) { + try { + outputBudget.add(value) + } catch (error) { + void this.kill().catch(() => {}) + throw error + } } callback?.(value) } @@ -405,11 +413,18 @@ class E2BSandboxHandle implements SandboxHandle { timeoutMs: e2bTimeoutMs(options.timeoutMs), ...(options.signal ? { signal: options.signal } : {}), ...(options.rootUser ? { user: 'root' as const } : {}), - onStdout: (chunk) => guardOutput(chunk, options.onStdout), - onStderr: (chunk) => guardOutput(chunk, options.onStderr), + onStdout: (chunk) => guardOutput(chunk, retainStdout, options.onStdout), + onStderr: (chunk) => guardOutput(chunk, retainStderr, options.onStderr), }) - assertSandboxProcessOutputWithinLimit([result.stdout, result.stderr], options.maxOutputBytes) - return { stdout: result.stdout, stderr: result.stderr, exitCode: result.exitCode } + assertSandboxProcessOutputWithinLimit( + [retainStdout ? result.stdout : undefined, retainStderr ? result.stderr : undefined], + options.maxOutputBytes + ) + return { + stdout: retainStdout ? result.stdout : tailStreamedSandboxOutput(result.stdout), + stderr: retainStderr ? result.stderr : tailStreamedSandboxOutput(result.stderr), + exitCode: result.exitCode, + } } catch (error) { if (outputBudget.error) throw outputBudget.error if (reachedE2BProviderLimit(error, this.providerLimitAtMs, options.signal)) { @@ -432,21 +447,33 @@ class E2BSandboxHandle implements SandboxHandle { ) { throw error } + // The SDK throws on a non-zero exit, so this is the ordinary path for a failing streamed + // command — the same retention exemption has to apply here or a failing Pi turn still trips + // the budget on output the caller already consumed. `message` never streams, so it is always + // retained and billed. assertSandboxProcessOutputWithinLimit( - [failure.stdout, failure.stderr, failure.message], + [ + retainStdout ? failure.stdout : undefined, + retainStderr ? failure.stderr : undefined, + failure.message, + ], options.maxOutputBytes ) + const tailIfStreamed = (value: string | undefined, retain: boolean) => + retain || value === undefined ? value : tailStreamedSandboxOutput(value) + const failureStdout = tailIfStreamed(failure.stdout, retainStdout) + const failureStderr = tailIfStreamed(failure.stderr, retainStderr) if (isE2BExecutionTimeout(error)) { return { - stdout: failure.stdout ?? '', - stderr: failure.stderr ?? failure.message ?? '', + stdout: failureStdout ?? '', + stderr: failureStderr ?? failure.message ?? '', exitCode: 124, timedOut: true, } } return { - stdout: failure.stdout ?? '', - stderr: failure.stderr ?? failure.message ?? getErrorMessage(error), + stdout: failureStdout ?? '', + stderr: failureStderr ?? failure.message ?? getErrorMessage(error), exitCode: failure.exitCode ?? 1, } } diff --git a/apps/sim/lib/execution/remote-sandbox/output-limits.ts b/apps/sim/lib/execution/remote-sandbox/output-limits.ts index d58d0464e36..a273822d90d 100644 --- a/apps/sim/lib/execution/remote-sandbox/output-limits.ts +++ b/apps/sim/lib/execution/remote-sandbox/output-limits.ts @@ -7,6 +7,51 @@ export const MAX_SANDBOX_OUTPUT_BYTES = 50 * 1024 * 1024 */ export const MAX_SANDBOX_PROCESS_OUTPUT_BYTES = 10 * 1024 * 1024 +/** + * Diagnostic tail kept from a stream the caller consumed itself. + * + * A caller that passes `onStdout`/`onStderr` takes delivery of every chunk as it arrives, so the + * adapter's accumulated copy is never the result — it is only ever read back to explain a failure. + * Billing that copy to the retention budget kills runs whose live stream is legitimately long while + * producing no oversized result: a Pi agent turn emits one JSONL event per step and passes 10 MB on + * an ordinary session, even though the caller has already parsed every event and keeps none of it. + */ +export const MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES = 64 * 1024 + +const STREAMED_OUTPUT_TRUNCATION_NOTE = + '[earlier output truncated — it was streamed to the caller]\n' + +/** + * Keeps the last {@link MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES} of a streamed output. The cut is + * advanced past any UTF-8 continuation bytes so the tail starts on a code-point boundary rather + * than decoding to replacement characters. + */ +export function tailStreamedSandboxOutput( + value: string | undefined, + limitBytes = MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES +): string { + if (!value) return '' + const buffer = Buffer.from(value, 'utf8') + if (buffer.length <= limitBytes) return value + + let start = buffer.length - limitBytes + while (start < buffer.length && (buffer[start] & 0xc0) === 0x80) start += 1 + return `${STREAMED_OUTPUT_TRUNCATION_NOTE}${buffer.subarray(start).toString('utf8')}` +} + +/** + * Appends to a streamed-output accumulator, collapsing it back to the diagnostic tail once it grows + * past twice that tail. Truncating on every chunk would be quadratic over a long stream. The + * threshold compares UTF-16 length rather than bytes because it only decides *when* to collapse — + * {@link tailStreamedSandboxOutput} does the byte-exact cut. + */ +export function appendStreamedSandboxOutput(current: string, chunk: string): string { + const next = current + chunk + return next.length > MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES * 2 + ? tailStreamedSandboxOutput(next) + : next +} + export const SANDBOX_OUTPUT_LIMIT_CODE = 'sandbox_output_limit_exceeded' as const export const SANDBOX_OUTPUT_FILE_INVALID_CODE = 'sandbox_output_file_invalid' as const