oai-web/server/tools/pushover_tool.py

"""
tools/pushover_tool.py — Pushover push notifications.

Sends to exactly one hard-coded user key (defined in security.py).
Normal priority (0) and below: no confirmation required.
Emergency priority (2): always requires confirmation.
"""
from __future__ import annotations

import httpx

from ..database import credential_store
from .base import BaseTool, ToolResult

PUSHOVER_API_URL = "https://api.pushover.net/1/messages.json"


class PushoverTool(BaseTool):
    name = "pushover"
    description = (
        "Send a push notification to the owner's phone via Pushover. "
        "Use for alerts, reminders, and status updates. "
        "Priority: -2 (silent), -1 (quiet), 0 (normal), 1 (high), 2 (emergency — requires confirmation)."
    )
    input_schema = {
        "type": "object",
        "properties": {
            "title": {
                "type": "string",
                "description": "Notification title (short, shown in bold)",
            },
            "message": {
                "type": "string",
                "description": "Notification body text",
            },
            "priority": {
                "type": "integer",
                "enum": [-2, -1, 0, 1, 2],
                "description": "Priority level (-2 silent to 2 emergency). Default: 0",
            },
            "url": {
                "type": "string",
                "description": "Optional URL to attach to the notification",
            },
            "url_title": {
                "type": "string",
                "description": "Display text for the attached URL",
            },
        },
        "required": ["title", "message"],
    }
    requires_confirmation = False   # overridden dynamically for priority=2
    allowed_in_scheduled_tasks = True

    async def execute(
        self,
        title: str,
        message: str,
        priority: int = 0,
        url: str = "",
        url_title: str = "",
        **kwargs,
    ) -> ToolResult:
        # Validate priority
        if priority not in (-2, -1, 0, 1, 2):
            return ToolResult(success=False, error=f"Invalid priority: {priority}")

        # Emergency always requires confirmation — enforced here as defence-in-depth
        # (the agent loop also checks requires_confirmation before calling execute)
        if priority == 2:
            # The agent loop should have asked for confirmation already.
            # If we got here, it was approved.
            pass

        # Load credentials — per-user key first, then system fallback
        try:
            app_token = await credential_store.require("pushover_app_token")
        except RuntimeError as e:
            return ToolResult(success=False, error=str(e))

        user_key: str | None = None
        try:
            from ..context_vars import current_user as _cu
            u = _cu.get()
            if u:
                from ..database import user_settings_store as _us
                user_key = await _us.get(u.id, "pushover_user_key")
        except Exception:
            pass
        if not user_key:
            user_key = await credential_store.get("pushover_user_key")
        if not user_key:
            return ToolResult(success=False, error="Pushover user key not configured. Set it in Settings → Pushover.")

        payload: dict = {
            "token": app_token,
            "user": user_key,
            "title": title[:250],
            "message": message[:1024],
            "priority": priority,
        }

        if priority == 2:
            # Emergency: retry every 30s for 1 hour until acknowledged
            payload["retry"] = 30
            payload["expire"] = 3600

        if url:
            payload["url"] = url[:512]
        if url_title:
            payload["url_title"] = url_title[:100]

        try:
            async with httpx.AsyncClient(timeout=10) as client:
                resp = await client.post(PUSHOVER_API_URL, data=payload)
                resp.raise_for_status()
                data = resp.json()

            if data.get("status") != 1:
                return ToolResult(
                    success=False,
                    error=f"Pushover API error: {data.get('errors', 'unknown')}",
                )

            return ToolResult(
                success=True,
                data={"sent": True, "request_id": data.get("request", "")},
            )

        except httpx.HTTPStatusError as e:
            return ToolResult(success=False, error=f"Pushover HTTP error: {e.response.status_code}")
        except Exception as e:
            return ToolResult(success=False, error=f"Pushover error: {e}")

    def confirmation_description(self, title: str = "", message: str = "", priority: int = 0, **kwargs) -> str:
        return f"Send emergency Pushover notification: '{title}' — {message[:100]}"

    def get_schema(self) -> dict:
        """Override to make requires_confirmation dynamic for priority 2."""
        schema = super().get_schema()
        # The tool itself handles emergency confirmation — marked as requiring it always
        # so the agent loop treats it consistently. For non-emergency, the agent loop
        # still calls execute(), which works fine without confirmation.
        return schema