Build a Dead-Letter Queue Triage Agent: Cluster SQS DLQ Messages by Failure Signature, Redrive the Transient Ones, and Quarantine Poison Messages Without Replaying the Outage

What This Agent Does

A dead-letter queue triage agent samples the messages sitting in an SQS DLQ, joins each one to the consumer log line that explains why it failed, clusters the failures into a handful of signatures, and decides per signature whether the messages should go back to the source queue, go to a quarantine queue for a human, or stay put until a fix ships. It redrives with the native StartMessageMoveTask API at a rate the consumer can absorb, and it cancels the move the moment the consumer's error rate climbs. It never calls SendMessage on the source queue directly, never redrives while the DLQ is still filling, and never replays a message whose consumer is not declared idempotent.

The reason to build this instead of pressing the console's Redrive button is that the button has no judgement. A DLQ after an incident holds a mix: a few thousand messages that failed because a downstream was down for twenty minutes, a few hundred that fail every single time because the producer shipped a schema change, and a dozen that reference an entity someone deleted. Redriving all of them together sends the poison messages straight back through the consumer, into the DLQ again, and the retry loop eats the error budget twice. Deleting all of them loses real orders. Someone has to sort them, and sorting ten thousand JSON bodies by hand is exactly the toil an agent should take.

Step 1: Peek Without Consuming, and Strip the Payload

The sampler reads the DLQ with a visibility timeout of zero so every message lands back in the queue immediately. Long polling matters here: a short poll on a standard queue only asks a subset of SQS servers, so a one-second wait is what makes the sample representative.

# dlq/sample.py
import json, re, boto3

sqs = boto3.client("sqs")

def shape(body: str) -> dict:
    """Keep keys and types, drop values. The model never sees payload data."""
    try:
        obj = json.loads(body)
    except ValueError:
        return {"_raw": "non-json", "_len": len(body)}
    if not isinstance(obj, dict):
        return {"_type": type(obj).__name__}
    return {k: (shape(json.dumps(v)) if isinstance(v, dict) else type(v).__name__)
            for k, v in obj.items()}

def sample_dlq(queue_url: str, n: int = 200) -> list[dict]:
    seen, out, idle = set(), [], 0
    while len(out) < n and idle < 5:
        resp = sqs.receive_message(
            QueueUrl=queue_url, MaxNumberOfMessages=10,
            VisibilityTimeout=0, WaitTimeSeconds=1,
            AttributeNames=["ApproximateReceiveCount", "SentTimestamp"],
            MessageAttributeNames=["All"],
        )
        msgs = resp.get("Messages", [])
        idle = idle + 1 if not msgs else 0
        for m in msgs:
            if m["MessageId"] in seen:
                continue
            seen.add(m["MessageId"])
            out.append({
                "id": m["MessageId"],
                "sent_at": int(m["Attributes"]["SentTimestamp"]) // 1000,
                "receives": int(m["Attributes"]["ApproximateReceiveCount"]),
                "shape": shape(m["Body"]),
                "attrs": {k: v.get("StringValue")
                          for k, v in m.get("MessageAttributes", {}).items()},
            })
    return out

Two things are deliberate. The model gets the shape of the body, never the body, because DLQs are full of customer records and a triage prompt is not a place for them. And the sampler records ApproximateReceiveCount before anything else touches the queue, because every peek increments it. Two hundred messages is enough to find every signature that holds more than about two percent of the queue. Rarer signatures are the honest limit at the end of this post.

Step 2: Join Each Message to the Error That Put It There

The DLQ message does not say why it failed. The consumer's logs do, provided the consumer logs the SQS message ID on every failure, which is the one instrumentation requirement this agent imposes. With CloudWatch Logs Insights the join is a single query per batch of IDs.

# dlq/errors.py
import re, time, boto3

logs = boto3.client("logs")

QUERY = """
fields @timestamp, sqs_message_id, error_class, error_message
| filter sqs_message_id in [{ids}]
| sort @timestamp desc
"""

NOISE = [
    (re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"), "<uuid>"),
    (re.compile(r"\b[0-9a-f]{12,}\b"), "<hex>"),
    (re.compile(r"\b\d{5,}\b"), "<n>"),
]

def signature(error_class: str, message: str) -> str:
    for rx, repl in NOISE:
        message = rx.sub(repl, message)
    return f"{error_class}: {message[:160]}"

def consumer_errors(log_group: str, ids: list[str], hours: int = 72) -> dict[str, dict]:
    q = logs.start_query(
        logGroupName=log_group, startTime=int(time.time()) - hours * 3600,
        endTime=int(time.time()),
        queryString=QUERY.format(ids=",".join(f'"{i}"' for i in ids)), limit=10000)
    while True:
        r = logs.get_query_results(queryId=q["queryId"])
        if r["status"] in ("Complete", "Failed", "Cancelled"):
            break
        time.sleep(1)
    out = {}
    for row in r["results"]:
        f = {c["field"]: c["value"] for c in row}
        mid = f.get("sqs_message_id")
        if mid and mid not in out:          # newest failure per message wins
            out[mid] = {"class": f.get("error_class", "?"),
                        "sig": signature(f.get("error_class", "?"), f.get("error_message", ""))}
    return out

If your consumers run on Kubernetes and log to Loki, the same join is a LogQL query with a |= "sqs_message_id" filter and the identical normalization. The normalizer is what turns four thousand distinct NotFound: customer 8d1c... lines into one signature, and one signature is one decision.

Group the sample by signature and the DLQ stops being a number and becomes a table:

The last row is a real category. Messages older than the consumer's log retention, or that failed inside a crash the logger never flushed, have no evidence. The agent reports them and refuses to classify them.

Step 3: The Model Decides Per Signature, Not Per Message

The model sees one signature at a time with its share, receive counts, age range, body shape, the consumer's declared idempotency, and whether the downstream named in the error is currently healthy. It returns a strict schema.

VERDICT_SCHEMA = {
    "type": "object",
    "properties": {
        "verdict": {"type": "string", "enum": [
            "redrive", "quarantine", "redrive_after_fix", "needs_human"]},
        "root_cause": {"type": "string"},
        "evidence": {"type": "string"},
        "what_breaks_if_wrong": {"type": "string"},
    },
    "required": ["verdict", "root_cause", "evidence", "what_breaks_if_wrong"],
}

SYSTEM = (
    "You triage one failure signature from an SQS dead-letter queue.\n"
    "redrive: the failure was transient (timeout, 5xx, throttling, dependency "
    "outage) AND the dependency is healthy now AND the consumer is idempotent.\n"
    "quarantine: the message itself is the problem (schema, parse, validation, "
    "entity that no longer exists). Replaying it will fail every time.\n"
    "redrive_after_fix: the message is valid but the consumer has a bug. Name "
    "the fix. Nothing moves until a human marks the fix deployed.\n"
    "needs_human: evidence is missing or contradictory, or the consumer is not "
    "idempotent. Never guess idempotency.\n"
    "Cite the signature, share, and health check you were given. Do not infer "
    "business impact from the body shape."
)

The idempotency fact comes from the service catalog, not from the model. The Backstage MCP server exposes an idempotent_consumer annotation per component along with the owner, and a consumer without the annotation is treated as non-idempotent. That rule alone turns the scariest DLQ mistake, replaying a payment message that already half-executed, into a needs_human with the owner's name attached.

The dependency health check is a wrapper lookup, not a model opinion. For ConnectTimeout: inventory-svc:8080 the wrapper queries the target's 5xx rate and p99 over the last fifteen minutes and passes a single boolean. A model told "dependency healthy: false" cannot return redrive, because the wrapper drops that verdict before it is acted on.

Step 4: Quarantine First, Then Move, Then Watch

StartMessageMoveTask moves every message in the DLQ. There is no filter. So the order of operations is the whole safety story: pull the poison and the unknowns out into a quarantine queue first, confirm the DLQ now holds only redrive-verdict signatures, and only then start the move.

# dlq/execute.py
import time, boto3
sqs = boto3.client("sqs")

def quarantine(dlq_url: str, q_url: str, target_ids: set[str], verdict: str) -> int:
    moved, empty = 0, 0
    while target_ids and empty < 5:
        r = sqs.receive_message(QueueUrl=dlq_url, MaxNumberOfMessages=10,
                                VisibilityTimeout=20, WaitTimeSeconds=1,
                                MessageAttributeNames=["All"])
        msgs = r.get("Messages", [])
        empty = empty + 1 if not msgs else 0
        for m in msgs:
            if m["MessageId"] not in target_ids:
                continue                      # hidden 20s, then back in the DLQ
            attrs = dict(m.get("MessageAttributes", {}))
            attrs["dlq_origin_id"] = {"DataType": "String", "StringValue": m["MessageId"]}
            attrs["dlq_verdict"] = {"DataType": "String", "StringValue": verdict}
            sqs.send_message(QueueUrl=q_url, MessageBody=m["Body"], MessageAttributes=attrs)
            sqs.delete_message(QueueUrl=dlq_url, ReceiptHandle=m["ReceiptHandle"])
            target_ids.discard(m["MessageId"]); moved += 1
    return moved

def redrive(dlq_arn: str, source_arn: str, per_second: int) -> str:
    t = sqs.start_message_move_task(SourceArn=dlq_arn, DestinationArn=source_arn,
                                    MaxNumberOfMessagesPerSecond=per_second)
    return t["TaskHandle"]

def watch(dlq_arn: str, handle: str, consumer_unhealthy) -> str:
    while True:
        # most recent task first; TaskHandle is only present while RUNNING
        task = sqs.list_message_move_tasks(SourceArn=dlq_arn, MaxResults=1)["Results"][0]
        if task["Status"] != "RUNNING":
            return task["Status"]
        if consumer_unhealthy():
            sqs.cancel_message_move_task(TaskHandle=handle)
            return "CANCELLED_BY_GUARD"
        time.sleep(15)

The quarantine step is send-then-delete, which is at-least-once: a crash between the two calls leaves a duplicate in quarantine, never a lost message. That is the right failure direction for a queue nobody consumes automatically. Quarantine is a plain standard queue with the maximum fourteen-day retention, no consumer, and an alarm on ApproximateAgeOfOldestMessage so it cannot become a second DLQ that everyone forgets.

The rate is sized from what the consumer actually did last hour: the wrapper reads NumberOfMessagesDeleted on the source queue, converts it to a per-second steady state, and redrives at a fifth of that, floored at one and capped at one hundred. A DLQ of fifty thousand messages redriven at full speed into a consumer that just recovered is how you cause the second outage. The watch loop is the circuit breaker: the same health check that gated the verdict runs every fifteen seconds during the move, and a cancelled task leaves the unmoved remainder in the DLQ for the next run, which is a far better place for it than the source queue.

The retention clock is the other reason to be quick about the quarantine step. A message's expiry is based on its original enqueue time, and moving to the DLQ does not reset it. If the source queue keeps messages four days and the DLQ was created with the same default, a message that failed on day three of its life expires silently from the DLQ a day later. The agent computes that deadline per signature, puts it in the report, and opens a one-line Terraform PR raising DLQ retention to fourteen days when the DLQ's period is not longer than its source's, which is the configuration AWS itself recommends.

The Approval Gate and the Run Record

The plan is one message in Slack, posted before anything moves, following the approval-gate pattern:

DLQ orders-dlq: 12,480 visible, oldest 4d 2h (expires in 1d 22h)
  redrive            8,860  ConnectTimeout: inventory-svc:8080      dependency healthy 38m
  quarantine         2,750  ValidationError: 'currency' is required  producer release v2.31
  quarantine           620  NotFound: customer <uuid>
  needs_human          250  no consumer log line within 72h
Consumer orders-worker: idempotent=true (catalog), steady 41 msg/s, errors 0.0% (15m)
Plan: quarantine 3,370 -> orders-quarantine, then move 8,860 at 8 msg/s (~18 min)
Approve: /dlq approve 7c1e   Deny: /dlq deny 7c1e

One approval covers one plan for one queue. The approver's identity, the plan hash, the task handle, and every SendMessage and DeleteMessage the wrapper issued are written to the run record described in the audit trail post, so a message that turns up in quarantine three weeks later can be traced to the verdict and the human that put it there. The producer release named in the validation row is the agent's reading of the deploy timeline, and the plan says so; it is a lead for the owner, not a finding.

Guardrails That Live in Code

The prompt expresses policy. The wrapper enforces it.

• No redrive while the DLQ is still filling. If ApproximateNumberOfMessagesVisible on the DLQ grew in the last ten minutes, the fault is ongoing and every verdict degrades to needs_human.
• No redrive into an unhealthy consumer. Error rate over one percent on the consumer, or a 5xx rate over one percent on the dependency named in the signature, blocks the move and cancels a running one.
• Non-idempotent means never automatic. Missing catalog annotation is treated as non-idempotent. The agent can still sort and quarantine, which is most of the value; it just cannot replay.
• Loop detection. A signature whose messages carry a receive count higher than the source queue's maxReceiveCount plus one has already been redriven once and failed again. Those go to quarantine regardless of what the model says.
• One move task per queue per hour, and a daily cap on messages moved. The circuit-breaker post covers why action budgets beat good intentions; a redrive loop that fires every five minutes is the failure mode that budget exists for.
• Least privilege that admits its limits. The role gets ReceiveMessage, DeleteMessage, GetQueueAttributes, and the move-task actions on the DLQ, and SendMessage on the quarantine queue. The move task also requires SendMessage on the source queue, so IAM cannot stop the role from writing arbitrary messages there. The guarantee that it only ever does so through a move task is a code-review guarantee, and the code is small enough to hold to that.

Run the agent in shadow mode for the first few incidents: it posts the plan, a human does the redrive by hand, and the team scores whether the signatures and verdicts matched what they found. The score to watch is quarantine precision. A transient signature wrongly quarantined costs a human ten minutes. A poison signature wrongly redriven costs a second round through the consumer and a second alert.

Honest Limits

Sampling-based classification misses rare signatures. Two hundred messages from a queue of fifty thousand will not surface a signature that holds forty messages, and those forty will ride along with whichever bulk verdict wins. The after-move check catches them when they land back in the DLQ, which is why the loop-detection rule exists, but they do take one extra trip.

The join depends on the consumer logging the message ID on failure. A consumer that logs the error without the ID, or that crashed before logging, leaves the agent with nothing, and the honest output is the needs_human row, not a guess from the body shape. Fixing that logging is a fifteen-minute change and it pays for itself on the first incident.

FIFO queues support move tasks, but the quarantine step breaks ordering within a message group, because pulling one message out and copying it elsewhere changes what the consumer sees next. For FIFO the agent only offers all-or-nothing verdicts per queue and says why.

The pattern is SQS-specific because StartMessageMoveTask is. A Kafka dead-letter topic has no equivalent; replay there means a consumer that reads the DLT and re-produces to the source topic, and the lag and partition facts come from the Kafka MCP server rather than CloudWatch. The triage logic transfers; the execution code does not.

Finally, redrive_after_fix is only as good as the human who marks the fix deployed. The agent stores the signature and the verdict in its incident memory so the next run on the same queue starts from "this signature was waiting on a consumer fix" instead of re-deriving it, but it will not decide on its own that the fix is live. That handoff is a deploy event, and deploy events belong to the pipeline, not to a queue-reading agent.