feat(queue): add auto triage handling

Author: haasonsaas

agent_pm/observability/metrics.py

@@ -38,6 +38,12 @@
     labelnames=("queue", "error_type"),
 )
 
+dead_letter_alert_total = Counter(
+    "task_dead_letter_alert_total",
+    "Dead-letter alert notifications",
+    labelnames=("queue", "error_type"),
+)
+
 planner_requests_total = Counter(
     "planner_requests_total",
     "Total planner invocations",
@@ -246,4 +252,5 @@ def latest_metrics() -> bytes:
     "dead_letter_purged_total",
     "dead_letter_active_gauge",
     "dead_letter_auto_requeue_total",
+    "dead_letter_alert_total",
 ]

agent_pm/settings.py

@@ -76,6 +76,10 @@ def _parse_google_scopes(cls, value):
     task_queue_retry_backoff_base: float = Field(2.0, alias="TASK_QUEUE_RETRY_BACKOFF_BASE")
     task_queue_retry_backoff_max: float = Field(60.0, alias="TASK_QUEUE_RETRY_BACKOFF_MAX")
     task_queue_worker_heartbeat_ttl: int = Field(60, alias="TASK_QUEUE_WORKER_HEARTBEAT_TTL")
+    task_queue_auto_requeue_errors: list[str] = Field(default_factory=list, alias="TASK_QUEUE_AUTO_REQUEUE_ERRORS")
+    task_queue_alert_threshold: int = Field(5, alias="TASK_QUEUE_ALERT_THRESHOLD")
+    task_queue_alert_window_minutes: int = Field(5, alias="TASK_QUEUE_ALERT_WINDOW_MINUTES")
+    task_queue_alert_channel: str | None = Field(None, alias="TASK_QUEUE_ALERT_CHANNEL")
     database_url: str | None = Field("sqlite+aiosqlite:///./data/agent_pm.db", alias="DATABASE_URL")
     database_echo: bool = Field(False, alias="DATABASE_ECHO")
     redis_url: str = Field("redis://localhost:6379", alias="REDIS_URL")

agent_pm/storage/tasks.py

@@ -15,6 +15,7 @@
 
 from ..observability.metrics import (
     dead_letter_active_gauge,
+    dead_letter_alert_total,
     dead_letter_auto_requeue_total,
     dead_letter_purged_total,
     dead_letter_recorded_total,
@@ -318,7 +319,50 @@ async def pop(self) -> dict[str, Any] | None:
 
                 async def _worker(self, worker_id: int):
                     logger.info("Redis worker %d started", worker_id)
+                    recent_failures: dict[str, list[datetime]] = {}
+
                     while self.running:
+                        auto_errors = set(settings.task_queue_auto_requeue_errors)
+                        alert_threshold = settings.task_queue_alert_threshold
+                        alert_window = timedelta(minutes=settings.task_queue_alert_window_minutes)
+                        alert_channel = settings.task_queue_alert_channel or settings.slack_status_channel
+
+                        def _should_auto_requeue(err_type: str | None) -> bool:
+                            if not err_type:
+                                return False
+                            return err_type in auto_errors
+
+                        def _record_failure(err_type: str | None, task_identifier: str) -> bool:
+                            if not err_type:
+                                return False
+                            now = utc_now()
+                            key = f"{err_type}:{task_identifier}"
+                            entries = recent_failures.setdefault(key, [])
+                            entries.append(now)
+                            cutoff = now - alert_window
+                            recent_failures[key] = [ts for ts in entries if ts >= cutoff]
+                            return len(recent_failures[key]) >= alert_threshold
+
+                        async def _send_alert(error_type: str, payload: dict[str, Any]) -> None:
+                            if not alert_channel:
+                                return
+                            if settings.dry_run or not slack_client.enabled:
+                                logger.warning(
+                                    "Slack alert skipped (dry run): %s", error_type
+                                )
+                                return
+                            body = (
+                                f":rotating_light: Dead-letter threshold exceeded\n"
+                                f"Queue: `{self.queue_name}`\n"
+                                f"Error: `{error_type}`\n"
+                                f"Task: `{payload.get('name', 'unknown')}`"
+                            )
+                            try:
+                                await slack_client.post_digest(body, alert_channel)
+                                dead_letter_alert_total.labels(queue=self.queue_name, error_type=error_type).inc()
+                            except Exception as exc:  # pragma: no cover - logging
+                                logger.error("Failed to send Slack alert: %s", exc)
+
                         payload = await self.pop()
                         if not payload:
                             await asyncio.sleep(self._poll_interval)
@@ -383,13 +427,23 @@ async def _worker(self, worker_id: int):
                                 payload["error_message"] = str(exc)
                                 payload["stack_trace"] = traceback.format_exc()
                                 payload["worker_id"] = worker_id
+                                error_type = payload.get("error_type", "unknown")
                                 await record_dead_letter(self._redis, payload)
                                 dead_letter_recorded_total.labels(
                                     queue=self.queue_name,
-                                    error_type=payload.get("error_type", "unknown"),
+                                    error_type=error_type,
                                 ).inc()
                                 record_task_completion(self.queue_name, TaskStatus.FAILED.value)
                                 record_task_latency(self.queue_name, (utc_now() - start).total_seconds())
+                                identifier = payload.get("metadata", {}).get("workflow_id") or payload.get("name", "unknown")
+                                if _should_auto_requeue(error_type):
+                                    await self.requeue_dead_letter(
+                                        task_id,
+                                        automatic=True,
+                                        notify=False,
+                                    )
+                                if _record_failure(error_type, identifier):
+                                    await _send_alert(error_type, payload)
                                 continue
 
                             backoff = min(

tests/storage/test_redis_worker.py

@@ -4,6 +4,7 @@
 import pytest
 import pytest_asyncio
 
+from agent_pm.observability.metrics import dead_letter_alert_total, dead_letter_auto_requeue_total
 from agent_pm.settings import settings
 from agent_pm.storage import redis as redis_helpers
 import agent_pm.storage.tasks as tasks_module
@@ -66,6 +67,10 @@ async def redis_queue(monkeypatch):
     monkeypatch.setattr(settings, "task_queue_retry_backoff_base", 1.1)
     monkeypatch.setattr(settings, "task_queue_retry_backoff_max", 0.05)
     monkeypatch.setattr(settings, "task_queue_task_timeout", 1)
+    monkeypatch.setattr(settings, "task_queue_auto_requeue_errors", [])
+    monkeypatch.setattr(settings, "task_queue_alert_threshold", 10)
+    monkeypatch.setattr(settings, "task_queue_alert_window_minutes", 5)
+    monkeypatch.setattr(settings, "task_queue_alert_channel", None)
 
     async def fake_client():
         return fake
@@ -136,3 +141,63 @@ async def fail_task() -> None:
 
     queued_len = await fake.llen("agent_pm:tasks")
     assert queued_len >= 1
+
+
+@pytest.mark.asyncio
+async def test_auto_triage_requeues_and_alerts(redis_queue, monkeypatch):
+    queue, fake = redis_queue
+
+    monkeypatch.setattr(tasks_module.settings, "dry_run", False)
+    monkeypatch.setattr(tasks_module.settings, "task_queue_auto_requeue_errors", ["RuntimeError"])
+    monkeypatch.setattr(tasks_module.settings, "task_queue_alert_threshold", 1)
+    monkeypatch.setattr(tasks_module.settings, "task_queue_alert_window_minutes", 5)
+    monkeypatch.setattr(tasks_module.settings, "task_queue_alert_channel", "alerts")
+
+    tasks_module.slack_client.token = "token"
+    tasks_module.slack_client.channel = "alerts"
+
+    calls: list[dict[str, str]] = []
+
+    async def fake_digest(body_md: str, channel: str | None = None):
+        calls.append({"body": body_md, "channel": channel or ""})
+        return {"ok": True}
+
+    monkeypatch.setattr(tasks_module.slack_client, "post_digest", fake_digest)
+
+    try:
+        dead_letter_auto_requeue_total.remove("redis", "RuntimeError")
+    except KeyError:
+        pass
+    try:
+        dead_letter_alert_total.remove("redis", "RuntimeError")
+    except KeyError:
+        pass
+
+    state = {"count": 0}
+
+    async def flaky_task() -> str:
+        state["count"] += 1
+        if state["count"] == 1:
+            raise RuntimeError("transient")
+        return "ok"
+
+    metadata = {"workflow_id": "wf-1"}
+    task_id = await queue.enqueue("flaky", flaky_task, max_retries=1, metadata=metadata)
+
+    result = None
+    for _ in range(100):
+        result = await redis_helpers.get_task_result(fake, task_id)
+        if result:
+            break
+        await asyncio.sleep(0.05)
+
+    assert result is not None
+    assert result["status"] == "completed"
+    assert result["result"] == "ok"
+
+    auto_metric = dead_letter_auto_requeue_total.labels(queue="redis", error_type="RuntimeError")._value.get()
+    alert_metric = dead_letter_alert_total.labels(queue="redis", error_type="RuntimeError")._value.get()
+
+    assert auto_metric == 1
+    assert alert_metric == 1
+    assert calls