fix(pi): compact cloud event streams before sandbox retention (#6464)

* fix(pi): compact cloud event streams

* fix(pi): select bash for event pipeline

* address comments

---------

Co-authored-by: Bill Leoutsakos <billleoutsakos@Bills-MacBook-Pro.local>
Co-authored-by: Vikhyath Mondreti <vikhyath@simstudio.ai>

apps/sim/executor/handlers/pi/cloud/authoring/backend.test.ts

@@ -284,13 +284,24 @@ describe('runCloudPi', () => {
 
     // Untrusted text is written through the sandbox FS API, not interpolated into a shell command.
     expect(mockWriteFile).toHaveBeenCalledWith('/workspace/pi-prompt.txt', 'PROMPT')
+    expect(mockWriteFile).toHaveBeenCalledWith(
+      '/workspace/sim-pi-event-filter.mjs',
+      expect.stringContaining("case 'message_update'")
+    )
     expect(mockWriteFile).toHaveBeenCalledWith('/workspace/pi-commit.txt', 'Pi: do it')
 
     const [piCmd, piOpts] = mockRun.mock.calls[1]
     // Prompt arrives on stdin from a fixed path; never a CLI arg or env value.
     expect(piCmd).toContain('< /workspace/pi-prompt.txt')
+    expect(piCmd).toContain('| node /workspace/sim-pi-event-filter.mjs')
     expect(piCmd).not.toContain('PROMPT')
     expect(piOpts.envs.PI_TASK).toBeUndefined()
+    const filterWrite = mockWriteFile.mock.calls.findIndex(
+      ([path]: [string]) => path === '/workspace/sim-pi-event-filter.mjs'
+    )
+    expect(mockWriteFile.mock.invocationCallOrder[filterWrite]).toBeLessThan(
+      mockRun.mock.invocationCallOrder[1]
+    )
 
     const [prepareCmd, prepareOpts] = mockRun.mock.calls[2]
     // Commit message is read from a file, not passed as -m "...".

apps/sim/executor/handlers/pi/cloud/authoring/backend.ts

@@ -28,6 +28,10 @@ import {
   resolvePiSandboxLifetimeMs,
 } from '@/lib/execution/remote-sandbox/pi-lifetime'
 import { runBabysitPi } from '@/executor/handlers/pi/cloud/babysit/backend'
+import {
+  PI_EVENT_FILTER_PATH,
+  PI_EVENT_FILTER_SOURCE,
+} from '@/executor/handlers/pi/cloud/event-filter-source'
 import {
   type BranchPullRequest,
   fetchOpenPrForBranch,
@@ -466,6 +470,7 @@ async function runCloudAuthoringPi(
 
       // Outside REPO_DIR: a path inside the cloned tree would be staged by `git add -A` into the
       // user's pull request, and the agent holds write/edit/bash on that tree for the whole run.
+      await runner.writeFile(PI_EVENT_FILTER_PATH, PI_EVENT_FILTER_SOURCE)
       if (params.search) {
         await runner.writeFile(PI_SEARCH_EXTENSION_PATH, PI_SEARCH_EXTENSION_SOURCE)
       }

apps/sim/executor/handlers/pi/cloud/babysit/backend.test.ts

@@ -655,6 +655,20 @@ describe('runBabysitPiWithOptions', () => {
       changedFiles: ['src/a.ts', 'src/b.ts'],
       diff: 'round-one-diff\nround-two-diff',
     })
+    const filterWrites = runner.writeFile.mock.calls.filter(
+      ([path]: [string]) => path === '/workspace/sim-pi-event-filter.mjs'
+    )
+    expect(filterWrites).toHaveLength(1)
+    expect(filterWrites[0]?.[1]).toContain("case 'message_update'")
+    const firstPiRun = runner.run.mock.calls.findIndex(([command]: [string]) =>
+      command.includes('pi -p --mode json')
+    )
+    const filterWrite = runner.writeFile.mock.calls.findIndex(
+      ([path]: [string]) => path === '/workspace/sim-pi-event-filter.mjs'
+    )
+    expect(runner.writeFile.mock.invocationCallOrder[filterWrite]).toBeLessThan(
+      runner.run.mock.invocationCallOrder[firstPiRun]
+    )
     expect(
       runCalls
         .filter(({ command }) => command.includes('CURRENT_DIGEST='))

apps/sim/executor/handlers/pi/cloud/babysit/backend.ts

@@ -36,6 +36,10 @@ import {
   MAX_THREADS_PER_ROUND,
   parseBabysitRound,
 } from '@/executor/handlers/pi/cloud/babysit/round'
+import {
+  PI_EVENT_FILTER_PATH,
+  PI_EVENT_FILTER_SOURCE,
+} from '@/executor/handlers/pi/cloud/event-filter-source'
 import {
   buildPiScript,
   CLONE_TIMEOUT_MS,
@@ -850,6 +854,7 @@ export async function runBabysitPiWithOptions(
           lastKnownChecksGreen
         )
       }
+      await runner.writeFile(PI_EVENT_FILTER_PATH, PI_EVENT_FILTER_SOURCE)
       if (params.search) {
         await runner.writeFile(PI_SEARCH_EXTENSION_PATH, PI_SEARCH_EXTENSION_SOURCE)
       }

apps/sim/executor/handlers/pi/cloud/event-filter-source.ts

@@ -0,0 +1,142 @@
+/**
+ * The stdout filter the sandbox modes pipe the Pi CLI through, written at runtime like the search
+ * extension and the review tools script next to it.
+ *
+ * Pi's `--mode json` writes `JSON.stringify(event)` for every session event with no filtering, and
+ * `message_update` repeats the whole assistant message alongside each delta — so raw stdout grows
+ * with the square of the response length, and `tool_execution_end`, `turn_end`, and `agent_end`
+ * each add a full tool result, turn transcript, or run transcript on top of that.
+ *
+ * The reduction has to happen in the sandbox because by the time Sim could drop the bytes they are
+ * already retained: E2B's SDK accumulates every callback-delivered chunk internally, so its adapter
+ * charges all of it against `MAX_SANDBOX_PROCESS_OUTPUT_BYTES` even though the caller streams it,
+ * and crossing that limit kills the sandbox mid-run — which on Create PR loses finished agent work
+ * before the push. Daytona keeps only a rolling tail of a streamed output, so there the same stream
+ * costs bandwidth and parse time rather than the run.
+ */
+
+/**
+ * Written outside `REPO_DIR` so `git add -A` cannot stage it into the user's pull request. The
+ * agent can still read or overwrite it — it holds bash on the sandbox — but the pipeline loads it
+ * at startup, before the agent's first turn, so a rewrite cannot change the events of the
+ * invocation that read it.
+ */
+export const PI_EVENT_FILTER_PATH = '/workspace/sim-pi-event-filter.mjs'
+
+/**
+ * Every emitted shape is a subset of the fields `normalizePiEvent` reads, so the events Sim parses
+ * are identical to the unfiltered stream's — an event it would normalize to `other` is dropped
+ * outright, since nothing downstream distinguishes those from an absent line.
+ *
+ * That makes the two modules a pair: a field added to `normalizePiEvent` has to be added here too,
+ * or it arrives only on the local and review paths while the cloud backends silently lose it.
+ * `shared.test.ts` pins the equivalence by running this filter and comparing normalized output
+ * against the normalized raw events, for the shapes it covers.
+ */
+export const PI_EVENT_FILTER_SOURCE = `/**
+ * Generated by Sim. Compacts the Pi CLI's JSON event stream before the sandbox provider retains it.
+ * Mirrors apps/sim/executor/handlers/pi/cloud/event-filter-source.ts.
+ */
+
+function asRecord(value) {
+  return typeof value === 'object' && value !== null && !Array.isArray(value) ? value : null
+}
+
+function asString(value) {
+  return typeof value === 'string' ? value : undefined
+}
+
+function compactUsage(value) {
+  const usage = asRecord(value)
+  if (!usage) return null
+  return {
+    input: usage.input,
+    output: usage.output,
+    inputTokens: usage.inputTokens,
+    outputTokens: usage.outputTokens,
+    prompt_tokens: usage.prompt_tokens,
+    completion_tokens: usage.completion_tokens,
+  }
+}
+
+function compactAssistantMessage(value) {
+  const message = asRecord(value)
+  if (!message || message.role !== 'assistant') return null
+  return {
+    role: 'assistant',
+    stopReason: asString(message.stopReason),
+    errorMessage: asString(message.errorMessage),
+  }
+}
+
+function compactEvent(value) {
+  const event = asRecord(value)
+  if (!event) return null
+
+  switch (event.type) {
+    case 'message_update': {
+      const update = asRecord(event.assistantMessageEvent)
+      if (update?.type !== 'text_delta' && update?.type !== 'thinking_delta') return null
+      return {
+        type: 'message_update',
+        assistantMessageEvent: { type: update.type, delta: asString(update.delta) },
+      }
+    }
+    case 'tool_execution_start':
+      return { type: 'tool_execution_start', toolName: asString(event.toolName) }
+    case 'tool_execution_end':
+      return {
+        type: 'tool_execution_end',
+        toolName: asString(event.toolName),
+        isError: event.isError === true,
+      }
+    case 'turn_end': {
+      const message = asRecord(event.message)
+      const usage = compactUsage(event.usage)
+      const messageUsage = compactUsage(message?.usage)
+      if (!usage && !messageUsage) return null
+      return {
+        type: 'turn_end',
+        ...(usage ? { usage } : {}),
+        ...(messageUsage ? { message: { usage: messageUsage } } : {}),
+      }
+    }
+    case 'agent_end': {
+      if (event.willRetry === true) return { type: 'agent_end', willRetry: true }
+      const messages = Array.isArray(event.messages) ? event.messages : []
+      let assistant = null
+      for (let index = messages.length - 1; index >= 0; index -= 1) {
+        assistant = compactAssistantMessage(messages[index])
+        if (assistant) break
+      }
+      return { type: 'agent_end', messages: assistant ? [assistant] : [] }
+    }
+    case 'error':
+      return {
+        type: 'error',
+        error: asString(event.error),
+        message: asString(event.message),
+      }
+    default:
+      return null
+  }
+}
+
+function processLine(line) {
+  if (!line.trim()) return
+  try {
+    const event = compactEvent(JSON.parse(line))
+    if (event) process.stdout.write(JSON.stringify(event) + '\\n')
+  } catch {}
+}
+
+process.stdin.setEncoding('utf8')
+let buffer = ''
+process.stdin.on('data', (chunk) => {
+  buffer += chunk
+  const lines = buffer.split('\\n')
+  buffer = lines.pop() ?? ''
+  for (const line of lines) processLine(line)
+})
+process.stdin.on('end', () => processLine(buffer))
+`

apps/sim/executor/handlers/pi/cloud/shared.test.ts

@@ -1,11 +1,17 @@
 /**
  * @vitest-environment node
  */
+import { spawn } from 'node:child_process'
 import { describe, expect, it } from 'vitest'
+import { MAX_SANDBOX_PROCESS_OUTPUT_BYTES } from '@/lib/execution/remote-sandbox/output-limits'
 import {
   PI_SANDBOX_MAX_LIFETIME_MS,
   PI_SANDBOX_MIN_LIFETIME_MS,
 } from '@/lib/execution/remote-sandbox/pi-lifetime'
+import {
+  PI_EVENT_FILTER_PATH,
+  PI_EVENT_FILTER_SOURCE,
+} from '@/executor/handlers/pi/cloud/event-filter-source'
 import {
   buildPiScript,
   CLONE_TIMEOUT_MS,
@@ -13,6 +19,41 @@ import {
   MIN_PI_TIMEOUT_MS,
   resolvePiTimeoutMs,
 } from '@/executor/handlers/pi/cloud/shared'
+import { normalizePiEvent } from '@/executor/handlers/pi/core/events'
+
+async function runPiEventFilter(chunks: Array<string | Uint8Array>): Promise<string> {
+  const child = spawn('node', ['--input-type=module', '--eval', PI_EVENT_FILTER_SOURCE], {
+    stdio: ['pipe', 'pipe', 'pipe'],
+  })
+  child.stdout.setEncoding('utf8')
+  child.stderr.setEncoding('utf8')
+  let stdout = ''
+  let stderr = ''
+  child.stdout.on('data', (chunk: string) => {
+    stdout += chunk
+  })
+  child.stderr.on('data', (chunk: string) => {
+    stderr += chunk
+  })
+
+  const completed = new Promise<number | null>((resolve, reject) => {
+    child.once('error', reject)
+    child.once('close', resolve)
+  })
+  for (const chunk of chunks) child.stdin.write(chunk)
+  child.stdin.end()
+
+  const exitCode = await completed
+  if (exitCode !== 0) throw new Error(`Pi event filter exited ${exitCode}: ${stderr}`)
+  return stdout
+}
+
+function parseJsonLines(output: string): Array<Record<string, unknown>> {
+  return output
+    .split('\n')
+    .filter(Boolean)
+    .map((line) => JSON.parse(line) as Record<string, unknown>)
+}
 
 describe('resolvePiTimeoutMs', () => {
   it('reserves every command budget that brackets the agent turn', () => {
@@ -73,8 +114,182 @@ describe('buildPiScript', () => {
     )
   })
 
-  it('preserves the Create PR command when no hardening option is passed', () => {
+  it('preserves Create PR repository-resource behavior when no hardening option is passed', () => {
     expect(buildPiScript()).not.toContain('--no-extensions')
     expect(buildPiScript()).not.toContain('--no-prompt-templates')
   })
+
+  it('filters Pi output inside the sandbox without masking pipeline failures', () => {
+    const script = buildPiScript()
+    expect(script).toMatch(/^\/bin\/bash -o pipefail -c '/)
+    expect(script).toContain(`| node ${PI_EVENT_FILTER_PATH}`)
+    expect(PI_EVENT_FILTER_PATH.startsWith('/workspace/repo')).toBe(false)
+  })
+})
+
+describe('PI_EVENT_FILTER_SOURCE', () => {
+  it('preserves every meaningful normalized event while removing cumulative payloads', async () => {
+    const rawEvents = [
+      {
+        type: 'message_update',
+        message: { role: 'assistant', content: [{ type: 'text', text: 'large history' }] },
+        assistantMessageEvent: { type: 'text_delta', delta: 'hello' },
+      },
+      {
+        type: 'message_update',
+        message: { role: 'assistant', content: [{ type: 'thinking', thinking: 'large thought' }] },
+        assistantMessageEvent: { type: 'thinking_delta', delta: 'hmm' },
+      },
+      { type: 'tool_execution_start', toolName: 'bash', args: { command: 'large command' } },
+      {
+        type: 'tool_execution_end',
+        toolName: 'bash',
+        result: { content: 'large result' },
+        isError: true,
+      },
+      {
+        type: 'turn_end',
+        usage: { inputTokens: 3, outputTokens: 4, ignored: 'provider detail' },
+        message: { usage: { input: 5, output: 6 }, content: 'large message' },
+        toolResults: [{ content: 'large tool result' }],
+      },
+      {
+        type: 'agent_end',
+        willRetry: true,
+        messages: [{ role: 'assistant', stopReason: 'error', errorMessage: 'retry me' }],
+      },
+      {
+        type: 'agent_end',
+        messages: [
+          { role: 'user', content: [{ type: 'text', text: 'old prompt' }] },
+          {
+            role: 'assistant',
+            stopReason: 'stop',
+            content: [
+              { type: 'thinking', thinking: 'private reasoning' },
+              { type: 'text', text: 'final answer' },
+            ],
+          },
+        ],
+      },
+      {
+        type: 'agent_end',
+        messages: [{ role: 'assistant', stopReason: 'aborted', errorMessage: 'stopped' }],
+      },
+      { type: 'error', error: 'provider failed', details: { response: 'large response' } },
+      { type: 'tool_execution_update', partialResult: { content: 'unused progress' } },
+    ]
+    const output = parseJsonLines(
+      await runPiEventFilter([`${rawEvents.map((event) => JSON.stringify(event)).join('\n')}\n`])
+    )
+
+    const meaningful = (events: unknown[]) =>
+      events.map(normalizePiEvent).filter((event) => event !== null && event.type !== 'other')
+    expect(meaningful(output)).toEqual(meaningful(rawEvents))
+
+    const update = output.find((event) => event.type === 'message_update')
+    expect(update).not.toHaveProperty('message')
+    const toolStart = output.find((event) => event.type === 'tool_execution_start')
+    expect(toolStart).not.toHaveProperty('args')
+    const toolEnd = output.find((event) => event.type === 'tool_execution_end')
+    expect(toolEnd).not.toHaveProperty('result')
+    const completed = output.find(
+      (event) =>
+        event.type === 'agent_end' &&
+        Array.isArray(event.messages) &&
+        event.messages.length > 0 &&
+        (event.messages[0] as { stopReason?: string }).stopReason === 'stop'
+    )
+    // Reduced to the one message `normalizePiEvent` inspects, and to the three fields it reads off
+    // it. The transcript, the thinking block, and the final text all go: the run's text reaches
+    // Sim through the deltas, so carrying it again here would be the whole answer twice.
+    expect(completed).toEqual({
+      type: 'agent_end',
+      messages: [{ role: 'assistant', stopReason: 'stop' }],
+    })
+    expect(output.some((event) => event.type === 'tool_execution_update')).toBe(false)
+  })
+
+  it('ignores malformed lines, accepts an unterminated final line, and preserves split UTF-8', async () => {
+    const line = Buffer.from(
+      JSON.stringify({
+        type: 'message_update',
+        assistantMessageEvent: { type: 'text_delta', delta: 'hello 🙂' },
+      })
+    )
+    const emojiStart = line.indexOf(Buffer.from('🙂'))
+    expect(emojiStart).toBeGreaterThan(0)
+    const output = parseJsonLines(
+      await runPiEventFilter([
+        '{not json}\n',
+        line.subarray(0, emojiStart + 1),
+        line.subarray(emojiStart + 1, emojiStart + 3),
+        line.subarray(emojiStart + 3),
+      ])
+    )
+
+    expect(output).toEqual([
+      {
+        type: 'message_update',
+        assistantMessageEvent: { type: 'text_delta', delta: 'hello 🙂' },
+      },
+    ])
+  })
+
+  it('keeps a cumulative stream over the sandbox limit compact without changing totals', async () => {
+    const totalCharacters = 12_000
+    const delta = 'xxxx'
+    const lines: string[] = []
+    for (let length = delta.length; length <= totalCharacters; length += delta.length) {
+      lines.push(
+        JSON.stringify({
+          type: 'message_update',
+          message: {
+            role: 'assistant',
+            content: [{ type: 'text', text: 'x'.repeat(length) }],
+          },
+          assistantMessageEvent: { type: 'text_delta', delta },
+        })
+      )
+    }
+    lines.push(
+      JSON.stringify({
+        type: 'turn_end',
+        message: { usage: { input: 10, output: 20 } },
+        toolResults: [{ content: 'x'.repeat(totalCharacters) }],
+      }),
+      JSON.stringify({
+        type: 'tool_execution_end',
+        toolName: 'read',
+        result: { content: 'x'.repeat(totalCharacters) },
+        isError: false,
+      }),
+      JSON.stringify({
+        type: 'agent_end',
+        messages: [
+          {
+            role: 'assistant',
+            stopReason: 'stop',
+            content: [{ type: 'text', text: 'x'.repeat(totalCharacters) }],
+          },
+        ],
+      })
+    )
+    const raw = `${lines.join('\n')}\n`
+    expect(Buffer.byteLength(raw)).toBeGreaterThan(MAX_SANDBOX_PROCESS_OUTPUT_BYTES)
+
+    const filtered = await runPiEventFilter([raw])
+    expect(Buffer.byteLength(filtered)).toBeLessThan(MAX_SANDBOX_PROCESS_OUTPUT_BYTES)
+    const events = parseJsonLines(filtered)
+      .map(normalizePiEvent)
+      .filter((event) => event !== null)
+    const text = events
+      .filter((event) => event.type === 'text')
+      .map((event) => event.text)
+      .join('')
+    expect(text).toHaveLength(totalCharacters)
+    expect(events).toContainEqual({ type: 'usage', inputTokens: 10, outputTokens: 20 })
+    expect(events).toContainEqual({ type: 'tool_end', toolName: 'read', isError: false })
+    expect(events).toContainEqual({ type: 'final' })
+  })
 })

apps/sim/executor/handlers/pi/cloud/shared.ts

@@ -7,6 +7,7 @@
 
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { resolvePiSandboxLifetimeMs } from '@/lib/execution/remote-sandbox/pi-lifetime'
+import { PI_EVENT_FILTER_PATH } from '@/executor/handlers/pi/cloud/event-filter-source'
 import { scrubPiSecrets } from '@/executor/handlers/pi/core/redaction'
 
 export const REPO_DIR = '/workspace/repo'
@@ -128,11 +129,19 @@ export const PUSH_SCRIPT = `cd ${REPO_DIR}
 /usr/bin/git -c core.hooksPath=/dev/null -c credential.helper= -c core.fsmonitor= push "https://x-access-token:$GITHUB_TOKEN@github.com/$REPO_OWNER/$REPO_NAME.git" "HEAD:refs/heads/$BRANCH" >/dev/null 2>${PUSH_ERR_PATH} && echo "__PUSHED__=1"`
 
 /**
- * The Pi CLI invocation for the sandbox modes. With no options it emits exactly what Create PR
- * always did.
+ * The Pi CLI invocation for the sandbox modes, piped through the sandbox event filter. The command
+ * names {@link PI_EVENT_FILTER_PATH}, so every caller must have written `PI_EVENT_FILTER_SOURCE`
+ * there first — skipping that write does not fall back to the raw stream, it fails the run on the
+ * missing module.
  *
- * With one, `--no-extensions` drops any extension the cloned repository ships while leaving the
- * explicit `-e` path loaded, so the loaded set is exactly Sim's own extension. That is deliberate —
+ * Selects `/bin/bash` explicitly because `pipefail` is not portable to `/bin/sh`, and without it
+ * the pipeline reports the filter's exit code rather than Pi's, so an upstream crash would read as
+ * a clean run. Both dedicated Pi images are Debian-based and provide Bash, so provider
+ * default-shell behavior cannot change whether an upstream Pi failure reaches the caller.
+ *
+ * With no options the repository resources Pi loads are exactly what Create PR always had. With an
+ * `extensionPath`, `--no-extensions` drops any extension the cloned repository ships while leaving
+ * the explicit `-e` path loaded, so the loaded set is exactly Sim's own extension. That is deliberate —
  * a repository must not be able to register tools into a run holding the workspace's keys — but it
  * does mean enabling search also stops loading a repository's own Pi extensions, which is why the
  * flag is not passed on Create PR's no-search path. Babysit supplies
@@ -149,8 +158,8 @@ export function buildPiScript(
       ? ' --no-extensions'
       : ''
   const extensionArgs = extensionPath ? ` -e ${extensionPath}` : ''
-  return `cd ${REPO_DIR}
-pi -p --mode json --provider "$PI_PROVIDER" --model "$PI_MODEL" --thinking "$PI_THINKING"${repositoryArgs}${extensionArgs} < ${PROMPT_PATH}`
+  return `/bin/bash -o pipefail -c 'cd ${REPO_DIR}
+pi -p --mode json --provider "$PI_PROVIDER" --model "$PI_MODEL" --thinking "$PI_THINKING"${repositoryArgs}${extensionArgs} < ${PROMPT_PATH} | node ${PI_EVENT_FILTER_PATH}'`
 }
 
 export function raceAbort<T>(promise: Promise<T>, signal?: AbortSignal): Promise<T> {

apps/sim/executor/handlers/pi/core/events.ts

@@ -114,7 +114,13 @@ function extractUsage(
   return null
 }
 
-/** Normalizes a raw Pi/SDK event object into a {@link PiEvent}. */
+/**
+ * Normalizes a raw Pi/SDK event object into a {@link PiEvent}.
+ *
+ * The cloud backends feed this from a stream the sandbox already reduced to the fields read below
+ * (`cloud/event-filter-source.ts`), so a field added here has to be added there too — otherwise it
+ * arrives only on the local and review paths and the cloud ones silently lose it.
+ */
 export function normalizePiEvent(raw: unknown): PiEvent | null {
   const ev = asRecord(raw)
   if (!ev) return null