feat(queue): add remediation playbooks and pagerduty alerts

Author: haasonsaas

agent_pm/clients/__init__.py

@@ -4,6 +4,7 @@
 from .github_client import GitHubClient, github_client
 from .jira_client import JiraClient, jira_client
 from .openai_client import OpenAIClient, openai_client
+from .pagerduty_client import PagerDutyClient, pagerduty_client
 from .slack_client import SlackClient, slack_client
 
 __all__ = [
@@ -17,4 +18,6 @@
     "openai_client",
     "SlackClient",
     "slack_client",
+    "PagerDutyClient",
+    "pagerduty_client",
 ]

agent_pm/clients/pagerduty_client.py

@@ -0,0 +1,44 @@
+"""PagerDuty client for incident notifications."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+from agent_pm.settings import settings
+
+
+class PagerDutyClient:
+    def __init__(self) -> None:
+        self.routing_key = settings.pagerduty_routing_key
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.routing_key)
+
+    async def trigger_incident(self, summary: str, source: str = "agent-pm", severity: str = "error", **details: Any) -> dict[str, Any]:
+        if not self.enabled:
+            return {"dry_run": True, "summary": summary, "details": details}
+
+        payload = {
+            "routing_key": self.routing_key,
+            "event_action": "trigger",
+            "payload": {
+                "summary": summary,
+                "source": source,
+                "severity": severity,
+                "custom_details": details,
+            },
+        }
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post("https://events.pagerduty.com/v2/enqueue", json=payload, timeout=10)
+            response.raise_for_status()
+            return response.json()
+
+
+pagerduty_client = PagerDutyClient()
+
+
+__all__ = ["pagerduty_client", "PagerDutyClient"]

agent_pm/observability/dashboard.py

@@ -0,0 +1,23 @@
+"""Helpers to build queue health dashboards."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ..storage.redis import count_dead_letters
+from ..storage.tasks import get_task_queue
+
+
+@dataclass
+class QueueHealth:
+    queue_name: str
+    dead_letters: int
+    auto_triage_enabled: bool
+
+
+async def gather_queue_health() -> QueueHealth:
+    queue = await get_task_queue()
+    client = await queue.get_client()  # type: ignore[attr-defined]
+    dead_letters = await count_dead_letters(client)
+    auto_triage_enabled = bool(queue)
+    return QueueHealth(queue_name=getattr(queue, "queue_name", "unknown"), dead_letters=dead_letters, auto_triage_enabled=auto_triage_enabled)

agent_pm/settings.py

@@ -82,6 +82,9 @@ def _parse_google_scopes(cls, value):
     task_queue_alert_channel: str | None = Field(None, alias="TASK_QUEUE_ALERT_CHANNEL")
     task_queue_alert_cooldown_minutes: int = Field(10, alias="TASK_QUEUE_ALERT_COOLDOWN_MINUTES")
     task_queue_max_auto_requeues: int = Field(3, alias="TASK_QUEUE_MAX_AUTO_REQUEUES")
+    task_queue_alert_webhook_url: str | None = Field(None, alias="TASK_QUEUE_ALERT_WEBHOOK_URL")
+    pagerduty_routing_key: str | None = Field(None, alias="PAGERDUTY_ROUTING_KEY")
+    task_queue_playbooks: dict[str, str] = Field(default_factory=dict, alias="TASK_QUEUE_PLAYBOOKS")
     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

@@ -26,7 +26,8 @@
 )
 from ..settings import settings
 from ..utils.datetime import utc_now
-from ..clients.slack_client import slack_client
+from ..clients import pagerduty_client, slack_client
+from ..tasks.playbooks import run_playbook
 from .redis import (
     append_dead_letter_audit,
     clear_dead_letter,
@@ -444,6 +445,9 @@ async def _send_alert(error_type: str, payload: dict[str, Any]) -> None:
                                 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")
+                                playbook_name = settings.task_queue_playbooks.get(error_type)
+                                if playbook_name:
+                                    await run_playbook(playbook_name, payload, self, error_type)
                                 if _should_auto_requeue(error_type):
                                     key = f"{identifier}:{error_type}"
                                     count = auto_requeue_counts.get(key, 0)
@@ -525,6 +529,9 @@ async def delete_dead_letter(self, task_id: str) -> None:
                 async def worker_heartbeats(self) -> dict[str, dict[str, Any]]:
                     return await list_heartbeats(self._redis)
 
+                async def get_client(self):
+                    return self._redis
+
                 async def requeue_dead_letter(
                     self,
                     task_id: str,

agent_pm/tasks/__init__.py

@@ -0,0 +1,5 @@
+"""Task utilities."""
+
+from .playbooks import PLAYBOOKS, run_playbook
+
+__all__ = ["PLAYBOOKS", "run_playbook"]

agent_pm/tasks/playbooks.py

@@ -0,0 +1,86 @@
+"""Remediation playbooks for automated task recovery."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Awaitable, Callable
+
+import httpx
+
+from agent_pm.clients import pagerduty_client, slack_client
+from agent_pm.settings import settings
+
+logger = logging.getLogger(__name__)
+
+PlaybookHandler = Callable[[dict[str, Any], Any, str], Awaitable[None]]
+
+
+async def _notify_slack(payload: dict[str, Any], queue: Any, error_type: str) -> None:
+    channel = settings.task_queue_alert_channel or slack_client.channel
+    if not channel:
+        return
+    message = (
+        ":rotating_light: Remediation triggered\n"
+        f"Queue: `{getattr(queue, 'queue_name', 'unknown')}`\n"
+        f"Error: `{error_type}`\n"
+        f"Task: `{payload.get('name', 'unknown')}`"
+    )
+    if settings.dry_run or not slack_client.enabled:
+        logger.warning("Remediation Slack notification skipped (dry run)")
+        return
+        await slack_client.post_digest(message, channel)
+
+
+async def _invoke_webhook(payload: dict[str, Any], queue: Any, error_type: str) -> None:
+    url = settings.task_queue_alert_webhook_url
+    if not url:
+        return
+    body = {
+        "queue": getattr(queue, "queue_name", "unknown"),
+        "error_type": error_type,
+        "task": payload.get("name", "unknown"),
+        "task_id": payload.get("task_id"),
+        "metadata": payload.get("metadata", {}),
+        "playbook": "webhook",
+    }
+    async with httpx.AsyncClient() as client:
+        response = await client.post(url, json=body, timeout=10)
+        response.raise_for_status()
+
+
+async def _log_only(payload: dict[str, Any], queue: Any, error_type: str) -> None:
+    logger.warning(
+        "Remediation playbook invoked", extra={"queue": getattr(queue, "queue_name", "unknown"), "error": error_type, "task": payload.get("task_id")}
+    )
+
+
+async def _notify_pagerduty(payload: dict[str, Any], queue: Any, error_type: str) -> None:
+    if not pagerduty_client.enabled:
+        return
+    summary = f"Auto-remediation triggered for {error_type}"
+    details = {
+        "queue": getattr(queue, "queue_name", "unknown"),
+        "task_id": payload.get("task_id"),
+        "task_name": payload.get("name"),
+        "metadata": payload.get("metadata", {}),
+    }
+    await pagerduty_client.trigger_incident(summary, severity="error", **details)
+
+
+PLAYBOOKS: dict[str, PlaybookHandler] = {
+    "log_only": _log_only,
+    "notify_slack": _notify_slack,
+    "webhook": _invoke_webhook,
+    "notify_pagerduty": _notify_pagerduty,
+}
+
+
+async def run_playbook(name: str, payload: dict[str, Any], queue: Any, error_type: str) -> None:
+    handler = PLAYBOOKS.get(name)
+    if not handler:
+        logger.warning("Unknown playbook requested: %s", name)
+        return
+    try:
+        await handler(payload, queue, error_type)
+    except Exception as exc:  # pragma: no cover - defensive logging
+        logger.exception("Playbook %s failed: %s", name, exc)

tests/storage/test_redis_worker.py

@@ -154,6 +154,7 @@ async def test_auto_triage_requeues_and_alerts(redis_queue, monkeypatch):
     monkeypatch.setattr(tasks_module.settings, "task_queue_alert_channel", "alerts")
     monkeypatch.setattr(tasks_module.settings, "task_queue_alert_cooldown_minutes", 0)
     monkeypatch.setattr(tasks_module.settings, "task_queue_max_auto_requeues", 2)
+    monkeypatch.setattr(tasks_module.settings, "task_queue_playbooks", {"RuntimeError": "log_only"})
 
     tasks_module.slack_client.token = "token"
     tasks_module.slack_client.channel = "alerts"