fix(sandbox): exempt caller-consumed streams from the output retention budget

A Pi agent turn emits one JSONL event per step and passes the 10 MB process
output budget on an ordinary session, killing the run. The bytes were never a
result: `handleChunk` parses every chunk as it arrives and keeps none of it,
and the accumulated copy is only ever read back to build an error message.

The budget bounds what Sim RETAINS, so a stream the caller consumes itself is
exempt and only a 64 KB diagnostic tail is kept. The limit is unchanged for
everything else.

Gated per stream, not per command: a caller that streams stdout but not stderr
still has stderr fully bounded. Both adapters gate on the handler's presence, so
the calls that parse markers out of stdout (Pi's clone/prepare/push, which do
not stream) keep full retention and full budgeting — the case daytona.ts already
warns about.

E2B's SDK still accumulates internally, so this bounds what Sim retains rather
than the provider's peak; Daytona accumulates locally and is bounded outright.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: icecrasher321

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,
@@ -373,6 +374,70 @@ 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: string, options: any) => {
+        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')
+    expect(Buffer.byteLength(result.stdout)).toBeLessThanOrEqual(
+      MAX_SANDBOX_STREAMED_OUTPUT_TAIL_BYTES * 2
+    )
+  })
+
+  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: string, options: any) => {
+        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)
 

apps/sim/lib/execution/remote-sandbox/daytona.ts

@@ -13,6 +13,7 @@ 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,
@@ -246,6 +247,11 @@ 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
     try {
       await this.sandbox.process.createSession(sessionId)
       sessionCreated = true
@@ -282,14 +288,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,17 +311,19 @@ class DaytonaSandboxHandle implements SandboxHandle {
             appendOutput(
               chunk,
               (value) => {
-                stdout += value
+                stdout = retainStdout ? stdout + value : appendStreamedSandboxOutput(stdout, value)
               },
+              retainStdout,
               options.onStdout
             )
           },
           (chunk: string) => {
             appendOutput(
               chunk,
               (value) => {
-                stderr += value
+                stderr = retainStderr ? stderr + value : appendStreamedSandboxOutput(stderr, value)
               },
+              retainStderr,
               options.onStderr
             )
           }

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,
       }
     }

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