fix(pi): compact plan mode event streams

Author: Bill Leoutsakos

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

@@ -46,6 +46,18 @@ function asString(value) {
   return typeof value === 'string' ? value : undefined
 }
 
+function compactAssistantContent(value) {
+  if (!Array.isArray(value)) return undefined
+  const content = []
+  for (const valueBlock of value) {
+    const block = asRecord(valueBlock)
+    if (block?.type === 'text' && typeof block.text === 'string') {
+      content.push({ type: 'text', text: block.text })
+    }
+  }
+  return content.length > 0 ? content : undefined
+}
+
 function compactUsage(value) {
   const usage = asRecord(value)
   if (!usage) return null
@@ -64,6 +76,7 @@ function compactAssistantMessage(value) {
   if (!message || message.role !== 'assistant') return null
   return {
     role: 'assistant',
+    content: compactAssistantContent(message.content),
     stopReason: asString(message.stopReason),
     errorMessage: asString(message.errorMessage),
   }

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

@@ -30,6 +30,10 @@ vi.mock('@/executor/handlers/pi/core/keys', () => ({
   mapThinkingLevel: () => 'high',
 }))
 
+import {
+  PI_EVENT_FILTER_PATH,
+  PI_EVENT_FILTER_SOURCE,
+} from '@/executor/handlers/pi/cloud/event-filter-source'
 import { runCloudPlanPi } from '@/executor/handlers/pi/cloud/plan/backend'
 import type { PiCloudPlanRunParams } from '@/executor/handlers/pi/core/backend'
 import {
@@ -99,6 +103,7 @@ describe('runCloudPlanPi', () => {
 
     const [piCommand, piOptions] = mockRun.mock.calls[1]
     expect(piCommand).toContain('--no-extensions --no-prompt-templates --no-skills --no-approve')
+    expect(piCommand).toContain(`| node ${PI_EVENT_FILTER_PATH}`)
     expect(piCommand).not.toContain('git commit')
     expect(piCommand).not.toContain('git push')
     expect(piOptions.envs.ANTHROPIC_API_KEY).toBe('sk-model-secret')
@@ -116,6 +121,18 @@ describe('runCloudPlanPi', () => {
       })
     )
     expect(mockWriteFile).toHaveBeenCalledWith('/workspace/pi-prompt.txt', 'PLAN PROMPT')
+    expect(mockWriteFile).toHaveBeenCalledWith(PI_EVENT_FILTER_PATH, PI_EVENT_FILTER_SOURCE)
+    const filterWrites = mockWriteFile.mock.calls.filter(([path]) => path === PI_EVENT_FILTER_PATH)
+    expect(filterWrites).toHaveLength(1)
+    const filterWrite = mockWriteFile.mock.calls.findIndex(
+      ([path]) => path === PI_EVENT_FILTER_PATH
+    )
+    expect(mockRun.mock.invocationCallOrder[0]).toBeLessThan(
+      mockWriteFile.mock.invocationCallOrder[filterWrite]
+    )
+    expect(mockWriteFile.mock.invocationCallOrder[filterWrite]).toBeLessThan(
+      mockRun.mock.invocationCallOrder[1]
+    )
     expect(onEvent).toHaveBeenCalledWith({ type: 'text', text: '# Plan\nDo it' })
     expect(result.totals.finalText).toBe('# Plan\nDo it')
     expect(result).not.toHaveProperty('changedFiles')

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

@@ -8,6 +8,10 @@
 import { createLogger } from '@sim/logger'
 import { withPiSandbox } from '@/lib/execution/remote-sandbox'
 import { resolvePiRunLifetimeMs } 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,
@@ -97,6 +101,7 @@ export const runCloudPlanPi: PiBackendRun<PiCloudPlanRunParams> = async (params,
       }
 
       await runner.writeFile(PROMPT_PATH, prompt)
+      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/shared.test.ts

@@ -207,12 +207,18 @@ 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.
+    // Reduced to the one message `normalizePiEvent` inspects and only the fields it reads. Keeping
+    // the final text once lets Plan replace streamed progress with the authoritative final answer;
+    // the transcript, thinking blocks, and tool payloads still go.
     expect(completed).toEqual({
       type: 'agent_end',
-      messages: [{ role: 'assistant', stopReason: 'stop' }],
+      messages: [
+        {
+          role: 'assistant',
+          content: [{ type: 'text', text: 'final answer' }],
+          stopReason: 'stop',
+        },
+      ],
     })
     expect(output.some((event) => event.type === 'tool_execution_update')).toBe(false)
   })
@@ -297,6 +303,6 @@ describe('PI_EVENT_FILTER_SOURCE', () => {
     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' })
+    expect(events).toContainEqual({ type: 'final', text: 'x'.repeat(totalCharacters) })
   })
 })