[FEAT] #512: Slack Integration for Smart Alerts via alert_smart_alert()

agentwatch/alerting/engine.py

@@ -58,6 +58,62 @@ async def alert_abuse(self, event: AbuseEvent) -> dict[str, bool]:
             )
         return sent
 
+    async def alert_custom(
+        self,
+        title: str,
+        text: str,
+        link: str | None = None,
+    ) -> dict[str, bool]:
+        """Surface a custom alert to Slack and PagerDuty channels."""
+        slack_blocks = [
+            {
+                "type": "section",
+                "text": {"type": "mrkdwn", "text": f"*{title}*"},
+            },
+            {"type": "section", "text": {"type": "mrkdwn", "text": text}},
+        ]
+        if link:
+            slack_blocks.append(
+                {
+                    "type": "section",
+                    "text": {
+                        "type": "mrkdwn",
+                        "text": f"<{link}|Review on Dashboard>",
+                    },
+                }
+            )
+
+        slack_payload = {
+            "text": title,
+            "blocks": slack_blocks,
+        }
+
+        # PagerDuty integration v2 payload
+        pd_payload = {
+            "routing_key": self._config.pagerduty_routing_key or "",
+            "event_action": "trigger",
+            "payload": {
+                "summary": f"{title}: {text}",
+                "source": "agentwatch",
+                "severity": "warning",
+            },
+        }
+
+        sent = {"slack": False, "pagerduty": False}
+        if self._config.slack_webhook_url:
+            sent["slack"] = await self._post(self._config.slack_webhook_url, slack_payload)
+        if self._config.pagerduty_webhook_url:
+            sent["pagerduty"] = await self._post(self._config.pagerduty_webhook_url, pd_payload)
+        return sent
+
+    async def alert_smart_alert(self, alert_text: str, dashboard_link: str) -> dict[str, bool]:
+        """Surface a Smart Alert generated by the Human-in-the-Loop ambient integration."""
+        return await self.alert_custom(
+            title="Smart Alert: Intervention Required",
+            text=alert_text,
+            link=dashboard_link,
+        )
+
     def _build_abuse_payload(self, event: AbuseEvent) -> dict[str, dict[str, Any]]:
         summary = (
             f"AgentWatch entitlement abuse: {event.subject} used on "

tests/test_alerting.py

@@ -71,3 +71,98 @@ async def mock_post(*args, **kwargs):
     sent = asyncio.run(engine.alert_event(event))
     assert sent["slack"] is False
     assert calls == 3
+
+
+def test_alert_custom_slack_and_pagerduty(monkeypatch):
+    slack_payloads = []
+    pd_payloads = []
+
+    class MockResponse:
+        def raise_for_status(self):
+            pass
+
+    async def mock_post(client, url, content, headers):
+        import json
+        data = json.loads(content.decode("utf-8"))
+        if "slack.com" in url:
+            slack_payloads.append(data)
+        elif "pagerduty.com" in url:
+            pd_payloads.append(data)
+        return MockResponse()
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", mock_post)
+
+    engine = AlertingEngine(
+        AlertingConfig(
+            slack_webhook_url="https://hooks.slack.com/services/TTEST1234/BTEST1234/abcdefghijklmn",
+            pagerduty_webhook_url="https://events.pagerduty.com/v2/enqueue",
+            pagerduty_routing_key="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+        )
+    )
+
+    sent = asyncio.run(
+        engine.alert_custom(
+            title="Custom Title",
+            text="Custom Text Detail",
+            link="https://dashboard.example.com",
+        )
+    )
+
+    assert sent["slack"] is True
+    assert sent["pagerduty"] is True
+
+    assert len(slack_payloads) == 1
+    assert slack_payloads[0]["text"] == "Custom Title"
+    assert slack_payloads[0]["blocks"][1]["text"]["text"] == "Custom Text Detail"
+
+    assert len(pd_payloads) == 1
+    assert pd_payloads[0]["payload"]["summary"] == "Custom Title: Custom Text Detail"
+    assert pd_payloads[0]["routing_key"] == "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+
+
+def test_alert_custom_no_webhooks_configured():
+    engine = AlertingEngine(AlertingConfig())
+    sent = asyncio.run(
+        engine.alert_custom(
+            title="Custom Title",
+            text="Custom Text Detail",
+        )
+    )
+    assert sent["slack"] is False
+    assert sent["pagerduty"] is False
+
+
+def test_alert_smart_alert_delivery(monkeypatch):
+    slack_payloads = []
+
+    class MockResponse:
+        def raise_for_status(self):
+            pass
+
+    async def mock_post(client, url, content, headers):
+        import json
+        data = json.loads(content.decode("utf-8"))
+        slack_payloads.append(data)
+        return MockResponse()
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", mock_post)
+
+    engine = AlertingEngine(
+        AlertingConfig(
+            slack_webhook_url="https://hooks.slack.com/services/TTEST1234/BTEST1234/abcdefghijklmn",
+        )
+    )
+
+    sent = asyncio.run(
+        engine.alert_smart_alert(
+            alert_text="Smart alert detail",
+            dashboard_link="https://dashboard.example.com/smart",
+        )
+    )
+
+    assert sent["slack"] is True
+    assert len(slack_payloads) == 1
+    assert slack_payloads[0]["text"] == "Smart Alert: Intervention Required"
+    assert slack_payloads[0]["blocks"][1]["text"]["text"] == "Smart alert detail"
+    assert "https://dashboard.example.com/smart" in slack_payloads[0]["blocks"][2]["text"]["text"]
+