From 35ab23cbe35684606ba71bc7c72da617d137b828 Mon Sep 17 00:00:00 2001 From: Bill Leoutsakos Date: Sat, 8 Aug 2026 17:40:19 -0700 Subject: [PATCH 1/3] fix(pi): compact cloud event streams --- .../pi/cloud/authoring/backend.test.ts | 11 + .../handlers/pi/cloud/authoring/backend.ts | 3 + .../handlers/pi/cloud/babysit/backend.test.ts | 14 ++ .../handlers/pi/cloud/babysit/backend.ts | 3 + .../executor/handlers/pi/cloud/shared.test.ts | 218 +++++++++++++++++- apps/sim/executor/handlers/pi/cloud/shared.ts | 127 +++++++++- 6 files changed, 371 insertions(+), 5 deletions(-) diff --git a/apps/sim/executor/handlers/pi/cloud/authoring/backend.test.ts b/apps/sim/executor/handlers/pi/cloud/authoring/backend.test.ts index d5fe5254094..c670c810deb 100644 --- a/apps/sim/executor/handlers/pi/cloud/authoring/backend.test.ts +++ b/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 "...". diff --git a/apps/sim/executor/handlers/pi/cloud/authoring/backend.ts b/apps/sim/executor/handlers/pi/cloud/authoring/backend.ts index ee93afbe397..84f9f6ba47b 100644 --- a/apps/sim/executor/handlers/pi/cloud/authoring/backend.ts +++ b/apps/sim/executor/handlers/pi/cloud/authoring/backend.ts @@ -43,6 +43,8 @@ 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, @@ -466,6 +468,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) } diff --git a/apps/sim/executor/handlers/pi/cloud/babysit/backend.test.ts b/apps/sim/executor/handlers/pi/cloud/babysit/backend.test.ts index 0ec2876f183..79ff9993515 100644 --- a/apps/sim/executor/handlers/pi/cloud/babysit/backend.test.ts +++ b/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=')) diff --git a/apps/sim/executor/handlers/pi/cloud/babysit/backend.ts b/apps/sim/executor/handlers/pi/cloud/babysit/backend.ts index df2ec3919e8..3b48137d419 100644 --- a/apps/sim/executor/handlers/pi/cloud/babysit/backend.ts +++ b/apps/sim/executor/handlers/pi/cloud/babysit/backend.ts @@ -47,6 +47,8 @@ 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, @@ -850,6 +852,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) } diff --git a/apps/sim/executor/handlers/pi/cloud/shared.test.ts b/apps/sim/executor/handlers/pi/cloud/shared.test.ts index f58c020064d..ec260ef5d83 100644 --- a/apps/sim/executor/handlers/pi/cloud/shared.test.ts +++ b/apps/sim/executor/handlers/pi/cloud/shared.test.ts @@ -1,7 +1,9 @@ /** * @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, @@ -11,8 +13,45 @@ import { 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' + +async function runPiEventFilter(chunks: Array): Promise { + 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((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> { + return output + .split('\n') + .filter(Boolean) + .map((line) => JSON.parse(line) as Record) +} describe('resolvePiTimeoutMs', () => { it('reserves every command budget that brackets the agent turn', () => { @@ -73,8 +112,185 @@ 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).toContain('set -o pipefail') + 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' + ) + expect(completed).toEqual({ + type: 'agent_end', + messages: [ + { + role: 'assistant', + content: [{ type: 'text', text: 'final answer' }], + 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' }) + }) }) diff --git a/apps/sim/executor/handlers/pi/cloud/shared.ts b/apps/sim/executor/handlers/pi/cloud/shared.ts index a0cc91157cf..39fe38c6e30 100644 --- a/apps/sim/executor/handlers/pi/cloud/shared.ts +++ b/apps/sim/executor/handlers/pi/cloud/shared.ts @@ -14,6 +14,7 @@ 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 @@ -128,8 +129,125 @@ 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. + * 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. * * 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 — @@ -149,8 +267,9 @@ 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 `set -o pipefail +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(promise: Promise, signal?: AbortSignal): Promise { From 0453c8ecb767320fd9e95024e8c2c3a85bba18a8 Mon Sep 17 00:00:00 2001 From: Bill Leoutsakos Date: Sat, 8 Aug 2026 17:49:02 -0700 Subject: [PATCH 2/3] fix(pi): select bash for event pipeline --- apps/sim/executor/handlers/pi/cloud/shared.test.ts | 2 +- apps/sim/executor/handlers/pi/cloud/shared.ts | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/apps/sim/executor/handlers/pi/cloud/shared.test.ts b/apps/sim/executor/handlers/pi/cloud/shared.test.ts index ec260ef5d83..d5a78371b2f 100644 --- a/apps/sim/executor/handlers/pi/cloud/shared.test.ts +++ b/apps/sim/executor/handlers/pi/cloud/shared.test.ts @@ -119,7 +119,7 @@ describe('buildPiScript', () => { it('filters Pi output inside the sandbox without masking pipeline failures', () => { const script = buildPiScript() - expect(script).toContain('set -o pipefail') + 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) }) diff --git a/apps/sim/executor/handlers/pi/cloud/shared.ts b/apps/sim/executor/handlers/pi/cloud/shared.ts index 39fe38c6e30..b101537c3f8 100644 --- a/apps/sim/executor/handlers/pi/cloud/shared.ts +++ b/apps/sim/executor/handlers/pi/cloud/shared.ts @@ -249,6 +249,10 @@ 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. * + * 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. + * * 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 — * a repository must not be able to register tools into a run holding the workspace's keys — but it @@ -267,9 +271,8 @@ export function buildPiScript( ? ' --no-extensions' : '' const extensionArgs = extensionPath ? ` -e ${extensionPath}` : '' - return `set -o pipefail -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}` + 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(promise: Promise, signal?: AbortSignal): Promise { From 1234a31fc606bdd7ddd5718532e8a7f8efe52cb4 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Sat, 8 Aug 2026 18:22:08 -0700 Subject: [PATCH 3/3] address comments --- .../handlers/pi/cloud/authoring/backend.ts | 6 +- .../handlers/pi/cloud/babysit/backend.ts | 6 +- .../handlers/pi/cloud/event-filter-source.ts | 142 ++++++++++++++++++ .../executor/handlers/pi/cloud/shared.test.ts | 17 +-- apps/sim/executor/handlers/pi/cloud/shared.ts | 137 ++--------------- apps/sim/executor/handlers/pi/core/events.ts | 8 +- 6 files changed, 177 insertions(+), 139 deletions(-) create mode 100644 apps/sim/executor/handlers/pi/cloud/event-filter-source.ts diff --git a/apps/sim/executor/handlers/pi/cloud/authoring/backend.ts b/apps/sim/executor/handlers/pi/cloud/authoring/backend.ts index 84f9f6ba47b..fa2a7bcde13 100644 --- a/apps/sim/executor/handlers/pi/cloud/authoring/backend.ts +++ b/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, diff --git a/apps/sim/executor/handlers/pi/cloud/babysit/backend.ts b/apps/sim/executor/handlers/pi/cloud/babysit/backend.ts index 3b48137d419..91b71b72ede 100644 --- a/apps/sim/executor/handlers/pi/cloud/babysit/backend.ts +++ b/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, diff --git a/apps/sim/executor/handlers/pi/cloud/event-filter-source.ts b/apps/sim/executor/handlers/pi/cloud/event-filter-source.ts new file mode 100644 index 00000000000..aa0a0a396a5 --- /dev/null +++ b/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)) +` diff --git a/apps/sim/executor/handlers/pi/cloud/shared.test.ts b/apps/sim/executor/handlers/pi/cloud/shared.test.ts index d5a78371b2f..bf3ca198e7d 100644 --- a/apps/sim/executor/handlers/pi/cloud/shared.test.ts +++ b/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) }) diff --git a/apps/sim/executor/handlers/pi/cloud/shared.ts b/apps/sim/executor/handlers/pi/cloud/shared.ts index b101537c3f8..8d8c7ccc7e4 100644 --- a/apps/sim/executor/handlers/pi/cloud/shared.ts +++ b/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' @@ -14,7 +15,6 @@ 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 diff --git a/apps/sim/executor/handlers/pi/core/events.ts b/apps/sim/executor/handlers/pi/core/events.ts index 4d79683f09a..feff476ec67 100644 --- a/apps/sim/executor/handlers/pi/core/events.ts +++ b/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