Function cellOnBlockComplete

apps/sim/background/resume-execution.ts:209–255

(blockId: string, output: unknown)

  let terminalWritten = false

  const cellOnBlockComplete = async (blockId: string, output: unknown) => {
    const outputs = outputsByBlockId.get(blockId)
    if (!outputs) return
    const blockResult =
      output && typeof output === 'object' && 'output' in (output as object)
        ? (output as { output: unknown }).output
        : output
    const errorMessage =
      blockResult &&
      typeof blockResult === 'object' &&
      typeof (blockResult as { error?: unknown }).error === 'string'
        ? (blockResult as { error: string }).error
        : null
    if (errorMessage) {
      blockErrors[blockId] = errorMessage
    } else {
      for (const out of outputs) {
        const plucked = pluckByPath(blockResult, out.path)
        if (plucked === undefined) continue
        accumulatedData[out.columnName] = plucked as RowData[string]
      }
    }
    const dataSnapshot: RowData = { ...accumulatedData }
    const blockErrorsSnapshot = { ...blockErrors }
    writeChain = writeChain
      .then(async () => {
        if (terminalWritten) return
        const partial: RowExecutionMetadata = {
          status: 'running',
          executionId: parentExecutionId,
          jobId: null,
          workflowId: cellContext.workflowId,
          error: null,
          blockErrors: blockErrorsSnapshot,
        }
        await writeWorkflowGroupState(writeCtx, {
          executionState: partial,
          dataPatch: dataSnapshot,
        })
      })
      .catch((err) => {
        logger.warn(
          `Resume per-block partial write failed (table=${cellContext.tableId} row=${cellContext.rowId} group=${cellContext.groupId}):`,
          err
        )
      })
  }

  const writeCellTerminal = async (
    status: 'completed' | 'error' | 'paused',