Phase 5: Notifications — WebSocket, APScheduler, AI tool, health review

Backend:
- Notification model + Alembic migration
- Notification service: CRUD, mark read, unread count, pending scheduled
- WebSocket manager singleton for real-time push
- WebSocket endpoint /ws/notifications with JWT auth via query param
- APScheduler integration: periodic notification sender (every 60s),
  daily proactive health review job (8 AM)
- AI tool: schedule_notification (immediate or scheduled)
- Health review worker: analyzes user memory via Claude, creates
  ai_generated notifications with WebSocket push

Frontend:
- Notification API client + Zustand store
- WebSocket hook with auto-reconnect (exponential backoff)
- Notification bell in header with unread count badge + dropdown
- Notifications page with type badges, mark read, mark all read
- WebSocket initialized in AppLayout for app-wide real-time updates
- Enabled notifications nav in sidebar
- English + Russian translations

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend/app/services/ai_service.py

@@ -14,6 +14,8 @@ from app.services.context_service import DEFAULT_SYSTEM_PROMPT, get_primary_cont
 from app.services.chat_service import get_chat, save_message
 from app.services.memory_service import get_critical_memories, create_memory, get_user_memories
 from app.services.document_service import search_documents
+from app.services.notification_service import create_notification
+from app.services.ws_manager import manager
 
 client = AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY)
 
@@ -68,6 +70,28 @@ AI_TOOLS = [
             "required": [],
         },
     },
+    {
+        "name": "schedule_notification",
+        "description": "Schedule a notification or reminder for the user. Can be immediate or scheduled for a future time.",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "title": {"type": "string", "description": "Notification title"},
+                "body": {"type": "string", "description": "Notification body text"},
+                "scheduled_at": {
+                    "type": "string",
+                    "description": "ISO 8601 datetime for scheduled delivery. Omit for immediate.",
+                },
+                "type": {
+                    "type": "string",
+                    "enum": ["reminder", "alert", "info"],
+                    "description": "Notification type",
+                    "default": "reminder",
+                },
+            },
+            "required": ["title", "body"],
+        },
+    },
 ]
 
 
@@ -104,6 +128,40 @@ async def _execute_tool(
         items = [{"category": e.category, "title": e.title, "content": e.content, "importance": e.importance} for e in entries]
         return json.dumps({"entries": items, "count": len(items)})
 
+    elif tool_name == "schedule_notification":
+        from datetime import datetime
+        scheduled_at = None
+        if tool_input.get("scheduled_at"):
+            scheduled_at = datetime.fromisoformat(tool_input["scheduled_at"])
+
+        notif = await create_notification(
+            db, user_id,
+            title=tool_input["title"],
+            body=tool_input["body"],
+            type=tool_input.get("type", "reminder"),
+            scheduled_at=scheduled_at,
+        )
+        await db.commit()
+
+        # Push immediately if not scheduled
+        if not scheduled_at:
+            await manager.send_to_user(user_id, {
+                "type": "new_notification",
+                "notification": {
+                    "id": str(notif.id),
+                    "title": notif.title,
+                    "body": notif.body,
+                    "type": notif.type,
+                    "created_at": notif.created_at.isoformat(),
+                },
+            })
+
+        return json.dumps({
+            "status": "scheduled" if scheduled_at else "sent",
+            "id": str(notif.id),
+            "title": notif.title,
+        })
+
     return json.dumps({"error": f"Unknown tool: {tool_name}"})
 
 

backend/app/services/notification_service.py

@@ -0,0 +1,102 @@
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import HTTPException, status
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.notification import Notification
+
+
+async def create_notification(
+    db: AsyncSession,
+    user_id: uuid.UUID,
+    title: str,
+    body: str,
+    type: str = "info",
+    channel: str = "in_app",
+    scheduled_at: datetime | None = None,
+    metadata: dict | None = None,
+) -> Notification:
+    notif = Notification(
+        user_id=user_id,
+        title=title,
+        body=body,
+        type=type,
+        channel=channel,
+        status="pending" if scheduled_at else "sent",
+        scheduled_at=scheduled_at,
+        sent_at=None if scheduled_at else datetime.now(timezone.utc),
+        metadata_=metadata,
+    )
+    db.add(notif)
+    await db.flush()
+    return notif
+
+
+async def get_user_notifications(
+    db: AsyncSession, user_id: uuid.UUID,
+    status_filter: str | None = None, limit: int = 50, offset: int = 0,
+) -> list[Notification]:
+    stmt = select(Notification).where(Notification.user_id == user_id)
+    if status_filter:
+        stmt = stmt.where(Notification.status == status_filter)
+    stmt = stmt.order_by(Notification.created_at.desc()).limit(limit).offset(offset)
+    result = await db.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def get_unread_count(db: AsyncSession, user_id: uuid.UUID) -> int:
+    result = await db.scalar(
+        select(func.count()).select_from(Notification).where(
+            Notification.user_id == user_id,
+            Notification.status.in_(["sent", "delivered"]),
+            Notification.read_at.is_(None),
+        )
+    )
+    return result or 0
+
+
+async def mark_as_read(db: AsyncSession, notification_id: uuid.UUID, user_id: uuid.UUID) -> Notification:
+    result = await db.execute(
+        select(Notification).where(Notification.id == notification_id, Notification.user_id == user_id)
+    )
+    notif = result.scalar_one_or_none()
+    if not notif:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Notification not found")
+    notif.status = "read"
+    notif.read_at = datetime.now(timezone.utc)
+    await db.flush()
+    return notif
+
+
+async def mark_all_read(db: AsyncSession, user_id: uuid.UUID) -> int:
+    result = await db.execute(
+        update(Notification)
+        .where(
+            Notification.user_id == user_id,
+            Notification.read_at.is_(None),
+        )
+        .values(status="read", read_at=datetime.now(timezone.utc))
+    )
+    return result.rowcount
+
+
+async def get_pending_scheduled(db: AsyncSession) -> list[Notification]:
+    now = datetime.now(timezone.utc)
+    result = await db.execute(
+        select(Notification).where(
+            Notification.status == "pending",
+            Notification.scheduled_at <= now,
+        )
+    )
+    return list(result.scalars().all())
+
+
+async def mark_as_sent(db: AsyncSession, notification_id: uuid.UUID) -> None:
+    result = await db.execute(select(Notification).where(Notification.id == notification_id))
+    notif = result.scalar_one_or_none()
+    if notif:
+        notif.status = "sent"
+        notif.sent_at = datetime.now(timezone.utc)
+        await db.flush()

backend/app/services/scheduler_service.py

@@ -0,0 +1,41 @@
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.date import DateTrigger
+
+scheduler = AsyncIOScheduler()
+
+
+def start_scheduler():
+    if not scheduler.running:
+        # Register periodic jobs
+        from app.workers.notification_sender import send_pending_notifications
+        scheduler.add_job(
+            send_pending_notifications,
+            trigger=CronTrigger(second="0", minute="*"),  # every minute
+            id="send_pending_notifications",
+            replace_existing=True,
+        )
+
+        from app.workers.health_review import run_daily_health_review
+        scheduler.add_job(
+            run_daily_health_review,
+            trigger=CronTrigger(hour=8, minute=0),
+            id="daily_health_review",
+            replace_existing=True,
+        )
+
+        scheduler.start()
+
+
+def shutdown_scheduler():
+    if scheduler.running:
+        scheduler.shutdown(wait=False)
+
+
+def schedule_one_time(job_id: str, run_date, func, **kwargs):
+    scheduler.add_job(func, trigger=DateTrigger(run_date=run_date), id=job_id, replace_existing=True, kwargs=kwargs)
+
+
+def schedule_recurring(job_id: str, cron_expr: str, func, **kwargs):
+    trigger = CronTrigger.from_crontab(cron_expr)
+    scheduler.add_job(func, trigger=trigger, id=job_id, replace_existing=True, kwargs=kwargs)

backend/app/services/ws_manager.py

@@ -0,0 +1,37 @@
+import json
+import uuid
+
+from fastapi import WebSocket
+
+
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: dict[uuid.UUID, list[WebSocket]] = {}
+
+    async def connect(self, user_id: uuid.UUID, websocket: WebSocket):
+        await websocket.accept()
+        if user_id not in self.active_connections:
+            self.active_connections[user_id] = []
+        self.active_connections[user_id].append(websocket)
+
+    def disconnect(self, user_id: uuid.UUID, websocket: WebSocket):
+        if user_id in self.active_connections:
+            self.active_connections[user_id] = [
+                ws for ws in self.active_connections[user_id] if ws != websocket
+            ]
+            if not self.active_connections[user_id]:
+                del self.active_connections[user_id]
+
+    async def send_to_user(self, user_id: uuid.UUID, data: dict):
+        connections = self.active_connections.get(user_id, [])
+        dead = []
+        for ws in connections:
+            try:
+                await ws.send_text(json.dumps(data))
+            except Exception:
+                dead.append(ws)
+        for ws in dead:
+            self.disconnect(user_id, ws)
+
+
+manager = ConnectionManager()