feat(queue): enrich dead-letter insights

Author: haasonsaas

agent_pm/observability/metrics.py

@@ -5,7 +5,32 @@
 from contextlib import contextmanager
 from time import perf_counter
 
-from prometheus_client import Counter, Histogram, Summary, generate_latest
+from prometheus_client import Counter, Gauge, Histogram, Summary, generate_latest
+
+
+dead_letter_recorded_total = Counter(
+    "task_dead_letter_recorded_total",
+    "Dead-letter entries recorded",
+    labelnames=("queue", "error_type"),
+)
+
+dead_letter_requeued_total = Counter(
+    "task_dead_letter_requeued_total",
+    "Dead-letter entries requeued",
+    labelnames=("queue", "error_type"),
+)
+
+dead_letter_purged_total = Counter(
+    "task_dead_letter_purged_total",
+    "Dead-letter entries purged",
+    labelnames=("queue", "mode"),
+)
+
+dead_letter_active_gauge = Gauge(
+    "task_dead_letter_active",
+    "Current count of dead-letter entries",
+    labelnames=("queue",),
+)
 
 planner_requests_total = Counter(
     "planner_requests_total",
@@ -210,4 +235,8 @@ def latest_metrics() -> bytes:
     "record_task_completion",
     "record_task_latency",
     "latest_metrics",
+    "dead_letter_recorded_total",
+    "dead_letter_requeued_total",
+    "dead_letter_purged_total",
+    "dead_letter_active_gauge",
 ]

agent_pm/storage/redis.py

@@ -5,7 +5,7 @@
 import json
 import logging
 import uuid
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
 import redis.asyncio as redis
@@ -73,7 +73,7 @@ async def get_task_result(client: redis.Redis, task_id: str) -> dict[str, Any] |
 async def record_dead_letter(client: redis.Redis, payload: dict[str, Any]) -> None:
     task_id = payload.get("task_id", uuid.uuid4().hex)
     if "recorded_at" not in payload:
-        payload["recorded_at"] = datetime.now(timezone.utc).isoformat()
+        payload["recorded_at"] = datetime.now(UTC).isoformat()
     await client.hset(_dead_letter_key(), task_id, json.dumps(payload))
 
 
@@ -89,21 +89,35 @@ async def get_dead_letter(client: redis.Redis, task_id: str) -> dict[str, Any] |
     return data
 
 
-async def fetch_dead_letters(client: redis.Redis, limit: int = 100) -> list[dict[str, Any]]:
+async def fetch_dead_letters(
+    client: redis.Redis,
+    *,
+    limit: int | None = 100,
+    offset: int = 0,
+    include_total: bool = False,
+) -> tuple[list[dict[str, Any]], int] | list[dict[str, Any]]:
     items = await client.hgetall(_dead_letter_key())
     tasks: list[dict[str, Any]] = []
-    for task_id, value in items.items():
-        if len(tasks) >= limit:
+    entries = list(items.items())
+    for task_id, value in entries[offset:]:
+        if limit is not None and len(tasks) >= limit:
             break
         try:
             data = json.loads(value)
             data.setdefault("task_id", task_id)
             tasks.append(data)
         except json.JSONDecodeError:
             tasks.append({"task_id": task_id, "raw": value})
+    total = len(entries)
+    if include_total:
+        return tasks, total
     return tasks
 
 
+async def count_dead_letters(client: redis.Redis) -> int:
+    return await client.hlen(_dead_letter_key())
+
+
 async def clear_dead_letter(client: redis.Redis, task_id: str) -> None:
     await client.hdel(_dead_letter_key(), task_id)
 

agent_pm/storage/tasks.py

@@ -4,15 +4,20 @@
 
 import asyncio
 import logging
+import traceback
 import uuid
 from collections import deque
 from collections.abc import Callable, Coroutine
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 from enum import Enum
 from typing import Any
 
 from ..observability.metrics import (
+    dead_letter_active_gauge,
+    dead_letter_purged_total,
+    dead_letter_recorded_total,
+    dead_letter_requeued_total,
     record_task_completion,
     record_task_enqueued,
     record_task_latency,
@@ -21,6 +26,7 @@
 from ..utils.datetime import utc_now
 from .redis import (
     clear_dead_letter,
+    count_dead_letters,
     enqueue_task as redis_enqueue_task,
     fetch_dead_letters,
     get_dead_letter,
@@ -53,6 +59,7 @@ class Task:
     coro_fn: Callable[..., Coroutine[Any, Any, Any]]
     args: tuple
     kwargs: dict
+    metadata: dict[str, Any] = field(default_factory=dict)
     status: TaskStatus = TaskStatus.PENDING
     created_at: datetime = None  # type: ignore[assignment]
     started_at: datetime | None = None
@@ -104,6 +111,7 @@ async def enqueue(
         coro_fn: Callable[..., Coroutine[Any, Any, Any]],
         *args: Any,
         max_retries: int = 3,
+        metadata: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> str:
         """Enqueue a task and return task ID."""
@@ -115,6 +123,7 @@ async def enqueue(
             args=args,
             kwargs=kwargs,
             max_retries=max_retries,
+            metadata=metadata or {},
         )
         async with self._lock:
             self.queue.append(task)
@@ -199,8 +208,14 @@ async def _execute_task(self, task: Task):
                 record_task_latency(self.queue_name, (task.completed_at - task.started_at).total_seconds())
                 logger.error("Task permanently failed: %s (id=%s)", task.name, task.task_id)
 
-    async def list_dead_letters(self, limit: int = 100) -> list[dict[str, Any]]:
-        return []
+    async def list_dead_letters(
+        self,
+        limit: int = 100,
+        offset: int = 0,
+        workflow_id: str | None = None,
+        error_type: str | None = None,
+    ) -> tuple[list[dict[str, Any]], int]:
+        return [], 0
 
     async def delete_dead_letter(self, task_id: str) -> None:
         return None
@@ -254,6 +269,7 @@ async def enqueue(  # type: ignore[override]
                     coro_fn: Callable[..., Coroutine[Any, Any, Any]],
                     *args: Any,
                     max_retries: int = 3,
+                    metadata: dict[str, Any] | None = None,
                     **kwargs: Any,
                 ) -> str:
                     self.register(name, coro_fn)
@@ -264,6 +280,7 @@ async def enqueue(  # type: ignore[override]
                         "kwargs": kwargs,
                         "max_retries": max_retries,
                         "enqueued_at": utc_now().isoformat(),
+                        "metadata": metadata or {},
                     }
                     await redis_enqueue_task(self._redis, name, payload)
                     record_task_enqueued(self.queue_name)
@@ -289,10 +306,20 @@ async def _worker(self, worker_id: int):
                         coro_fn = self._registry.get(name)
                         if coro_fn is None:
                             logger.error("No registered task callable for %s", name)
-                            await record_dead_letter(
-                                self._redis,
-                                {**payload, "error": "missing_callable", "worker_id": worker_id},
-                            )
+                            payload.setdefault("metadata", {})
+                            payload["error_type"] = "MissingCallable"
+                            payload["error_message"] = f"Task callable not registered: {name}"
+                            payload["stack_trace"] = None
+                            payload["worker_id"] = worker_id
+                            await record_dead_letter(self._redis, payload)
+                            dead_letter_recorded_total.labels(
+                                queue=self.queue_name,
+                                error_type="TimeoutError",
+                            ).inc()
+                            dead_letter_recorded_total.labels(
+                                queue=self.queue_name,
+                                error_type=payload.get("error_type", payload.get("last_error", "unknown")),
+                            ).inc()
                             record_task_completion(self.queue_name, TaskStatus.FAILED.value)
                             continue
 
@@ -305,21 +332,35 @@ async def _worker(self, worker_id: int):
                                 coro_fn(*payload.get("args", ()), **payload.get("kwargs", {})),
                                 timeout=self._task_timeout,
                             )
-                        except asyncio.TimeoutError:
+                        except TimeoutError:
+                            payload.setdefault("metadata", {})
                             payload["retry_count"] = retries + 1
                             payload["last_error"] = "timeout"
+                            payload["error_type"] = "TimeoutError"
+                            payload["error_message"] = (
+                                f"Task execution exceeded timeout of {self._task_timeout} seconds"
+                            )
+                            payload["stack_trace"] = None
                             payload["worker_id"] = worker_id
                             await record_dead_letter(self._redis, payload)
                             record_task_completion(self.queue_name, TaskStatus.FAILED.value)
                             record_task_latency(self.queue_name, (utc_now() - start).total_seconds())
                             continue
                         except Exception as exc:  # pylint: disable=broad-except
+                            payload.setdefault("metadata", {})
                             retries += 1
                             payload["retry_count"] = retries
                             payload["last_error"] = str(exc)
                             if retries >= max_retries:
+                                payload["error_type"] = exc.__class__.__name__
+                                payload["error_message"] = str(exc)
+                                payload["stack_trace"] = traceback.format_exc()
                                 payload["worker_id"] = worker_id
                                 await record_dead_letter(self._redis, payload)
+                                dead_letter_recorded_total.labels(
+                                    queue=self.queue_name,
+                                    error_type=payload.get("error_type", "unknown"),
+                                ).inc()
                                 record_task_completion(self.queue_name, TaskStatus.FAILED.value)
                                 record_task_latency(self.queue_name, (utc_now() - start).total_seconds())
                                 continue
@@ -343,8 +384,25 @@ async def _worker(self, worker_id: int):
 
                     logger.info("Redis worker %d stopped", worker_id)
 
-                async def list_dead_letters(self, limit: int = 100) -> list[dict[str, Any]]:
-                    return await fetch_dead_letters(self._redis, limit)
+                async def list_dead_letters(
+                    self,
+                    limit: int = 100,
+                    offset: int = 0,
+                    workflow_id: str | None = None,
+                    error_type: str | None = None,
+                ) -> tuple[list[dict[str, Any]], int]:
+                    items, total_raw = await fetch_dead_letters(self._redis, limit=None, include_total=True)
+                    filtered: list[dict[str, Any]] = []
+                    for item in items:
+                        if workflow_id and item.get("metadata", {}).get("workflow_id") != workflow_id:
+                            continue
+                        if error_type and item.get("error_type") != error_type:
+                            continue
+                        filtered.append(item)
+                    total_filtered = len(filtered)
+                    window = filtered[offset : offset + limit]
+                    dead_letter_active_gauge.labels(queue=self.queue_name).set(total_filtered)
+                    return window, total_filtered
 
                 async def delete_dead_letter(self, task_id: str) -> None:
                     await clear_dead_letter(self._redis, task_id)
@@ -363,6 +421,9 @@ async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
                     payload["requeued_at"] = utc_now().isoformat()
                     await redis_enqueue_task(self._redis, payload.get("name", "unknown"), payload)
                     record_task_enqueued(self.queue_name)
+                    dead_letter_requeued_total.labels(
+                        queue=self.queue_name, error_type=payload.get("error_type", "unknown")
+                    ).inc()
                     logger.info("Dead-letter task requeued: %s", task_id)
                     return payload
 
@@ -371,9 +432,15 @@ async def get_dead_letter(self, task_id: str) -> dict[str, Any] | None:
 
                 async def purge_dead_letters(self, *, older_than: timedelta | None = None) -> int:
                     if older_than is None:
-                        return await purge_dead_letters(self._redis)
+                        deleted = await purge_dead_letters(self._redis)
+                        dead_letter_purged_total.labels(queue=self.queue_name, mode="all").inc(deleted)
+                        dead_letter_active_gauge.labels(queue=self.queue_name).set(await count_dead_letters(self._redis))
+                        return deleted
                     cutoff = utc_now() - older_than
-                    return await purge_dead_letters(self._redis, older_than=cutoff)
+                    deleted = await purge_dead_letters(self._redis, older_than=cutoff)
+                    dead_letter_purged_total.labels(queue=self.queue_name, mode="age_filter").inc(deleted)
+                    dead_letter_active_gauge.labels(queue=self.queue_name).set(await count_dead_letters(self._redis))
+                    return deleted
 
             _task_queue = RedisTaskQueue(max_workers=settings.task_queue_workers)
         else:

app.py

@@ -456,11 +456,33 @@ async def list_tasks(status: str | None = None, limit: int = 50, _admin_key: Adm
 
 
 @app.get("/tasks/dead-letter")
-async def list_dead_letter(limit: int = 50, _admin_key: AdminKeyDep = None) -> dict[str, Any]:
+async def list_dead_letter(
+    limit: int = 50,
+    offset: int = 0,
+    workflow_id: str | None = None,
+    error_type: str | None = None,
+    _admin_key: AdminKeyDep = None,
+) -> dict[str, Any]:
     if not _task_queue:
         raise HTTPException(status_code=503, detail="Task queue not initialized")
-    items = await _task_queue.list_dead_letters(limit)
-    return {"dead_letter": items, "total": len(items)}
+    items, total = await _task_queue.list_dead_letters(
+        limit=limit, offset=offset, workflow_id=workflow_id, error_type=error_type
+    )
+    return {
+        "dead_letter": [
+            {
+                **item,
+                "metadata": item.get("metadata", {}),
+                "error_type": item.get("error_type"),
+                "error_message": item.get("error_message"),
+                "stack_trace": item.get("stack_trace"),
+            }
+            for item in items
+        ],
+        "total": total,
+        "limit": limit,
+        "offset": offset,
+    }
 
 
 @app.get("/tasks/dead-letter/{task_id}")

tests/storage/test_redis.py

@@ -96,6 +96,7 @@ async def test_dead_letter_and_heartbeat_helpers():
     payload = {"task_id": "abc", "name": "job", "args": [], "kwargs": {}}
     await redis.record_dead_letter(client, payload)
     records = await redis.fetch_dead_letters(client)
+    assert isinstance(records, list)
     assert records[0]["task_id"] == "abc"
 
     stored = await redis.get_dead_letter(client, "abc")