improvement(sandbox): exempt caller-consumed streams from the output retention budget (#6353)

icecrasher321 · claude · web-flow · commit f340ad965d00 · 2026-08-06T17:32:19.000-07:00
* 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) &lt;noreply@anthropic.com&gt;

* test(sandbox): drop explicit any from the new conformance stream mocks

The two new E2B mocks annotated their arguments as `any`, which both violates
the repo's no-`any` rule and defeats the point of a mock: an invalid SDK shape
would type-check.

Matches the sibling mock a few lines above (`async (_code, options) =&gt;`) and
infers from the `vi.fn()` signature instead of naming a type, so the mock stays
bound to whatever the adapter actually calls.

Co-Authored-By: Claude Opus 5 (1M context) &lt;noreply@anthropic.com&gt;

* fix(sandbox): cut Daytona's retained tail to the same bound as E2B

`appendStreamedSandboxOutput` deliberately lets the accumulator grow to twice
the tail before collapsing, so a single re-cut is amortized across chunks rather
than paid on every one. That leaves it anywhere inside that band when the stream
ends. E2B tails the value it returns, Daytona returned the accumulator as-is, so
a stream finishing between one and two tails came back roughly 96 KB on Daytona
and 64 KB on E2B.

The two adapters must agree — a divergence here surfaces as changed behavior
during a failover, which is the one moment nobody wants surprises. Daytona now
takes the same final cut on every return path.

The conformance test that should have caught this asserted the bound as
`tail * 2`, which is satisfied by both the correct and the incorrect value. It
now asserts the tail plus the truncation note, and a second case exercises the
band between one and two tails where the two providers could disagree.

Co-Authored-By: Claude Opus 5 (1M context) &lt;noreply@anthropic.com&gt;

---------

Co-authored-by: Claude Opus 5 (1M context) &lt;noreply@anthropic.com&gt;
diff --git 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
@@ -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,17 +319,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
             )
           }
@@ -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
@@ -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
@@ -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
 
