Method runPi

apps/sim/executor/handlers/pi/pi-handler.ts:195–261

(
    ctx: ExecutionContext,
    block: SerializedBlock,
    backend: PiBackendRun<P>,
    params: P,
    memoryConfig: PiMemoryConfig
  )

  }

  private async runPi<P extends PiRunParams>(
    ctx: ExecutionContext,
    block: SerializedBlock,
    backend: PiBackendRun<P>,
    params: P,
    memoryConfig: PiMemoryConfig
  ): Promise<BlockOutput | StreamingExecution> {
    const startTime = Date.now()
    const startTimeISO = new Date(startTime).toISOString()

    logger.info('Executing Pi block', {
      blockId: block.id,
      mode: params.mode,
      model: params.model,
      workflowId: ctx.workflowId,
      executionId: ctx.executionId,
    })

    if (this.isContentSelectedForStreaming(ctx, block)) {
      const output: NormalizedBlockOutput = { content: '', model: params.model }
      const stream = new ReadableStream<Uint8Array>({
        start: async (controller) => {
          const encoder = new TextEncoder()
          try {
            const result = await backend(params, {
              onEvent: (event) => {
                const text = streamTextForEvent(event)
                if (text) controller.enqueue(encoder.encode(text))
              },
              signal: ctx.abortSignal,
            })
            if (result.totals.errorMessage) {
              controller.error(new Error(result.totals.errorMessage))
              return
            }
            Object.assign(
              output,
              this.buildOutput(result, params.model, params.isBYOK, startTime, startTimeISO)
            )
            await appendPiMemory(ctx, memoryConfig, params.task, result.totals.finalText)
            controller.close()
          } catch (error) {
            controller.error(error)
          }
        },
      })

      return {
        stream,
        execution: {
          success: true,
          output,
          blockId: block.id,
          logs: [],
          metadata: { startTime: startTimeISO, duration: 0 },
          isStreaming: true,
        } as StreamingExecution['execution'] & { blockId: string },
      }