Human-in-the-Loop Outreach Queue

Overview

AI-generated outreach — cold emails, follow-up messages, LinkedIn invitations — can be remarkably effective, but sending messages without human review creates compliance, brand, and relationship risks that most organizations cannot accept. Human-in-the-loop (HITL) outreach queues solve this by inserting an approval gate between the AI generation step and the delivery step: the workflow engine generates a candidate message, attaches approval metadata and sequence context, and parks it in a review queue where a sales rep or marketing coordinator can approve, edit, reject, or reschedule before the message goes out.

The queue is more than a simple inbox. Each queued item carries rich metadata: the workflow run ID that produced it, the sequence position (e.g., “step 3 of 5 in the onboarding drip”), the AI model and prompt that generated the draft, a confidence score if one was computed, and a failure trace if a prior delivery attempt bounced or timed out. This metadata turns the queue into an audit surface — reviewers can see not just what the AI wrote, but why and in what context, and operators can trace delivery failures back to specific workflow nodes.

Architecturally, the queue decouples the pace of AI generation from the pace of human review. Workflows can run asynchronously and batch-generate dozens of messages overnight; reviewers process them during business hours. The queue absorbs the impedance mismatch, and timeout policies ensure that stale messages are auto-expired or escalated rather than sitting indefinitely.

Core Concepts

How It Works

Workflow Emits Queue Items

When a workflow execution reaches a queue step node, the execution engine serializes the current context — including the AI-generated message, recipient data, and sequence metadata — into a queue item record and inserts it into the outreach_queue table. The workflow run transitions to a suspended state, recording the node ID and run context so execution can resume from the exact point where it paused. The queue insertion and run suspension happen within a single database transaction to prevent orphaned queue items or phantom run states.

Review Interface & Actions

The review UI presents queue items grouped by campaign or recipient, with the generated message body, sequence position, and AI metadata prominently displayed. Reviewers can take four actions: Approve (queue for immediate delivery), Edit & Approve (modify the message body in-line and then approve), Reject (cancel this message and optionally halt the entire sequence), or Snooze (delay review by N hours). Each action writes an audit log entry and publishes an event that the workflow engine consumes to advance or halt the run.

Delivery & Acknowledgement

Approved items are picked up by a delivery worker that sends the message via the configured channel (SMTP, API, webhook). The worker updates the queue item to sent on success or records a failure trace and transitions to retry_pending on error. For retries, the worker applies exponential backoff and respects a maximum retry count before marking the item as failed. The workflow engine listens for terminal states (sent or failed) to resume the suspended run and continue to the next sequence step.

Implementation Patterns

Queue Item Schema & State Machine

Define the queue item type with all metadata fields and a strict state machine for lifecycle transitions.

// types/outreach-queue.ts

export type QueueItemStatus =
  | "pending"
  | "approved"
  | "edited"
  | "rejected"
  | "sent"
  | "retry_pending"
  | "failed"
  | "expired";

export interface DeliveryAttempt {
  timestamp: string;       // ISO 8601
  channel: "email" | "sms" | "webhook";
  success: boolean;
  errorCode?: string;      // e.g., "550_BOUNCE", "RATE_LIMITED"
  errorMessage?: string;
}

export interface ApprovalRecord {
  action: "approve" | "edit" | "reject" | "snooze" | "escalate";
  actor: string;           // reviewer user ID
  timestamp: string;
  comment?: string;
  editedBody?: string;     // populated only for "edit" action
}

export interface OutreachQueueItem {
  id: string;
  workflowRunId: string;
  nodeId: string;          // the queue_step node that emitted this item

  // Message content
  recipientId: string;
  recipientEmail: string;
  subject: string;
  body: string;            // AI-generated draft
  channel: "email" | "sms" | "webhook";

  // Sequence context
  sequenceName: string;
  sequenceStep: number;    // 1-indexed
  totalSteps: number;
  delayBeforeNextMs: number;

  // AI provenance
  aiModel: string;
  promptHash: string;      // SHA-256 of the prompt template + variables
  confidenceScore?: number;

  // Lifecycle
  status: QueueItemStatus;
  approvals: ApprovalRecord[];
  deliveryAttempts: DeliveryAttempt[];

  // Timing
  createdAt: string;
  timeoutAt: string;       // auto-expire if still pending after this
  updatedAt: string;
}

// Valid state transitions
const VALID_TRANSITIONS: Record<QueueItemStatus, QueueItemStatus[]> = {
  pending:       ["approved", "edited", "rejected", "expired"],
  edited:        ["approved", "rejected"],
  approved:      ["sent", "retry_pending"],
  retry_pending: ["sent", "failed"],
  sent:          [],
  rejected:      [],
  failed:        [],
  expired:       [],
};

export function canTransition(from: QueueItemStatus, to: QueueItemStatus): boolean {
  return VALID_TRANSITIONS[from]?.includes(to) ?? false;
}

Queue Service with Approval & Delivery

The service layer handles enqueueing from workflows, processing reviewer actions, and coordinating delivery with retry logic.

// services/outreach-queue.ts
import type {
  OutreachQueueItem,
  QueueItemStatus,
  ApprovalRecord,
  DeliveryAttempt,
} from "../types/outreach-queue";
import { canTransition } from "../types/outreach-queue";

interface QueueStore {
  insert(item: OutreachQueueItem): Promise<void>;
  update(id: string, fields: Partial<OutreachQueueItem>): Promise<void>;
  findPending(limit: number): Promise<OutreachQueueItem[]>;
  findApproved(limit: number): Promise<OutreachQueueItem[]>;
  findExpired(now: string): Promise<OutreachQueueItem[]>;
}

interface DeliveryChannel {
  send(item: OutreachQueueItem): Promise<{ success: boolean; errorCode?: string; errorMessage?: string }>;
}

interface WorkflowResumeService {
  resume(runId: string, nodeId: string, result: "sent" | "failed" | "rejected"): Promise<void>;
}

export class OutreachQueueService {
  constructor(
    private store: QueueStore,
    private delivery: DeliveryChannel,
    private workflows: WorkflowResumeService,
    private maxRetries = 3,
    private timeoutHours = 48
  ) {}

  /** Called by the workflow engine when a queue_step node executes */
  async enqueue(params: {
    workflowRunId: string;
    nodeId: string;
    recipientId: string;
    recipientEmail: string;
    subject: string;
    body: string;
    channel: "email" | "sms" | "webhook";
    sequenceName: string;
    sequenceStep: number;
    totalSteps: number;
    delayBeforeNextMs: number;
    aiModel: string;
    promptHash: string;
    confidenceScore?: number;
  }): Promise<string> {
    const now = new Date().toISOString();
    const timeoutAt = new Date(
      Date.now() + this.timeoutHours * 60 * 60 * 1000
    ).toISOString();

    const item: OutreachQueueItem = {
      id: crypto.randomUUID(),
      ...params,
      status: "pending",
      approvals: [],
      deliveryAttempts: [],
      createdAt: now,
      timeoutAt,
      updatedAt: now,
    };

    await this.store.insert(item);
    return item.id;
  }

  /** Process a reviewer action */
  async review(
    itemId: string,
    item: OutreachQueueItem,
    action: ApprovalRecord
  ): Promise<void> {
    let newStatus: QueueItemStatus;
    switch (action.action) {
      case "approve":
        newStatus = "approved";
        break;
      case "edit":
        newStatus = "edited";
        break;
      case "reject":
        newStatus = "rejected";
        break;
      case "snooze":
        // Extend timeout, stay pending
        const extended = new Date(
          Date.now() + 24 * 60 * 60 * 1000
        ).toISOString();
        await this.store.update(itemId, {
          timeoutAt: extended,
          approvals: [...item.approvals, action],
          updatedAt: new Date().toISOString(),
        });
        return;
      default:
        throw new Error(`Unknown action: ${action.action}`);
    }

    if (!canTransition(item.status, newStatus)) {
      throw new Error(
        `Invalid transition: ${item.status} -> ${newStatus}`
      );
    }

    const updates: Partial<OutreachQueueItem> = {
      status: newStatus,
      approvals: [...item.approvals, action],
      updatedAt: new Date().toISOString(),
    };
    if (action.editedBody) {
      updates.body = action.editedBody;
    }

    await this.store.update(itemId, updates);

    if (newStatus === "rejected") {
      await this.workflows.resume(
        item.workflowRunId, item.nodeId, "rejected"
      );
    }
  }

  /** Delivery worker: pick approved items and send them */
  async processDeliveryBatch(batchSize = 10): Promise<void> {
    const items = await this.store.findApproved(batchSize);

    for (const item of items) {
      const result = await this.delivery.send(item);
      const attempt: DeliveryAttempt = {
        timestamp: new Date().toISOString(),
        channel: item.channel,
        success: result.success,
        errorCode: result.errorCode,
        errorMessage: result.errorMessage,
      };

      if (result.success) {
        await this.store.update(item.id, {
          status: "sent",
          deliveryAttempts: [...item.deliveryAttempts, attempt],
          updatedAt: new Date().toISOString(),
        });
        await this.workflows.resume(
          item.workflowRunId, item.nodeId, "sent"
        );
      } else {
        const retryCount = item.deliveryAttempts.length + 1;
        const newStatus: QueueItemStatus =
          retryCount >= this.maxRetries ? "failed" : "retry_pending";

        await this.store.update(item.id, {
          status: newStatus,
          deliveryAttempts: [...item.deliveryAttempts, attempt],
          updatedAt: new Date().toISOString(),
        });

        if (newStatus === "failed") {
          await this.workflows.resume(
            item.workflowRunId, item.nodeId, "failed"
          );
        }
      }
    }
  }

  /** Timeout sweep: expire stale pending items */
  async expireStale(): Promise<void> {
    const now = new Date().toISOString();
    const expired = await this.store.findExpired(now);

    for (const item of expired) {
      await this.store.update(item.id, {
        status: "expired",
        updatedAt: now,
      });
      await this.workflows.resume(
        item.workflowRunId, item.nodeId, "failed"
      );
    }
  }
}

Practical Applications

In the merchant dashboard platform, workflow runs generate AI-drafted outreach messages for sales prospects and park them in a human review queue before delivery. Each queued message carries the workflow run context, the prospect’s CRM record, and the sequence step so that sales reps reviewing the queue have full context without switching between tools. The approval metadata — who approved, when, and whether they edited the draft — feeds back into analytics dashboards that measure AI draft quality over time and identify prompts that consistently require human edits.

The failure trace system handles a real operational pain point: when an email bounces or a webhook delivery times out, the system captures the structured error and surfaces it in the queue UI alongside the message. Reps can decide whether to retry on the same channel, switch to an alternative (e.g., from email to LinkedIn), or mark the contact as unreachable. This per-message delivery intelligence replaces the older pattern of bulk-sending with post-hoc bounce processing, catching delivery issues before they compound across an entire campaign.

Common Pitfalls

• Sending stale messages after long review delays — A message generated on Monday that sits in the queue until Friday may reference outdated context (“I saw you visited our pricing page today”). Implement a staleness check at delivery time that compares the message age against a configurable threshold and routes expired messages back to regeneration rather than sending them verbatim.
• Losing workflow run context on queue item expiration — When a queue item expires, the workflow run is still suspended at the queue step node. If expiration only updates the queue item without notifying the workflow engine, the run hangs indefinitely. Always publish a terminal event (expired/failed) that the engine can consume to resume or terminate the run.
• Race conditions between reviewer actions and timeout sweeps — A reviewer clicks “Approve” at the same instant the timeout sweep marks the item as “Expired”. Without optimistic concurrency control (e.g., a version column or CAS operation), both writes succeed and the item ends up in an inconsistent state. Use database-level locking or compare-and-swap on the status field to prevent conflicting transitions.
• Unbounded queue growth from high-volume campaigns — A workflow that fans out across 10,000 contacts generates 10,000 queue items that human reviewers cannot possibly process in time. Implement batch-approval features (approve all items from a specific campaign or matching a confidence threshold) and auto-approve policies for high-confidence drafts to keep the queue manageable.
• Missing audit trail for edited messages — When a reviewer edits a message before approving, the original AI-generated draft must be preserved alongside the edited version. Overwriting the body field without storing the original loses the ability to compare AI output versus human edits, which is essential for prompt improvement feedback loops.