chore: redesign notification hooks for better ergonomics pr feedback

Author: GangGreenTemperTatum

dreadnode/agent/agent.py

@@ -30,6 +30,7 @@
     _total_usage_from_events,
 )
 from dreadnode.agent.hooks import Hook, retry_with_feedback
+from dreadnode.agent.hooks.notification import NotificationBackend, TerminalNotificationBackend
 from dreadnode.agent.reactions import (
     Continue,
     Fail,
@@ -62,6 +63,16 @@
 CommitBehavior = t.Literal["always", "on-success"]
 
 
+async def _safe_send(
+    backend: NotificationBackend, event: AgentEvent, message: str
+) -> None:
+    """Send notification with error handling."""
+    try:
+        await backend.send(event, message)
+    except Exception:  # noqa: BLE001
+        logger.exception(f"Notification failed for {event.__class__.__name__}")
+
+
 class AgentWarning(UserWarning):
     """Warning raised when an agent is used in a way that may not be safe or intended."""
 
@@ -111,6 +122,24 @@ class Agent(Model):
     assert_scores: list[str] | t.Literal[True] = Field(default_factory=list)
     """Scores to ensure are truthy, otherwise the agent task is marked as failed."""
 
+    notifications: t.Annotated[
+        bool | NotificationBackend | None, SkipValidation
+    ] = Config(default=None, repr=False)
+    """
+    Enable notifications.
+    - True: Uses TerminalNotificationBackend (stderr output)
+    - NotificationBackend instance: Uses custom backend
+    - None/False: Disabled
+    """
+    notification_events: list[type[AgentEvent]] | t.Literal["all"] = Config(
+        default="all", repr=False
+    )
+    """Which event types to notify on. Defaults to all events."""
+    notification_formatter: t.Annotated[
+        t.Callable[[AgentEvent], str] | None, SkipValidation
+    ] = Config(default=None, repr=False)
+    """Custom formatter for notification messages. If None, uses event's default representation."""
+
     _generator: rg.Generator | None = PrivateAttr(None, init=False)
 
     @field_validator("tools", mode="before")
@@ -129,6 +158,49 @@ def validate_tools(cls, value: t.Any) -> t.Any:
 
         return tools
 
+    def model_post_init(self, context: t.Any) -> None:
+        super().model_post_init(context)
+
+        # Auto-inject notification hook if enabled
+        if self.notifications:
+            backend = (
+                self.notifications
+                if isinstance(self.notifications, NotificationBackend)
+                else TerminalNotificationBackend()
+            )
+
+            self.hooks.append(
+                self._create_notification_hook(
+                    backend,
+                    self.notification_events,
+                    self.notification_formatter,
+                )
+            )
+
+    def _create_notification_hook(
+        self,
+        backend: NotificationBackend,
+        events: list[type[AgentEvent]] | t.Literal["all"],
+        formatter: t.Callable[[AgentEvent], str] | None,
+    ) -> Hook:
+        """Create a notification hook that delegates formatting to events."""
+        import asyncio
+
+        async def notification_hook(event: AgentEvent) -> None:
+            # Filter events
+            if events != "all" and not any(isinstance(event, et) for et in events):
+                return
+
+            # Use custom formatter if provided, otherwise delegate to event
+            message = formatter(event) if formatter else event.format_notification()
+
+            # Fire and forget - don't block agent execution
+            _ = asyncio.create_task(_safe_send(backend, event, message))  # noqa: RUF006
+
+            return
+
+        return notification_hook
+
     def __repr__(self) -> str:
         description = shorten_string(self.description or "", 50)
 

dreadnode/agent/events.py

@@ -115,12 +115,22 @@ def format_as_panel(self, *, truncate: bool = False) -> Panel:  # noqa: ARG002
             border_style="dim",
         )
 
+    def format_notification(self) -> str:
+        """
+        Format this event as a human-readable notification message.
+        Override in subclasses for custom formatting.
+        """
+        return f"{self.__class__.__name__}"
+
     def __rich_console__(self, console: Console, options: ConsoleOptions) -> RenderResult:
         yield self.format_as_panel()
 
 
 @dataclass
 class AgentStart(AgentEvent):
+    def format_notification(self) -> str:
+        return f"Starting agent: {self.agent.name}"
+
     def format_as_panel(self, *, truncate: bool = False) -> Panel:
         return Panel(
             format_message(self.messages[0], truncate=truncate),
@@ -158,6 +168,10 @@ def __repr__(self) -> str:
         message = f"Message(role={self.message.role}, content='{message_content}', tool_calls={tool_call_count})"
         return f"GenerationEnd(message={message})"
 
+    def format_notification(self) -> str:
+        tokens = self.usage.total_tokens if self.usage else "unknown"
+        return f"Generation complete ({tokens} tokens)"
+
     def format_as_panel(self, *, truncate: bool = False) -> Panel:
         cost = round(self.estimated_cost, 6) if self.estimated_cost else ""
         usage = str(self.usage) or ""
@@ -173,6 +187,9 @@ def format_as_panel(self, *, truncate: bool = False) -> Panel:
 
 @dataclass
 class AgentStalled(AgentEventInStep):
+    def format_notification(self) -> str:
+        return "Agent stalled: no tool calls and no stop conditions met"
+
     def format_as_panel(self, *, truncate: bool = False) -> Panel:  # noqa: ARG002
         return Panel(
             Text(
@@ -189,6 +206,9 @@ def format_as_panel(self, *, truncate: bool = False) -> Panel:  # noqa: ARG002
 class AgentError(AgentEventInStep):
     error: BaseException
 
+    def format_notification(self) -> str:
+        return f"Error: {self.error.__class__.__name__}: {self.error!s}"
+
     def format_as_panel(self, *, truncate: bool = False) -> Panel:  # noqa: ARG002
         return Panel(
             repr(self),
@@ -205,6 +225,9 @@ class ToolStart(AgentEventInStep):
     def __repr__(self) -> str:
         return f"ToolStart(tool_call={self.tool_call})"
 
+    def format_notification(self) -> str:
+        return f"Starting tool: {self.tool_call.name}"
+
     def format_as_panel(self, *, truncate: bool = False) -> Panel:
         content: RenderableType
         try:
@@ -245,6 +268,10 @@ def __repr__(self) -> str:
         message = f"Message(role={self.message.role}, content='{message_content}')"
         return f"ToolEnd(tool_call={self.tool_call}, message={message}, stop={self.stop})"
 
+    def format_notification(self) -> str:
+        status = " (requesting stop)" if self.stop else ""
+        return f"Finished tool: {self.tool_call.name}{status}"
+
     def format_as_panel(self, *, truncate: bool = False) -> Panel:
         panel = format_message(self.message, truncate=truncate)
         subtitle = f"[dim]{self.tool_call.id}[/dim]"
@@ -294,6 +321,10 @@ class AgentEnd(AgentEvent):
     stop_reason: "AgentStopReason"
     result: "AgentResult"
 
+    def format_notification(self) -> str:
+        status = "❌ Failed" if self.result.failed else "✅ Finished"
+        return f"{status}: {self.stop_reason} (steps: {self.result.steps}, tokens: {self.result.usage.total_tokens})"
+
     def format_as_panel(self, *, truncate: bool = False) -> Panel:  # noqa: ARG002
         res = self.result
         status = "[bold red]Failed[/bold red]" if res.failed else "[bold green]Success[/bold green]"

dreadnode/agent/hooks/notification.py

@@ -4,8 +4,9 @@
 from loguru import logger
 
 if t.TYPE_CHECKING:
+    import httpx
+
     from dreadnode.agent.events import AgentEvent
-    from dreadnode.agent.reactions import Reaction
 
 
 class NotificationBackend(ABC):
@@ -27,29 +28,46 @@ async def send(self, event: "AgentEvent", message: str) -> None:
 
 
 class WebhookNotificationBackend(NotificationBackend):
-    def __init__(self, url: str, headers: dict[str, str] | None = None):
+    def __init__(self, url: str, headers: dict[str, str] | None = None, timeout: float = 5.0):
         self.url = url
         self.headers = headers or {}
+        self.timeout = timeout
+        self._client: httpx.AsyncClient | None = None
+
+    async def __aenter__(self) -> "WebhookNotificationBackend":
+        import httpx
+
+        self._client = httpx.AsyncClient(timeout=self.timeout)
+        return self
+
+    async def __aexit__(self, *args: t.Any) -> None:
+        if self._client:
+            await self._client.aclose()
 
     async def send(self, event: "AgentEvent", message: str) -> None:
         import httpx
 
-        payload = {
+        if not self._client:
+            self._client = httpx.AsyncClient(timeout=self.timeout)
+
+        payload = self._build_payload(event, message)
+        await self._client.post(self.url, json=payload, headers=self.headers)
+
+    def _build_payload(self, event: "AgentEvent", message: str) -> dict[str, str]:
+        """Override this to customize webhook payload."""
+        return {
             "agent": event.agent.name,
             "event": event.__class__.__name__,
             "message": message,
             "timestamp": event.timestamp.isoformat(),
         }
 
-        async with httpx.AsyncClient() as client:
-            await client.post(self.url, json=payload, headers=self.headers)
-
 
 def notify(
     event_type: "type[AgentEvent] | t.Callable[[AgentEvent], bool]",
-    message: str | t.Callable[["AgentEvent"], str],
+    message: str | t.Callable[["AgentEvent"], str] | None = None,
     backend: NotificationBackend | None = None,
-) -> t.Callable[["AgentEvent"], t.Awaitable["Reaction | None"]]:
+) -> t.Callable[["AgentEvent"], t.Awaitable[None]]:
     """
     Create a notification hook that sends notifications when events occur.
 
@@ -58,7 +76,8 @@ def notify(
 
     Args:
         event_type: Event type to trigger on, or predicate function
-        message: Static message or callable that generates message from event
+        message: Static message or callable that generates message from event.
+                 If None, uses event.format_notification()
         backend: Notification backend (defaults to terminal output)
 
     Returns:
@@ -73,6 +92,7 @@ def notify(
         agent = Agent(
             name="analyzer",
             hooks=[
+                notify(ToolStart),  # Uses default formatting
                 notify(
                     ToolStart,
                     lambda e: f"Starting tool: {e.tool_name}",
@@ -83,7 +103,7 @@ def notify(
     """
     notification_backend = backend or TerminalNotificationBackend()
 
-    async def notification_hook(event: "AgentEvent") -> "Reaction | None":
+    async def notification_hook(event: "AgentEvent") -> None:
         should_notify = False
 
         if isinstance(event_type, type):
@@ -92,15 +112,19 @@ async def notification_hook(event: "AgentEvent") -> "Reaction | None":
             should_notify = event_type(event)
 
         if not should_notify:
-            return None
+            return
 
-        msg = message(event) if callable(message) else message
+        # Use custom message if provided, otherwise delegate to event
+        if message is None:
+            msg = event.format_notification()
+        else:
+            msg = message(event) if callable(message) else message
 
         try:
             await notification_backend.send(event, msg)
         except Exception:  # noqa: BLE001
             logger.exception("Notification hook failed")
 
-        return None
+        return
 
     return notification_hook

tests/test_notification_hook.py

@@ -64,16 +64,16 @@ async def test_terminal_notification_backend(mock_event: AgentEvent) -> None:
 
 
 async def test_webhook_notification_backend(mock_event: AgentEvent) -> None:
+    from unittest.mock import patch
+
+    import httpx
+
     mock_client = AsyncMock()
     mock_post = AsyncMock()
-    mock_client.__aenter__.return_value.post = mock_post
+    mock_client.post = mock_post
 
     backend = WebhookNotificationBackend("https://example.com/webhook")
 
-    from unittest.mock import patch
-
-    import httpx
-
     with patch.object(httpx, "AsyncClient", return_value=mock_client):
         await backend.send(mock_event, "Test notification")
 
@@ -159,3 +159,55 @@ async def test_notify_hook_handles_backend_failure(mock_event: AgentEvent) -> No
 
     assert reaction is None
     mock_logger.assert_called_once_with("Notification hook failed")
+
+
+async def test_notify_hook_uses_default_formatter(mock_event: AgentEvent) -> None:
+    backend = MagicMock(spec=NotificationBackend)
+    backend.send = AsyncMock()
+
+    hook = notify(MockEvent, backend=backend)
+
+    reaction = await hook(mock_event)
+
+    assert reaction is None
+    backend.send.assert_called_once()
+    # Check that it used event.format_notification()
+    call_args = backend.send.call_args[0]
+    assert call_args[1] == "MockEvent"  # Default format_notification returns class name
+
+
+def test_agent_auto_inject_notifications_terminal() -> None:
+    from dreadnode.agent import Agent
+
+    agent = Agent(name="test", notifications=True)
+
+    # Should have auto-injected a notification hook
+    assert len(agent.hooks) == 1
+
+
+def test_agent_auto_inject_notifications_custom_backend() -> None:
+    from dreadnode.agent import Agent
+
+    backend = LogNotificationBackend()
+    agent = Agent(name="test", notifications=backend)
+
+    # Should have auto-injected a notification hook
+    assert len(agent.hooks) == 1
+
+
+def test_agent_no_notifications_by_default() -> None:
+    from dreadnode.agent import Agent
+
+    agent = Agent(name="test")
+
+    # Should not have any auto-injected hooks
+    assert len(agent.hooks) == 0
+
+
+def test_agent_notifications_disabled() -> None:
+    from dreadnode.agent import Agent
+
+    agent = Agent(name="test", notifications=False)
+
+    # Should not have any auto-injected hooks
+    assert len(agent.hooks) == 0