Human-in-the-Loop Workflows

import { ApiRouteConfig, Handlers } from 'motia'
 
export const config: ApiRouteConfig = {
  type: 'api',
  name: 'SubmitOrder',
  path: '/orders',
  method: 'POST',
  emits: ['order.submitted'],
  flows: ['order-approval']
}
 
export const handler: Handlers['SubmitOrder'] = async (req, { state, emit, logger }) => {
  const orderId = crypto.randomUUID()
  
  const order = {
    id: orderId,
    items: req.body.items,
    total: req.body.total,
    status: 'pending_analysis',
    currentStep: 'submitted',
    completedSteps: ['submitted'],  // Track progress for idempotency
    createdAt: new Date().toISOString()
  }
  
  // Save immediately - this is our first checkpoint
  await state.set('orders', orderId, order)
  
  logger.info('Order submitted', { orderId })
  
  // Kick off risk analysis
  await emit({ 
    topic: 'order.submitted', 
    data: { orderId } 
  })
  
  return { status: 201, body: { orderId, status: order.status } }
}

import { EventConfig, Handlers } from 'motia'
 
export const config: EventConfig = {
  type: 'event',
  name: 'AnalyzeRisk',
  subscribes: ['order.submitted'],
  flows: ['order-approval'],
  emits: ['order.auto_approved'],
  virtualEmits: [{ topic: 'approval.required', label: 'Requires human approval' }]
}
 
export const handler: Handlers['AnalyzeRisk'] = async (input, { state, emit, logger }) => {
  const { orderId } = input
  const order = await state.get('orders', orderId)
  
  // Skip if already analyzed (idempotency)
  if (order.completedSteps.includes('risk_analysis')) {
    return
  }
  
  // Calculate risk score
  const riskScore = calculateRiskScore(order)
  order.riskScore = riskScore
  order.completedSteps.push('risk_analysis')
  
  if (riskScore > 70) {
    // High risk - PAUSE for human approval
    order.status = 'awaiting_approval'
    order.currentStep = 'awaiting_approval'
    await state.set('orders', orderId, order)
    
    logger.warn('Order requires approval - workflow paused', { orderId, riskScore })
    
    // NO EMIT - workflow stops here
    // Webhook will restart it when human makes decision
    
  } else {
    // Low risk - bypass gate and auto-approve
    order.status = 'approved'
    order.approvedBy = 'system'
    order.completedSteps.push('approved')
    await state.set('orders', orderId, order)
    
    logger.info('Order auto-approved', { orderId, riskScore })
    
    // Continue immediately
    await emit({ topic: 'order.auto_approved', data: { orderId } })
  }
}
 
function calculateRiskScore(order: any): number {
  let score = 0
  if (order.total > 1000) score += 40
  else if (order.total > 500) score += 20
  
  const itemCount = order.items.reduce((sum, item) => sum + item.quantity, 0)
  if (itemCount > 10) score += 30
  
  score += Math.random() * 40
  return Math.min(Math.round(score), 100)
}

import { ApiRouteConfig, Handlers } from 'motia'
 
export const config: ApiRouteConfig = {
  type: 'api',
  name: 'ApprovalWebhook',
  path: '/webhooks/orders/:orderId/approve',
  method: 'POST',
  emits: ['order.approved'],
  virtualSubscribes: ['human.decision'],  // Shows connection from Noop
  flows: ['order-approval']
}
 
export const handler: Handlers['ApprovalWebhook'] = async (req, { state, emit, logger }) => {
  const { orderId } = req.pathParams
  const { approved, approvedBy, notes } = req.body
  
  // Load the checkpoint
  const order = await state.get('orders', orderId)
  
  if (!order) {
    return { status: 404, body: { error: 'Order not found' } }
  }
  
  // Verify we're at the right pause point
  if (order.currentStep !== 'awaiting_approval') {
    return { status: 400, body: { error: 'Order not awaiting approval' } }
  }
  
  if (approved) {
    // Apply decision
    order.status = 'approved'
    order.approvedBy = approvedBy
    order.approvedAt = new Date().toISOString()
    order.completedSteps.push('approved')
    await state.set('orders', orderId, order)
    
    logger.info('Order approved - resuming workflow', { orderId, approvedBy })
    
    // Resume the workflow
    await emit({ topic: 'order.approved', data: { orderId } })
    
    return { status: 200, body: { success: true, message: 'Order approved' } }
  } else {
    // Rejected - workflow ends
    order.status = 'rejected'
    order.rejectedBy = approvedBy
    order.rejectionReason = notes
    await state.set('orders', orderId, order)
    
    logger.info('Order rejected - workflow ends', { orderId })
    return { status: 200, body: { success: true, message: 'Order rejected' } }
  }
}

import { EventConfig, Handlers } from 'motia'
 
export const config: EventConfig = {
  type: 'event',
  name: 'CompleteOrder',
  subscribes: ['order.approved', 'order.auto_approved'],
  flows: ['order-approval'],
  emits: []
}
 
export const handler: Handlers['CompleteOrder'] = async (input, { state, logger }) => {
  const { orderId } = input
  const order = await state.get('orders', orderId)
  
  // Skip if already completed (idempotency)
  if (order.completedSteps.includes('completed')) {
    return
  }
  
  // Process fulfillment
  await simulateFulfillment(order)
  
  order.status = 'completed'
  order.completedAt = new Date().toISOString()
  order.completedSteps.push('completed')
  await state.set('orders', orderId, order)
  
  logger.info('Order completed', { orderId, approvedBy: order.approvedBy })
}
 
async function simulateFulfillment(order: any) {
  // In production: charge payment, create shipment, send email
  await new Promise(resolve => setTimeout(resolve, 1000))
}

import { CronConfig, Handlers } from 'motia'
 
export const config: CronConfig = {
  type: 'cron',
  name: 'DetectTimeouts',
  description: 'Find orders stuck awaiting approval and escalate',
  cron: '*/5 * * * *',  // Every 5 minutes (for demo - use hourly in production)
  flows: ['order-approval'],
  emits: []
}
 
export const handler: Handlers['DetectTimeouts'] = async ({ state, logger }) => {
  const orders = await state.getGroup('orders')
  const now = Date.now()
  const timeoutMs = 10 * 60 * 1000  // 10 minutes (demo - use 24 hours in production)
  
  let stuckCount = 0
  
  for (const order of orders) {
    // Find orders stuck in awaiting_approval
    if (order.status === 'awaiting_approval') {
      const lastUpdate = new Date(order.updatedAt || order.createdAt).getTime()
      const age = now - lastUpdate
      
      if (age > timeoutMs) {
        stuckCount++
        
        logger.warn('Approval timeout detected', {
          orderId: order.id,
          ageMinutes: Math.round(age / (60 * 1000)),
          riskScore: order.riskScore,
          total: order.total
        })
        
        // Mark as timed out
        order.status = 'timeout'
        order.timeoutAt = new Date().toISOString()
        order.timeoutReason = `No approval within ${Math.round(timeoutMs / (60 * 1000))} min`
        await state.set('orders', order.id, order)
        
        // In production, take action:
        // 1. Send escalation notification (Slack, email)
        // 2. Auto-reject if too old
        // 3. Assign to different manager
        // 4. Create support ticket
        
        logger.info('Escalation triggered', {
          orderId: order.id,
          action: 'timeout_escalation',
          notifyChannel: '#urgent-approvals'
        })
      }
    }
  }
  
  if (stuckCount > 0) {
    logger.info('Timeout detection complete', { 
      stuckCount,
      totalOrders: orders.length 
    })
  }
}