Human-in-the-Loop AI: Resumable Workflows for SaaS MVPs

+-------------------------------------------------------------+
|                  TRADITIONAL BRITTLE HITL                   |
|                                                             |
|  [Agent Step 1] -> [Agent Step 2] -> [Wait for User (Sync)] |
|                                              |              |
|                                     [HTTP Timeout / Crash]  |
|                                              v              |
|                                      (State Evaporates)     |
+-------------------------------------------------------------+

+-------------------------------------------------------------+
|               DURABLE RESUMABLE ARCHITECTURE                |
|                                                             |
|  [Step 1] -> [Persist State] -> [Emit Approval Webhook]     |
|                                           |                 |
|                               (Agent Process Suspends)      |
|                                           |                 |
|  [Human Approves in UI/Slack] -> [Resume via DB Checkpoint] |
|                                           |                 |
|                               [Step 2 (Deterministic)]      |
+-------------------------------------------------------------+

-- Migration: create_hitl_workflow_tables.sql
CREATE TYPE workflow_status AS ENUM ('running', 'suspended_for_approval', 'completed', 'rejected', 'failed');

CREATE TABLE hitl_workflows (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    workflow_type VARCHAR(64) NOT NULL,
    status workflow_status NOT NULL DEFAULT 'running',
    current_step VARCHAR(64) NOT NULL,
    state_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
    pending_action JSONB,
    approval_token VARCHAR(128),
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_hitl_tenant_status ON hitl_workflows(tenant_id, status);
CREATE UNIQUE INDEX idx_hitl_approval_token ON hitl_workflows(approval_token) WHERE approval_token IS NOT NULL;

// src/services/workflow/agentOrchestrator.ts
import { db } from '../../infra/db';
import crypto from 'crypto';

interface PendingToolCall {
  toolName: string;
  parameters: Record<string, unknown>;
  riskLevel: 'LOW' | 'HIGH';
}

export class ResumableAgentWorkflow {
  /**
   * Evaluates proposed agent tool call and pauses execution if high-risk.
   */
  public async handleToolInvocation(
    workflowId: string,
    tenantId: string,
    toolCall: PendingToolCall,
    currentState: Record<string, unknown>
  ): Promise<{ status: 'COMPLETED' | 'SUSPENDED'; result?: unknown }> {
    // 1. Low-risk queries (read-only) execute immediately
    if (toolCall.riskLevel === 'LOW') {
      const result = await this.executeTool(toolCall);
      return { status: 'COMPLETED', result };
    }

    // 2. High-risk mutations generate a secure approval token and suspend
    const approvalToken = crypto.randomBytes(32).toString('hex');

    await db('hitl_workflows')
      .where({ id: workflowId, tenant_id: tenantId })
      .update({
        status: 'suspended_for_approval',
        current_step: toolCall.toolName,
        state_payload: JSON.stringify(currentState),
        pending_action: JSON.stringify(toolCall),
        approval_token: approvalToken,
        updated_at: new Date(),
      });

    // 3. Dispatch out-of-band notification (Slack, Email, Webhook)
    await this.notifyApprover(tenantId, workflowId, toolCall, approvalToken);

    return { status: 'SUSPENDED' };
  }

  /**
   * Resumes workflow execution upon human sign-off.
   */
  public async resumeWorkflow(
    approvalToken: string,
    decision: 'APPROVE' | 'REJECT',
    actorUserId: string
  ): Promise<{ success: boolean; message: string }> {
    const workflow = await db('hitl_workflows')
      .where({ approval_token: approvalToken, status: 'suspended_for_approval' })
      .first();

    if (!workflow) {
      throw new Error('Invalid or expired approval token.');
    }

    if (decision === 'REJECT') {
      await db('hitl_workflows')
        .where({ id: workflow.id })
        .update({
          status: 'rejected',
          approval_token: null,
          updated_at: new Date(),
        });
      return { success: true, message: 'Workflow rejected by reviewer.' };
    }

    // Execute the frozen tool action with confirmed parameters
    const pendingTool: PendingToolCall = workflow.pending_action;
    const executionResult = await this.executeTool(pendingTool);

    // Advance state and resume execution
    await db('hitl_workflows')
      .where({ id: workflow.id })
      .update({
        status: 'completed',
        state_payload: JSON.stringify({
          ...workflow.state_payload,
          [pendingTool.toolName]: executionResult,
          approved_by: actorUserId,
        }),
        pending_action: null,
        approval_token: null,
        updated_at: new Date(),
      });

    return { success: true, message: 'Tool executed and workflow completed.' };
  }

  private async executeTool(tool: PendingToolCall): Promise<unknown> {
    // Concrete side-effect execution (e.g., Stripe API, CRM Update, Email Dispatch)
    return { executed: tool.toolName, timestamp: new Date().toISOString() };
  }

  private async notifyApprover(
    tenantId: string,
    workflowId: string,
    tool: PendingToolCall,
    token: string
  ): Promise<void> {
    // Integration with enterprise webhooks or notification channels
  }
}