address comments

Author: icecrasher321

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,
@@ -43,8 +47,6 @@ import {
   FINALIZE_TIMEOUT_MS,
   GIT_CONFIG_DIGEST_LINE,
   MAX_DIFF_BYTES,
-  PI_EVENT_FILTER_PATH,
-  PI_EVENT_FILTER_SOURCE,
   PREPARE_SCRIPT,
   PROMPT_PATH,
   PUSH_ERR_PATH,

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,
@@ -47,8 +51,6 @@ import {
   GIT_CONFIG_DIGEST_MARKER,
   MAX_DIFF_BYTES,
   MIN_PI_TIMEOUT_MS,
-  PI_EVENT_FILTER_PATH,
-  PI_EVENT_FILTER_SOURCE,
   PROMPT_PATH,
   PUSH_ERROR_MAX,
   REPO_DIR,

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

@@ -8,13 +8,15 @@ 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,
   FINALIZE_TIMEOUT_MS,
   MIN_PI_TIMEOUT_MS,
-  PI_EVENT_FILTER_PATH,
-  PI_EVENT_FILTER_SOURCE,
   resolvePiTimeoutMs,
 } from '@/executor/handlers/pi/cloud/shared'
 import { normalizePiEvent } from '@/executor/handlers/pi/core/events'
@@ -198,15 +200,12 @@ describe('PI_EVENT_FILTER_SOURCE', () => {
         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',
-          content: [{ type: 'text', text: 'final answer' }],
-          stopReason: 'stop',
-        },
-      ],
+      messages: [{ role: 'assistant', stopReason: 'stop' }],
     })
     expect(output.some((event) => event.type === 'tool_execution_update')).toBe(false)
   })

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

@@ -7,14 +7,14 @@
 
 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'
 export const PROMPT_PATH = '/workspace/pi-prompt.txt'
 export const DIFF_PATH = '/workspace/pi.diff'
 export const COMMIT_MSG_PATH = '/workspace/pi-commit.txt'
 export const PUSH_ERR_PATH = '/workspace/pi-push-err.txt'
-export const PI_EVENT_FILTER_PATH = '/workspace/sim-pi-event-filter.mjs'
 export const CLONE_TIMEOUT_MS = 10 * 60 * 1000
 export const FINALIZE_TIMEOUT_MS = 10 * 60 * 1000
 export const MAX_DIFF_BYTES = 200_000
@@ -129,132 +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"`
 
 /**
- * Reduces Pi's cumulative JSON event stream before either sandbox provider retains it.
- * The emitted shapes contain only fields consumed by `normalizePiEvent`; in particular,
- * every `message_update` drops the full message Pi repeats alongside its delta.
- */
-export const PI_EVENT_FILTER_SOURCE = `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
-  const content = Array.isArray(message.content)
-    ? message.content.flatMap((value) => {
-        const block = asRecord(value)
-        return block?.type === 'text' && typeof block.text === 'string'
-          ? [{ type: 'text', text: block.text }]
-          : []
-      })
-    : []
-  return {
-    role: 'assistant',
-    content,
-    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))
-`
-
-/**
- * The Pi CLI invocation for the sandbox modes. Every caller receives the compact event stream;
- * options only control which repository resources Pi may load.
+ * 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.
  *
- * Selects `/bin/bash` explicitly because `pipefail` is not portable to `/bin/sh`. 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.
+ * 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 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 —
+ * 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

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