SQS FIFO + Lambda: Ordered Agent Job Queues Without Double-Applies

Coding agents enqueue “apply patch”, “run tests”, “open PR” as jobs. A standard SQS queue + Lambda concurrency > 1 will reorder and overlap jobs for the same workspace — patch B lands before patch A, tests run on a half-applied tree, and visibility timeout retries double-apply. The unfair advantage is SQS FIFO with MessageGroupId = workspaceId, content-based or explicit deduplication, and handlers that stay idempotent when Lambda retries after a slow git apply.

⚡ TL;DR: Use FIFO for per-tenant/per-workspace agent jobs. Set MessageGroupId to the ordering key (never a constant). Cap in-flight per group naturally via FIFO; use Lambda partial batch failure. Dedup windows are 5 minutes — still add DynamoDB idempotency for side effects. Pair with Exactly-Once Illusions, EventBridge Pipes to Lambda, and Bedrock Agents Plus EventBridge.

Why standard queues break agent workspaces

// ❌ Parallel Lambda on standard SQS — race on same workspace
export async function handler(event: SQSEvent) {
  for (const rec of event.Records) {
    const job = JSON.parse(rec.body);
    await applyPatch(job.workspaceId, job.diff); // overlapping applies corrupt git
  }
}

FIFO gives you:

1. Order within a MessageGroupId
2. At-most-one in-flight consumer per group (for the group’s head)
3. Deduplication (content-based or MessageDeduplicationId) within ~5 minutes

It does not give exactly-once side effects after the dedup window or across different group IDs.

Model jobs with group + dedup ids

import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
import { createHash } from "crypto";

const sqs = new SQSClient({});
const QUEUE_URL = process.env.AGENT_JOBS_FIFO_URL!;

export type AgentJob = {
  jobId: string;
  workspaceId: string;
  tenantId: string;
  type: "apply_patch" | "run_tests" | "open_pr";
  payload: Record<string, unknown>;
};

export async function enqueueAgentJob(job: AgentJob) {
  // ✅ Group by workspace — order preserved per repo sandbox
  const groupId = `ws#${job.workspaceId}`;

  // ✅ Stable dedup for producer retries (same jobId)
  const dedupId = createHash("sha256")
    .update(`${job.jobId}:${job.type}`)
    .digest("hex")
    .slice(0, 128);

  await sqs.send(
    new SendMessageCommand({
      QueueUrl: QUEUE_URL,
      MessageBody: JSON.stringify(job),
      MessageGroupId: groupId,
      MessageDeduplicationId: dedupId,
      MessageAttributes: {
        tenantId: { DataType: "String", StringValue: job.tenantId },
        type: { DataType: "String", StringValue: job.type },
      },
    })
  );
}

❌ Using MessageGroupId: "global" serializes your entire platform. ❌ Random dedup ids on every producer retry create duplicates.

Lambda event source mapping for FIFO

// CDK-ish
const fn = new lambda.Function(this, "AgentJobWorker", {
  runtime: lambda.Runtime.NODEJS_20_X,
  handler: "worker.handler",
  timeout: cdk.Duration.minutes(2),
  reservedConcurrentExecutions: 50, // total, not per group
  code: lambda.Code.fromAsset("dist"),
});

fn.addEventSource(
  new sources.SqsEventSource(fifoQueue, {
    batchSize: 5, // FIFO still respects group ordering within batch semantics
    reportBatchItemFailures: true, // ✅ critical
    maxConcurrency: 20, // scaling config for the ESM
  })
);

Notes that bite agents:

• Visibility timeout must exceed worst-case job time (git apply + tests). If apply takes 90s and visibility is 30s, another delivery starts → double-apply risk without idempotency.
• High batchSize with mixed groups is OK; never process out-of-order within a group in your own code — process records sequentially per group if you fan out inside the handler.

import type { SQSEvent, SQSBatchResponse } from "aws-lambda";

export async function handler(event: SQSEvent): Promise<SQSBatchResponse> {
  const failures: { itemIdentifier: string }[] = [];

  // Optional: group records by MessageGroupId for clarity
  for (const rec of event.Records) {
    try {
      const job = JSON.parse(rec.body) as AgentJob;
      await processJobOnce(job, rec.messageId);
    } catch (e) {
      console.error("job_failed", rec.messageId, e);
      failures.push({ itemIdentifier: rec.messageId });
    }
  }
  return { batchItemFailures: failures };
}

Idempotency beyond the 5-minute dedup window

FIFO dedup will not save you when:

• The same logical patch is enqueued again an hour later with a new jobId
• Visibility timeout expires mid-handler and SQS redelivers (same messageId / approximate receive count)

import {
  DynamoDBClient,
  PutItemCommand,
  GetItemCommand,
} from "@aws-sdk/client-dynamodb";

const ddb = new DynamoDBClient({});
const TABLE = process.env.JOB_IDEMPOTENCY_TABLE!;

async function processJobOnce(job: AgentJob, sqsMessageId: string) {
  const pk = `job#${job.jobId}`;

  const prior = await ddb.send(
    new GetItemCommand({ TableName: TABLE, Key: { pk: { S: pk } } })
  );
  if (prior.Item?.status?.S === "SUCCEEDED") {
    return; // ✅ replay no-op
  }

  try {
    await ddb.send(
      new PutItemCommand({
        TableName: TABLE,
        Item: {
          pk: { S: pk },
          status: { S: "IN_FLIGHT" },
          sqsMessageId: { S: sqsMessageId },
          workspaceId: { S: job.workspaceId },
          ttl: { N: String(Math.floor(Date.now() / 1000) + 7 * 86400) },
        },
        ConditionExpression: "attribute_not_exists(pk) OR #s <> :ok",
        ExpressionAttributeNames: { "#s": "status" },
        ExpressionAttributeValues: { ":ok": { S: "SUCCEEDED" } },
      })
    );
  } catch {
    return; // lost race — other worker owns it
  }

  switch (job.type) {
    case "apply_patch":
      await applyPatch(job.workspaceId, job.payload);
      break;
    case "run_tests":
      await runTests(job.workspaceId, job.payload);
      break;
    case "open_pr":
      await openPr(job.workspaceId, job.payload);
      break;
  }

  await ddb.send(
    new PutItemCommand({
      TableName: TABLE,
      Item: {
        pk: { S: pk },
        status: { S: "SUCCEEDED" },
        sqsMessageId: { S: sqsMessageId },
        workspaceId: { S: job.workspaceId },
        ttl: { N: String(Math.floor(Date.now() / 1000) + 30 * 86400) },
      },
    })
  );
}

async function applyPatch(workspaceId: string, payload: Record<string, unknown>) {
  /* git apply — workspace-scoped lock optional */
}
async function runTests(workspaceId: string, payload: Record<string, unknown>) {}
async function openPr(workspaceId: string, payload: Record<string, unknown>) {}

For schema-strict job payloads from agents, align with AI Coding Agent Tool Schemas. If jobs are produced from EventBridge, prefer Pipes with fail-closed filters rather than a custom router — EventBridge Pipes to Lambda.

Throughput design for many workspaces

FIFO throughput is per message group high-throughput mode (when enabled) and account quotas. Design:

• Thousands of workspaces → thousands of group IDs → natural parallelism
• Hot workspace (one huge monorepo) → intentional serialization (good — protects git)
• Cross-workspace fan-out (notify, billing) → separate standard queue so FIFO agent jobs stay lean

Wire DLQ as FIFO too if you need ordered redrive; otherwise standard DLQ is fine for poison messages after maxReceiveCount.

Poison messages and agent DLQ playbooks

After maxReceiveCount, jobs land on the DLQ. For agent platforms, store the original jobId / workspaceId in message attributes so on-call can redrive without parsing opaque bodies. Never auto-redrive mutators blindly — inspect whether the idempotency row is IN_FLIGHT stuck (worker crashed) vs FAILED (bad diff). Clear or fix the claim row, then redrive once.

Checklist

• [ ] Queue is .fifo with content-based-deduplication or explicit MessageDeduplicationId
• [ ] MessageGroupId = workspace/tenant ordering key (not a constant)
• [ ] Lambda ESM: reportBatchItemFailures: true
• [ ] Visibility timeout > p99 job duration (+ buffer)
• [ ] DynamoDB (or equivalent) idempotency on jobId for side effects
• [ ] Hot-path agent jobs not mixed with bulk analytics on same FIFO queue
• [ ] DLQ + alarm on age / receive count; redrive playbook documented