feat: TTS/chime/beep alert system with notifications

python/api/alert_test.py

@@ -0,0 +1,26 @@
+from python.helpers.api import ApiHandler, Request, Response
+from agent import AgentContext
+from python.helpers.alert import emit_alert, AlertType
+from typing import cast
+
+
+class AlertTest(ApiHandler):
+    async def process(self, input: dict, request: Request) -> dict | Response:
+        alert_type = input.get("alert_type", "")
+        if alert_type not in ("task_complete", "input_needed", "subagent_complete"):
+            return {"success": False, "error": "Invalid alert_type"}
+
+        nm = AgentContext.get_notification_manager()
+        start = len(nm.updates)
+
+        # Emit as an alert.* notification, respecting current alert settings.
+        emit_alert(cast(AlertType, alert_type))
+
+        return {
+            "success": True,
+            "notifications": nm.output(start=start),
+            "notifications_guid": nm.guid,
+            "notifications_version": len(nm.updates),
+        }
+
+

python/extensions/monologue_end/_91_alert_subagent_complete.py

@@ -0,0 +1,18 @@
+from python.helpers.alert import emit_alert
+from python.helpers.extension import Extension
+from agent import AgentContextType, LoopData
+
+
+class AlertSubagentComplete(Extension):
+    async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
+        # Only subagents should emit this alert
+        if self.agent.number <= 0:
+            return
+
+        # Never alert for background contexts
+        if self.agent.context and self.agent.context.type == AgentContextType.BACKGROUND:
+            return
+
+        emit_alert("subagent_complete")
+
+

python/extensions/tool_execute_after/_80_alert_task_complete.py

@@ -0,0 +1,37 @@
+from python.helpers.alert import emit_alert
+from python.helpers.extension import Extension
+from python.helpers.task_scheduler import TaskScheduler
+from python.helpers.tool import Response
+from agent import AgentContextType
+
+
+class AlertTaskComplete(Extension):
+    async def execute(
+        self,
+        response: Response | None = None,
+        tool_name: str | None = None,
+        **kwargs,
+    ):
+        # Only main agent should emit alerts
+        if self.agent.number != 0:
+            return
+
+        # Never alert for background contexts
+        if self.agent.context and self.agent.context.type == AgentContextType.BACKGROUND:
+            return
+
+        # Only when the agent finishes via the response tool
+        if tool_name != "response":
+            return
+        if not response or not getattr(response, "break_loop", False):
+            return
+
+        # Only for scheduler task contexts
+        scheduler = TaskScheduler.get()
+        task = scheduler.get_task_by_uuid(self.agent.context.id)
+        if not task:
+            return
+
+        emit_alert("task_complete")
+
+

python/extensions/tool_execute_after/_81_alert_input_needed.py

@@ -0,0 +1,37 @@
+from python.helpers.alert import emit_alert
+from python.helpers.extension import Extension
+from python.helpers.task_scheduler import TaskScheduler
+from python.helpers.tool import Response
+from agent import AgentContextType
+
+
+class AlertInputNeeded(Extension):
+    async def execute(
+        self,
+        response: Response | None = None,
+        tool_name: str | None = None,
+        **kwargs,
+    ):
+        # Only main agent should emit alerts
+        if self.agent.number != 0:
+            return
+
+        # Never alert for background contexts
+        if self.agent.context and self.agent.context.type == AgentContextType.BACKGROUND:
+            return
+
+        # Only when the agent finishes via the response tool (end of a chat turn)
+        if tool_name != "response":
+            return
+        if not response or not getattr(response, "break_loop", False):
+            return
+
+        # Exclude scheduler task contexts (those are handled by task-complete alert)
+        scheduler = TaskScheduler.get()
+        task = scheduler.get_task_by_uuid(self.agent.context.id)
+        if task:
+            return
+
+        emit_alert("input_needed")
+
+

python/helpers/alert.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from python.helpers import settings
+from python.helpers.notification import (
+    NotificationManager,
+    NotificationPriority,
+    NotificationType,
+)
+
+
+AlertType = Literal["task_complete", "input_needed", "subagent_complete"]
+
+
+def _should_emit_alert(sett: settings.Settings, alert_type: AlertType) -> bool:
+    if not sett.get("alert_enabled", False):
+        return False
+
+    if alert_type == "task_complete":
+        return bool(sett.get("alert_on_task_complete", False))
+    if alert_type == "input_needed":
+        return bool(sett.get("alert_on_user_input_needed", False))
+    if alert_type == "subagent_complete":
+        return bool(sett.get("alert_on_subagent_complete", False))
+
+    return False
+
+
+def _get_default_message(alert_type: AlertType) -> str:
+    if alert_type == "task_complete":
+        return "Task completed"
+    if alert_type == "input_needed":
+        return "Waiting for your input"
+    if alert_type == "subagent_complete":
+        return "Subordinate agent completed"
+    return "Alert"
+
+
+def _get_message(sett: settings.Settings, alert_type: AlertType) -> str:
+    if alert_type == "task_complete":
+        return str(sett.get("alert_tts_message_task_complete") or "").strip()
+    if alert_type == "input_needed":
+        return str(sett.get("alert_tts_message_input_needed") or "").strip()
+    if alert_type == "subagent_complete":
+        return str(sett.get("alert_tts_message_subagent_complete") or "").strip()
+    return ""
+
+
+def _get_title(alert_type: AlertType) -> str:
+    if alert_type == "task_complete":
+        return "Task complete"
+    if alert_type == "input_needed":
+        return "Waiting for input"
+    if alert_type == "subagent_complete":
+        return "Subagent complete"
+    return "Alert"
+
+
+def _get_notification_type(alert_type: AlertType) -> NotificationType:
+    if alert_type == "task_complete":
+        return NotificationType.SUCCESS
+    if alert_type == "input_needed":
+        return NotificationType.INFO
+    if alert_type == "subagent_complete":
+        return NotificationType.SUCCESS
+    return NotificationType.INFO
+
+
+def emit_alert(
+    alert_type: AlertType,
+    *,
+    force: bool = False,
+    message_override: str | None = None,
+    display_time: int = 5,
+) -> None:
+    """
+    Emit an alert notification (no audio playback server-side).
+
+    WebUI will observe notifications via /poll and play sound/tts based on its settings.
+    """
+    sett = settings.get_settings()
+    if not force and not _should_emit_alert(sett, alert_type):
+        return
+
+    group = f"alert.{alert_type}"
+    message = (message_override or _get_message(sett, alert_type)).strip()
+    if not message:
+        message = _get_default_message(alert_type)
+
+    NotificationManager.send_notification(
+        _get_notification_type(alert_type),
+        NotificationPriority.NORMAL,
+        message=message,
+        title=_get_title(alert_type),
+        detail="",
+        display_time=display_time,
+        group=group,
+    )
+
+