Function executeInboxTask

apps/sim/lib/mothership/inbox/executor.ts:44–323

(taskId: string)

 * 7. Update task status
 */
export async function executeInboxTask(taskId: string): Promise<void> {
  const [inboxTask] = await db
    .select()
    .from(mothershipInboxTask)
    .where(eq(mothershipInboxTask.id, taskId))
    .limit(1)

  if (!inboxTask) {
    logger.error('Inbox task not found', { taskId })
    return
  }

  if (inboxTask.status === 'completed' || inboxTask.status === 'failed') {
    logger.info('Inbox task already terminal, skipping', { taskId, status: inboxTask.status })
    return
  }

  const [ws] = await db
    .select({
      id: workspace.id,
      ownerId: workspace.ownerId,
      inboxProviderId: workspace.inboxProviderId,
    })
    .from(workspace)
    .where(eq(workspace.id, inboxTask.workspaceId))
    .limit(1)

  if (!ws) {
    logger.error('Workspace not found for inbox task', {
      taskId,
      workspaceId: inboxTask.workspaceId,
    })
    await markTaskFailed(taskId, 'Workspace not found')
    return
  }

  let chatId = inboxTask.chatId
  let responseSent = false

  try {
    const [[claimed], userId] = await Promise.all([
      db
        .update(mothershipInboxTask)
        .set({ status: 'processing', processingStartedAt: new Date() })
        .where(and(eq(mothershipInboxTask.id, taskId), eq(mothershipInboxTask.status, 'received')))
        .returning({ id: mothershipInboxTask.id }),
      resolveUserId(inboxTask.fromEmail, ws),
    ])

    if (!claimed) {
      logger.info('Task already claimed by another execution, skipping', { taskId })
      return
    }

    // Blocked senders and banned accounts must not drive the agent; the sender
    // email is checked directly (domain list + the sender's own account ban)
    // because non-members resolve to the workspace owner, and the workspace
    // billed account is checked to match preprocessExecution's gate. Fails