feat: web push notifications (VAPID)

- push_utils.py: subscription storage + send helper (auto-prunes 410 endpoints)
- routers/push.py: GET /api/push/vapid-key (public), POST/DELETE /api/push/subscribe
- sw.js: push event listener shows notification; notificationclick focuses/opens tab
- app.js: subscribe/unsubscribe flow + "Enable notifications" toggle in settings dropdown
- tools/notify.py: web_push orchestrator tool (user-level, no admin required)
- VAPID keys in .env; pywebpush added to requirements.txt

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Scott Idem
2026-05-05 19:38:58 -04:00
parent 0b96772fa6
commit ddf44a2aee
14 changed files with 350 additions and 13 deletions

View File

@@ -63,7 +63,7 @@ from tools.scratch import (
scratch_append as _scratch_append,
scratch_clear as _scratch_clear,
)
from tools.notify import nc_talk_send as _nc_talk_send, email_send as _email_send
from tools.notify import nc_talk_send as _nc_talk_send, email_send as _email_send, web_push as _web_push
# ── Declaration imports ───────────────────────────────────────────────────────
@@ -89,7 +89,7 @@ TOOL_CATEGORIES: dict[str, list[str]] = {
"Cron": ["cron_list", "cron_add", "cron_remove", "cron_toggle"],
"Reminders": ["reminders_add", "reminders_list", "reminders_remove", "reminders_clear"],
"Scratchpad": ["scratch_read", "scratch_write", "scratch_append", "scratch_clear"],
"Notifications": ["email_send", "nc_talk_send"],
"Notifications": ["web_push", "email_send", "nc_talk_send"],
"Aether Journals": [
"ae_journal_list", "ae_journal_search",
"ae_journal_entries_list", "ae_journal_entry_read",
@@ -142,6 +142,7 @@ _CALLABLES: dict[str, callable] = {
"scratch_clear": _scratch_clear,
"email_send": _email_send,
"nc_talk_send": _nc_talk_send,
"web_push": _web_push,
}
# ── Role-based access control ─────────────────────────────────────────────────

View File

@@ -67,6 +67,16 @@ async def email_send(to: str, subject: str, body: str) -> str:
return "Failed to send email — check SMTP configuration in .env."
async def web_push(title: str, body: str, url: str = "") -> str:
"""Send a browser push notification to the current user's registered devices."""
import push_utils
username = get_user()
result = await push_utils.send_push(username, title, body, url)
if "error" in result:
return f"Push failed: {result['error']}"
return f"Push sent to {result['sent']} device(s) for {username} (pruned {result['pruned']} stale)."
async def nc_talk_send(message: str) -> str:
"""Send a message to the user via their configured notification channel.
@@ -84,6 +94,24 @@ async def nc_talk_send(message: str) -> str:
DECLARATIONS = [
types.FunctionDeclaration(
name="web_push",
description=(
"Send a browser push notification to the current user. Works even when the "
"Cortex tab is not open. Use for completing long tasks, reminders that fire "
"in the background, or anything the user should see immediately. "
"url is optional — if set, clicking the notification opens that URL."
),
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"title": types.Schema(type=types.Type.STRING, description="Notification title (short)"),
"body": types.Schema(type=types.Type.STRING, description="Notification body text"),
"url": types.Schema(type=types.Type.STRING, description="Optional URL to open on click"),
},
required=["title", "body"],
),
),
types.FunctionDeclaration(
name="email_send",
description=(