Async AI: Long-Running Agents and Event-Driven Workflows

A video generation agent, a deep research task, a codebase analysis — these take minutes, not milliseconds. Trying to fit them into a single HTTP request causes timeouts, retries, and bad UX. Async AI patterns decouple execution from waiting, letting long-running agents complete reliably regardless of network conditions or client timeouts.

Why Synchronous AI Fails for Long Tasks

Client → HTTP request → Agent starts (10 seconds)...
                                     ...still running (30 seconds)...
                                          ← Connection timeout (60s limit)
Client gets: 504 Gateway Timeout
Agent keeps running (user doesn't know)
Agent finishes → result lost

HTTP requests have timeouts (30–120s for most configurations). LLM-heavy workflows routinely exceed this. The solution: async execution with a job ID the client can use to check progress.

Pattern 1: Job Queue + Polling

The simplest async pattern. User submits a job, gets an ID, polls for completion.

API Design

// POST /ai/jobs → returns job ID immediately
// GET /ai/jobs/:id → returns job status + result when complete

interface Job {
  id: string;
  status: "queued" | "processing" | "completed" | "failed";
  type: string;
  input: unknown;
  result?: unknown;
  error?: string;
  createdAt: number;
  startedAt?: number;
  completedAt?: number;
  progress?: { step: number; total: number; message: string };
}

API Routes (Next.js / Express)

// app/api/ai/jobs/route.ts
export async function POST(req: Request) {
  const { type, input } = await req.json();

  const job: Job = {
    id: crypto.randomUUID(),
    status: "queued",
    type,
    input,
    createdAt: Date.now(),
  };

  await db.saveJob(job);
  await jobQueue.enqueue({ jobId: job.id });

  return Response.json({ jobId: job.id }, { status: 202 });
}

// app/api/ai/jobs/[id]/route.ts
export async function GET(req: Request, { params }: { params: { id: string } }) {
  const job = await db.getJob(params.id);
  if (!job) return new Response("Not Found", { status: 404 });
  return Response.json(job);
}

Worker Process

// worker.ts — runs separately from the API server
import Queue from "bull";

const queue = new Queue("ai-jobs", { redis: process.env.REDIS_URL });

queue.process(async (job) => {
  const { jobId } = job.data;
  const jobRecord = await db.getJob(jobId);

  await db.updateJob(jobId, { status: "processing", startedAt: Date.now() });

  try {
    const result = await runAITask(jobRecord.type, jobRecord.input, {
      onProgress: async (step, total, message) => {
        await db.updateJob(jobId, {
          progress: { step, total, message },
        });
      },
    });

    await db.updateJob(jobId, {
      status: "completed",
      result,
      completedAt: Date.now(),
    });
  } catch (err: any) {
    await db.updateJob(jobId, {
      status: "failed",
      error: err.message,
      completedAt: Date.now(),
    });
  }
});

Client-Side Polling

async function pollForResult(jobId: string, maxWaitMs = 300_000): Promise<Job> {
  const startTime = Date.now();
  let interval = 1000; // start polling every 1s

  while (Date.now() - startTime < maxWaitMs) {
    const response = await fetch(`/api/ai/jobs/${jobId}`);
    const job: Job = await response.json();

    if (job.status === "completed") return job;
    if (job.status === "failed") throw new Error(job.error);

    // Show progress
    if (job.progress) {
      console.log(`${job.progress.step}/${job.progress.total}: ${job.progress.message}`);
    }

    // Exponential backoff: 1s → 2s → 4s → max 10s
    await sleep(interval);
    interval = Math.min(interval * 1.5, 10_000);
  }

  throw new Error("Job timed out after waiting");
}

Pattern 2: Webhooks (Push Instead of Poll)

For server-to-server integrations, push the result instead of making the client poll:

interface WebhookConfig {
  url: string;
  secret: string;      // for HMAC signature verification
  events: string[];    // "job.completed", "job.failed", "job.progress"
}

async function deliverWebhook(config: WebhookConfig, event: string, payload: unknown) {
  const body = JSON.stringify({ event, payload, timestamp: Date.now() });
  const signature = createHmac("sha256", config.secret).update(body).digest("hex");

  const response = await fetch(config.url, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-Webhook-Event": event,
      "X-Webhook-Signature": `sha256=${signature}`,
    },
    body,
    signal: AbortSignal.timeout(10_000),
  });

  if (!response.ok) {
    // Retry webhook delivery with backoff
    await scheduleRetry(config, event, payload);
  }
}

// Receiving end: verify signature
function verifyWebhookSignature(body: string, signature: string, secret: string): boolean {
  const expected = `sha256=${createHmac("sha256", secret).update(body).digest("hex")}`;
  return timingSafeEqual(Buffer.from(signature), Buffer.from(expected));
}

Pattern 3: Temporal for Durable Workflows

For complex multi-step AI workflows, Temporal provides durable execution — if the worker crashes mid-task, Temporal replays from the last checkpoint.

// workflow.go — runs on a Temporal worker
package workflows

import (
    "time"
    "go.temporal.io/sdk/workflow"
    "go.temporal.io/sdk/activity"
)

type ResearchWorkflowInput struct {
    Topic  string `json:"topic"`
    Depth  int    `json:"depth"`
    UserID string `json:"userId"`
}

func ResearchWorkflow(ctx workflow.Context, input ResearchWorkflowInput) (string, error) {
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            MaximumAttempts: 3,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    // Step 1: Plan research
    var plan ResearchPlan
    if err := workflow.ExecuteActivity(ctx, PlanResearchActivity, input.Topic).Get(ctx, &plan); err != nil {
        return "", err
    }

    // Step 2: Execute sub-tasks in parallel
    var sources []string
    futures := make([]workflow.Future, len(plan.SubTopics))
    for i, subtopic := range plan.SubTopics {
        futures[i] = workflow.ExecuteActivity(ctx, SearchAndSummarizeActivity, subtopic)
    }
    for _, future := range futures {
        var result string
        if err := future.Get(ctx, &result); err == nil {
            sources = append(sources, result)
        }
    }

    // Step 3: Synthesize final report
    var report string
    if err := workflow.ExecuteActivity(ctx, SynthesizeReportActivity, sources, input.Topic).Get(ctx, &report); err != nil {
        return "", err
    }

    return report, nil
}

The QuantumSketch video generation pipeline at Shahriar Labs uses exactly this pattern — 4 activities (script, scenes, render, merge) in a Temporal workflow. If the Manim render crashes, Temporal replays from the render step automatically.

Pattern 4: SSE for Real-Time Progress

For web apps, Server-Sent Events push progress updates without polling:

// API route: GET /api/ai/jobs/:id/stream
export async function GET(req: Request, { params }: { params: { id: string } }) {
  const jobId = params.id;

  const stream = new ReadableStream({
    async start(controller) {
      const sendEvent = (event: string, data: unknown) => {
        controller.enqueue(
          new TextEncoder().encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
        );
      };

      // Poll job status and push events
      let attempts = 0;
      while (attempts < 300) { // max 5 minutes
        const job = await db.getJob(jobId);
        if (!job) { sendEvent("error", { message: "Job not found" }); break; }

        sendEvent("progress", {
          status: job.status,
          progress: job.progress,
        });

        if (job.status === "completed") {
          sendEvent("completed", { result: job.result });
          break;
        }
        if (job.status === "failed") {
          sendEvent("failed", { error: job.error });
          break;
        }

        await sleep(1000);
        attempts++;
      }

      controller.close();
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive",
    },
  });
}

// Client:
const eventSource = new EventSource(`/api/ai/jobs/${jobId}/stream`);
eventSource.addEventListener("progress", (e) => updateProgressUI(JSON.parse(e.data)));
eventSource.addEventListener("completed", (e) => showResult(JSON.parse(e.data)));
eventSource.addEventListener("failed", (e) => showError(JSON.parse(e.data)));

Choosing the Right Pattern

| Scenario | Pattern | |---------|---------| | Task <30s, web client | Streaming response (no async needed) | | Task 30s–5min, web client | SSE progress + job API | | Task 5min+, web client | Job queue + polling | | Server-to-server integration | Webhooks | | Complex multi-step workflow | Temporal | | Any pattern that needs durability | Temporal |

FAQ

How do I handle job expiry? Set a TTL on job records (7 days is standard). After TTL, delete the job and its result. Notify users before deletion for large outputs.

What if the client disconnects mid-stream and misses progress? Use the polling fallback: clients that disconnect reconnect and poll /api/ai/jobs/:id. The job record always reflects current state.

How do I prevent duplicate job submissions? Accept a client-provided idempotency key. Store idemp_key → job_id in Redis with a 24-hour TTL. If the same key arrives again, return the existing job ID.

Is Temporal worth the operational overhead for smaller projects? For simple 2-step workflows: no, use Bull/BullMQ. For 5+ step workflows with dependencies, retries, and timeouts: yes, Temporal's durability guarantees are worth it. The managed Temporal Cloud removes ops overhead.

Can I use this pattern for real-time chat? No — chat is inherently synchronous (user types, model responds). Use streaming responses for chat. Async patterns are for background processing where the user submits a task and checks back later.