feat(notifications): add Phase 4 Slice 2 — multi-channel notification system

Full notification infrastructure with in-app, email, Slack, and Teams channels:

Backend:
- NotificationConfig, NotificationLog, Notification models + migration
- Notification service with event routing, channel delivery, retry logic
- 9 API endpoints (config CRUD + in-app notifications)
- APScheduler retry job with exponential backoff (30s, 2m, 10m)
- Wired into escalation, proposal approval, and knowledge flywheel
- Pydantic event key validation, cross-tenant protection on recipients

Frontend:
- TypeScript types + API client for all notification endpoints
- NotificationsPanel: bell icon with unread badge, dropdown, mark-read
- NotificationSettings: channel config, event toggles, test, delete confirm
- Notifications tab on IntegrationsPage
- ARIA attributes, Escape handler, settings link on panel

Review fixes (13 issues resolved):
- notify() no longer commits/rolls back caller's transaction (critical)
- retry_failed_notifications returns count instead of None (critical)
- NotificationSettings moved inside dedicated tab (critical)
- target_user_ids scoped by account_id (security)
- Email loop collects all failures before raising
- Slack webhook validates response body
- events_enabled rejects unknown event keys
- link column widened to String(500)
- Dead code removed from _auto_reinforce
- Delete confirmation, ARIA, Escape key support

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

backend/alembic/versions/b09c3789b7e6_add_notification_tables.py

@@ -0,0 +1,83 @@
+"""add notification tables
+
+Revision ID: b09c3789b7e6
+Revises: 3266dd9d8111
+Create Date: 2026-03-19 06:16:46.817718
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = 'b09c3789b7e6'
+down_revision: Union[str, None] = '3266dd9d8111'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table('notification_configs',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('account_id', sa.UUID(), nullable=False),
+    sa.Column('channel', sa.String(length=20), nullable=False),
+    sa.Column('webhook_url', sa.String(length=500), nullable=True),
+    sa.Column('email_addresses', postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+    sa.Column('is_active', sa.Boolean(), nullable=False),
+    sa.Column('events_enabled', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+    sa.CheckConstraint("channel IN ('email', 'slack_webhook', 'teams_webhook')", name='ck_notification_configs_channel'),
+    sa.ForeignKeyConstraint(['account_id'], ['accounts.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_notification_configs_account_id'), 'notification_configs', ['account_id'], unique=False)
+
+    op.create_table('notifications',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('account_id', sa.UUID(), nullable=False),
+    sa.Column('user_id', sa.UUID(), nullable=False),
+    sa.Column('event', sa.String(length=50), nullable=False),
+    sa.Column('title', sa.String(length=200), nullable=False),
+    sa.Column('body', sa.String(length=500), nullable=True),
+    sa.Column('link', sa.String(length=500), nullable=True),
+    sa.Column('is_read', sa.Boolean(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.ForeignKeyConstraint(['account_id'], ['accounts.id'], ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_notifications_account_id'), 'notifications', ['account_id'], unique=False)
+    op.create_index(op.f('ix_notifications_created_at'), 'notifications', ['created_at'], unique=False)
+    op.create_index(op.f('ix_notifications_user_id'), 'notifications', ['user_id'], unique=False)
+
+    op.create_table('notification_logs',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('notification_config_id', sa.UUID(), nullable=False),
+    sa.Column('event', sa.String(length=50), nullable=False),
+    sa.Column('payload', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('status', sa.String(length=20), nullable=False),
+    sa.Column('retry_count', sa.Integer(), nullable=False),
+    sa.Column('max_retries', sa.Integer(), nullable=False),
+    sa.Column('last_error', sa.String(length=1000), nullable=True),
+    sa.Column('next_retry_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('delivered_at', sa.DateTime(timezone=True), nullable=True),
+    sa.CheckConstraint("status IN ('sent', 'failed', 'retrying', 'exhausted')", name='ck_notification_logs_status'),
+    sa.ForeignKeyConstraint(['notification_config_id'], ['notification_configs.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_notification_logs_notification_config_id'), 'notification_logs', ['notification_config_id'], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index(op.f('ix_notification_logs_notification_config_id'), table_name='notification_logs')
+    op.drop_table('notification_logs')
+    op.drop_index(op.f('ix_notifications_user_id'), table_name='notifications')
+    op.drop_index(op.f('ix_notifications_created_at'), table_name='notifications')
+    op.drop_index(op.f('ix_notifications_account_id'), table_name='notifications')
+    op.drop_table('notifications')
+    op.drop_index(op.f('ix_notification_configs_account_id'), table_name='notification_configs')
+    op.drop_table('notification_configs')

backend/app/api/endpoints/flow_proposals.py

@@ -18,6 +18,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from app.core.rate_limit import limiter
+from app.services.notification_service import notify
 from app.api.deps import get_current_active_user, get_db, require_engineer_or_admin, require_team_admin
 from app.models.user import User
 from app.models.tree import Tree
@@ -262,6 +263,13 @@ async def review_proposal(
     elif data.action == "dismiss":
         proposal.status = "dismissed"
 
+    if data.action == "approve":
+        await notify("proposal.approved", proposal.account_id, {
+            "title": proposal.title,
+            "reviewer_name": current_user.display_name if hasattr(current_user, 'display_name') else current_user.email,
+            "link": "/review-queue",
+        }, db, target_user_ids=[proposal.created_by_id] if proposal.created_by_id else None)
+
     await db.commit()
 
     return FlowProposalDetail.model_validate(proposal)

backend/app/api/endpoints/notifications.py

@@ -0,0 +1,255 @@
+"""Notification endpoints — config CRUD + in-app notification management.
+
+Config CRUD (team_admin):
+  GET    /notifications/configs          — List configs for account
+  POST   /notifications/configs          — Create config
+  PATCH  /notifications/configs/{id}     — Update config
+  DELETE /notifications/configs/{id}     — Delete config
+  POST   /notifications/configs/test     — Test a config
+
+In-app notifications (any authenticated user):
+  GET    /notifications                  — List notifications (paginated)
+  GET    /notifications/unread-count     — Unread count
+  PATCH  /notifications/{id}/read        — Mark one as read
+  POST   /notifications/mark-all-read    — Mark all as read
+"""
+import logging
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from sqlalchemy import select, func, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.rate_limit import limiter
+from app.api.deps import get_current_active_user, require_team_admin
+from app.core.database import get_db
+from app.models.user import User
+from app.models.notification_config import NotificationConfig
+from app.models.notification import Notification
+from app.schemas.notification import (
+    NotificationConfigCreate,
+    NotificationConfigUpdate,
+    NotificationConfigResponse,
+    NotificationResponse,
+    UnreadCountResponse,
+    NotificationTestRequest,
+    NotificationTestResponse,
+)
+from app.services.notification_service import send_test_notification
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/notifications", tags=["notifications"])
+
+
+# ---------------------------------------------------------------------------
+# Config CRUD (team_admin required)
+# ---------------------------------------------------------------------------
+
+
+@router.get("/configs", response_model=list[NotificationConfigResponse])
+@limiter.limit("30/minute")
+async def list_configs(
+    request: Request,
+    current_user: Annotated[User, Depends(require_team_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """List all notification configs for the current account."""
+    result = await db.execute(
+        select(NotificationConfig)
+        .where(NotificationConfig.account_id == current_user.account_id)
+        .order_by(NotificationConfig.created_at.desc())
+    )
+    return result.scalars().all()
+
+
+@router.post("/configs", response_model=NotificationConfigResponse, status_code=status.HTTP_201_CREATED)
+@limiter.limit("10/minute")
+async def create_config(
+    request: Request,
+    body: NotificationConfigCreate,
+    current_user: Annotated[User, Depends(require_team_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Create a new notification config."""
+    # Validate channel-specific requirements
+    if body.channel in ("slack_webhook", "teams_webhook") and not body.webhook_url:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"webhook_url is required for {body.channel} channel",
+        )
+    if body.channel == "email" and not body.email_addresses:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="email_addresses is required for email channel",
+        )
+
+    config = NotificationConfig(
+        account_id=current_user.account_id,
+        channel=body.channel,
+        webhook_url=body.webhook_url,
+        email_addresses=body.email_addresses,
+        events_enabled=body.events_enabled,
+    )
+    db.add(config)
+    await db.commit()
+    await db.refresh(config)
+    return config
+
+
+@router.patch("/configs/{config_id}", response_model=NotificationConfigResponse)
+@limiter.limit("20/minute")
+async def update_config(
+    request: Request,
+    config_id: UUID,
+    body: NotificationConfigUpdate,
+    current_user: Annotated[User, Depends(require_team_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Update an existing notification config."""
+    result = await db.execute(
+        select(NotificationConfig)
+        .where(NotificationConfig.id == config_id)
+        .where(NotificationConfig.account_id == current_user.account_id)
+    )
+    config = result.scalar_one_or_none()
+    if not config:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Config not found")
+
+    update_data = body.model_dump(exclude_unset=True)
+    for field, value in update_data.items():
+        setattr(config, field, value)
+
+    await db.commit()
+    await db.refresh(config)
+    return config
+
+
+@router.delete("/configs/{config_id}", status_code=status.HTTP_204_NO_CONTENT)
+@limiter.limit("10/minute")
+async def delete_config(
+    request: Request,
+    config_id: UUID,
+    current_user: Annotated[User, Depends(require_team_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Delete a notification config."""
+    result = await db.execute(
+        select(NotificationConfig)
+        .where(NotificationConfig.id == config_id)
+        .where(NotificationConfig.account_id == current_user.account_id)
+    )
+    config = result.scalar_one_or_none()
+    if not config:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Config not found")
+
+    await db.delete(config)
+    await db.commit()
+
+
+@router.post("/configs/test", response_model=NotificationTestResponse)
+@limiter.limit("5/minute")
+async def test_config(
+    request: Request,
+    body: NotificationTestRequest,
+    current_user: Annotated[User, Depends(require_team_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Send a test notification through a config."""
+    result = await db.execute(
+        select(NotificationConfig)
+        .where(NotificationConfig.id == body.config_id)
+        .where(NotificationConfig.account_id == current_user.account_id)
+    )
+    config = result.scalar_one_or_none()
+    if not config:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Config not found")
+
+    success, message = await send_test_notification(config)
+    return NotificationTestResponse(success=success, message=message)
+
+
+# ---------------------------------------------------------------------------
+# In-app notifications (any authenticated user)
+# ---------------------------------------------------------------------------
+
+
+@router.get("", response_model=list[NotificationResponse])
+@limiter.limit("60/minute")
+async def list_notifications(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+):
+    """List notifications for the current user, unread first."""
+    result = await db.execute(
+        select(Notification)
+        .where(Notification.user_id == current_user.id)
+        .order_by(Notification.is_read.asc(), Notification.created_at.desc())
+        .offset(skip)
+        .limit(limit)
+    )
+    return result.scalars().all()
+
+
+@router.get("/unread-count", response_model=UnreadCountResponse)
+@limiter.limit("120/minute")
+async def unread_count(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Get count of unread notifications for the current user."""
+    result = await db.execute(
+        select(func.count())
+        .select_from(Notification)
+        .where(Notification.user_id == current_user.id)
+        .where(Notification.is_read.is_(False))
+    )
+    count = result.scalar_one()
+    return UnreadCountResponse(count=count)
+
+
+@router.patch("/{notification_id}/read", response_model=NotificationResponse)
+@limiter.limit("60/minute")
+async def mark_read(
+    request: Request,
+    notification_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Mark a single notification as read."""
+    result = await db.execute(
+        select(Notification)
+        .where(Notification.id == notification_id)
+        .where(Notification.user_id == current_user.id)
+    )
+    notification = result.scalar_one_or_none()
+    if not notification:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Notification not found")
+
+    notification.is_read = True
+    await db.commit()
+    await db.refresh(notification)
+    return notification
+
+
+@router.post("/mark-all-read", response_model=UnreadCountResponse)
+@limiter.limit("10/minute")
+async def mark_all_read(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Mark all notifications as read for the current user."""
+    await db.execute(
+        update(Notification)
+        .where(Notification.user_id == current_user.id)
+        .where(Notification.is_read.is_(False))
+        .values(is_read=True)
+    )
+    await db.commit()
+    return UnreadCountResponse(count=0)

backend/app/api/router.py

@@ -24,6 +24,7 @@ from app.api.endpoints import supporting_data
 from app.api.endpoints import ai_sessions
 from app.api.endpoints import flow_proposals
 from app.api.endpoints import flowpilot_analytics
+from app.api.endpoints import notifications
 
 api_router = APIRouter()
 
@@ -73,3 +74,4 @@ api_router.include_router(supporting_data.router)
 api_router.include_router(ai_sessions.router)
 api_router.include_router(flow_proposals.router)
 api_router.include_router(flowpilot_analytics.router)
+api_router.include_router(notifications.router)

backend/app/core/email.py

@@ -484,6 +484,45 @@ class EmailService:
             logger.exception("Failed to send beta signup notification for %s", signup_email)
             return False
 
+    @staticmethod
+    async def send_notification_email(
+        to_email: str,
+        title: str,
+        body: str,
+        link_url: str | None = None,
+    ) -> bool:
+        """Send a notification email. Fire-and-forget."""
+        if not settings.email_enabled:
+            logger.warning("Email not sent — RESEND_API_KEY not configured")
+            return False
+
+        try:
+            import resend
+
+            resend.api_key = settings.RESEND_API_KEY
+
+            subject = f"[ResolutionFlow] {title}"
+            html = _render_notification_html(
+                title=title,
+                body=body,
+                link_url=link_url,
+            )
+
+            resend.Emails.send(
+                {
+                    "from": settings.FROM_EMAIL,
+                    "to": [to_email],
+                    "subject": subject,
+                    "html": html,
+                }
+            )
+            logger.info("Notification email sent to %s: %s", to_email, title)
+            return True
+
+        except Exception:
+            logger.exception("Failed to send notification email to %s", to_email)
+            return False
+
     @staticmethod
     async def send_survey_invite_email(
         to_email: str,
@@ -856,3 +895,49 @@ def _render_feedback_confirmation_html(
 </td></tr>
 </table>
 </body></html>"""
+
+
+def _render_notification_html(
+    title: str,
+    body: str,
+    link_url: str | None = None,
+) -> str:
+    import html as html_mod
+
+    safe_title = html_mod.escape(title)
+    safe_body = html_mod.escape(body)
+
+    link_section = ""
+    if link_url:
+        link_section = f"""
+  <tr><td style="padding:0 40px 32px;text-align:center;">
+    <a href="{link_url}" style="display:inline-block;background:linear-gradient(135deg,#06b6d4,#22d3ee);color:#101114;font-size:16px;font-weight:600;text-decoration:none;padding:14px 40px;border-radius:10px;">
+      View in ResolutionFlow
+    </a>
+  </td></tr>"""
+
+    return f"""<!DOCTYPE html>
+<html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width"></head>
+<body style="margin:0;padding:0;background:#101114;font-family:'Inter',Helvetica,Arial,sans-serif;">
+<table width="100%" cellpadding="0" cellspacing="0" style="background:#101114;padding:40px 0;">
+<tr><td align="center">
+<table width="560" cellpadding="0" cellspacing="0" style="background:#14161a;border:1px solid rgba(255,255,255,0.06);border-radius:16px;">
+  <tr><td style="padding:40px 40px 24px;text-align:center;">
+    <h1 style="margin:0;color:#f8fafc;font-size:24px;font-weight:600;">Resolution<span style="color:#06b6d4;">Flow</span></h1>
+  </td></tr>
+  <tr><td style="padding:0 40px 12px;">
+    <h2 style="margin:0;color:#f8fafc;font-size:18px;font-weight:600;">{safe_title}</h2>
+  </td></tr>
+  <tr><td style="padding:0 40px 24px;">
+    <p style="margin:0;color:#8891a0;font-size:16px;line-height:1.6;">{safe_body}</p>
+  </td></tr>
+  {link_section}
+  <tr><td style="padding:0 40px 32px;">
+    <p style="margin:0;color:#5a6170;font-size:12px;text-align:center;">
+      &mdash; ResolutionFlow
+    </p>
+  </td></tr>
+</table>
+</td></tr>
+</table>
+</body></html>"""

backend/app/main.py

@@ -33,6 +33,7 @@ from app.core.rate_limit import limiter
 from app.api.router import api_router
 from app.core.scheduler import scheduler, load_all_schedules, _cleanup_expired_ai_conversations
 from app.services.retention_cleanup import cleanup_expired_chats
+from app.services.notification_service import retry_failed_notifications
 from app.core.service_account import ensure_service_account
 
 # Initialize logging configuration
@@ -61,6 +62,14 @@ async def archive_stale_ai_sessions():
         logger.info(f"[archive] Archived {result.rowcount} stale AI chat sessions")
 
 
+async def _process_notification_retries():
+    """Retry failed notification deliveries."""
+    async with async_session_maker() as db:
+        retried = await retry_failed_notifications(db)
+        if retried:
+            logger.info("Retried %d failed notifications", retried)
+
+
 def _configure_seed_module(mod: object, api_url: str, email: str, password: str) -> None:
     """Set globals on a seed script module."""
     mod.API_BASE_URL = api_url  # type: ignore[attr-defined]
@@ -201,6 +210,16 @@ async def lifespan(app: FastAPI):
         max_instances=1,
     )
 
+    # Notification retry (every minute)
+    scheduler.add_job(
+        _process_notification_retries,
+        trigger="interval",
+        minutes=1,
+        id="notification_retry",
+        replace_existing=True,
+        max_instances=1,
+    )
+
     # Auto-seed trees in background on PR environments
     seed_task = None
     if settings.SEED_ON_DEPLOY:

backend/app/models/__init__.py

@@ -43,6 +43,9 @@ from .psa_post_log import PsaPostLog
 from .psa_member_mapping import PsaMemberMapping
 from .supporting_data import SessionSupportingData
 from .flow_proposal import FlowProposal
+from .notification_config import NotificationConfig
+from .notification_log import NotificationLog
+from .notification import Notification
 
 __all__ = [
     "User",
@@ -100,4 +103,7 @@ __all__ = [
     "PsaMemberMapping",
     "SessionSupportingData",
     "FlowProposal",
+    "NotificationConfig",
+    "NotificationLog",
+    "Notification",
 ]

backend/app/models/notification.py

@@ -0,0 +1,45 @@
+"""In-app notification model."""
+import uuid
+from datetime import datetime, timezone
+from typing import Optional, TYPE_CHECKING
+
+from sqlalchemy import String, Boolean, DateTime, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.user import User
+
+
+class Notification(Base):
+    __tablename__ = "notifications"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    account_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("accounts.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    event: Mapped[str] = mapped_column(String(50), nullable=False)
+    title: Mapped[str] = mapped_column(String(200), nullable=False)
+    body: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    link: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    is_read: Mapped[bool] = mapped_column(Boolean, default=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        index=True,
+    )
+
+    user: Mapped[Optional["User"]] = relationship("User", foreign_keys=[user_id])

backend/app/models/notification_config.py

@@ -0,0 +1,60 @@
+"""Notification channel configuration per account.
+
+Each account can have multiple notification configs (email, Slack webhook, Teams webhook).
+Each config specifies which events it receives.
+"""
+import uuid
+from datetime import datetime, timezone
+from typing import Optional, Any, TYPE_CHECKING
+
+from sqlalchemy import String, Boolean, DateTime, ForeignKey, CheckConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.account import Account
+
+
+class NotificationConfig(Base):
+    __tablename__ = "notification_configs"
+    __table_args__ = (
+        CheckConstraint(
+            "channel IN ('email', 'slack_webhook', 'teams_webhook')",
+            name="ck_notification_configs_channel",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    account_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("accounts.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    channel: Mapped[str] = mapped_column(String(20), nullable=False)
+    webhook_url: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    email_addresses: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    events_enabled: Mapped[dict[str, Any]] = mapped_column(
+        JSONB, default=lambda: {
+            "session.escalated": True,
+            "session.high_priority": True,
+            "proposal.pending": True,
+            "proposal.approved": True,
+            "knowledge_gap.detected": True,
+        }
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+    )
+
+    account: Mapped[Optional["Account"]] = relationship("Account", foreign_keys=[account_id])

backend/app/models/notification_log.py

@@ -0,0 +1,52 @@
+"""Notification delivery log with retry tracking."""
+import uuid
+from datetime import datetime, timezone
+from typing import Optional, Any, TYPE_CHECKING
+
+from sqlalchemy import String, Integer, DateTime, ForeignKey, CheckConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.notification_config import NotificationConfig
+
+
+class NotificationLog(Base):
+    __tablename__ = "notification_logs"
+    __table_args__ = (
+        CheckConstraint(
+            "status IN ('sent', 'failed', 'retrying', 'exhausted')",
+            name="ck_notification_logs_status",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    notification_config_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("notification_configs.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    event: Mapped[str] = mapped_column(String(50), nullable=False)
+    payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
+    status: Mapped[str] = mapped_column(String(20), default="sent")
+    retry_count: Mapped[int] = mapped_column(Integer, default=0)
+    max_retries: Mapped[int] = mapped_column(Integer, default=3)
+    last_error: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True)
+    next_retry_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    delivered_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+
+    config: Mapped[Optional["NotificationConfig"]] = relationship(
+        "NotificationConfig", foreign_keys=[notification_config_id]
+    )

backend/app/schemas/notification.py

@@ -0,0 +1,85 @@
+"""Pydantic schemas for notification system."""
+from datetime import datetime
+from uuid import UUID
+
+from pydantic import BaseModel, Field, field_validator
+
+
+VALID_EVENTS = {
+    "session.escalated",
+    "session.high_priority",
+    "proposal.pending",
+    "proposal.approved",
+    "knowledge_gap.detected",
+}
+
+
+class NotificationConfigCreate(BaseModel):
+    channel: str = Field(..., pattern="^(email|slack_webhook|teams_webhook)$")
+    webhook_url: str | None = None
+    email_addresses: list[str] | None = None
+    events_enabled: dict[str, bool] = Field(
+        default_factory=lambda: {e: True for e in VALID_EVENTS}
+    )
+
+    @field_validator("events_enabled")
+    @classmethod
+    def validate_event_keys(cls, v: dict[str, bool]) -> dict[str, bool]:
+        invalid = set(v) - VALID_EVENTS
+        if invalid:
+            raise ValueError(f"Unknown event keys: {invalid}")
+        return v
+
+
+class NotificationConfigUpdate(BaseModel):
+    webhook_url: str | None = None
+    email_addresses: list[str] | None = None
+    is_active: bool | None = None
+    events_enabled: dict[str, bool] | None = None
+
+    @field_validator("events_enabled")
+    @classmethod
+    def validate_event_keys(cls, v: dict[str, bool] | None) -> dict[str, bool] | None:
+        if v is not None:
+            invalid = set(v) - VALID_EVENTS
+            if invalid:
+                raise ValueError(f"Unknown event keys: {invalid}")
+        return v
+
+
+class NotificationConfigResponse(BaseModel):
+    id: UUID
+    channel: str
+    webhook_url: str | None
+    email_addresses: list[str] | None
+    is_active: bool
+    events_enabled: dict[str, bool]
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class NotificationResponse(BaseModel):
+    id: UUID
+    event: str
+    title: str
+    body: str | None
+    link: str | None
+    is_read: bool
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class UnreadCountResponse(BaseModel):
+    count: int
+
+
+class NotificationTestRequest(BaseModel):
+    config_id: UUID
+
+
+class NotificationTestResponse(BaseModel):
+    success: bool
+    message: str

backend/app/services/flowpilot_engine.py

@@ -17,6 +17,7 @@ from sqlalchemy.orm import selectinload
 
 from app.core.ai_provider import get_ai_provider
 from app.core.config import settings
+from app.services.notification_service import notify
 from app.models.ai_session import AISession
 from app.models.ai_session_step import AISessionStep
 from app.schemas.ai_session import (
@@ -497,6 +498,15 @@ async def escalate_session(
 
     await db.flush()
 
+    # Notify about escalation
+    await notify("session.escalated", session.account_id, {
+        "session_id": str(session_id),
+        "engineer_name": session.user.display_name if session.user else "Unknown",
+        "escalation_reason": request.escalation_reason,
+        "problem_summary": session.problem_summary or "N/A",
+        "link": f"/pilot/{session_id}",
+    }, db, target_user_ids=[request.escalated_to_id] if request.escalated_to_id else None)
+
     # Push documentation to PSA if ticket is linked
     psa_result = await _push_to_psa(session, user_id, db)
 

backend/app/services/knowledge_flywheel.py

@@ -20,6 +20,7 @@ from sqlalchemy.orm import selectinload
 
 from app.core.ai_provider import get_ai_provider
 from app.core.config import settings
+from app.services.notification_service import notify
 from app.models.ai_session import AISession
 from app.models.ai_session_step import AISessionStep
 from app.models.flow_proposal import FlowProposal
@@ -295,6 +296,7 @@ async def _auto_reinforce(session: AISession, db: AsyncSession) -> None:
         target_flow_id=session.matched_flow_id,
     )
     db.add(proposal)
+    # auto_reinforced proposals don't need review — no notification
     logger.info("Auto-reinforced flow %s from session %s", session.matched_flow_id, session.id)
 
 
@@ -355,6 +357,12 @@ async def _propose_new_flow(session: AISession, db: AsyncSession) -> None:
         status="pending",
     )
     db.add(proposal)
+    await notify("proposal.pending", proposal.account_id, {
+        "title": proposal.title,
+        "proposal_type": proposal.proposal_type,
+        "problem_domain": proposal.problem_domain or "General",
+        "link": "/review-queue",
+    }, db)
     logger.info("Created new_flow proposal for session %s: %s", session.id, title)
 
 
@@ -431,6 +439,12 @@ async def _propose_enhancement(session: AISession, db: AsyncSession) -> None:
         status="pending",
     )
     db.add(proposal)
+    await notify("proposal.pending", proposal.account_id, {
+        "title": proposal.title,
+        "proposal_type": proposal.proposal_type,
+        "problem_domain": proposal.problem_domain or "General",
+        "link": "/review-queue",
+    }, db)
     logger.info(
         "Created enhancement proposal for flow %s from session %s: %s",
         session.matched_flow_id, session.id, title,

backend/app/services/notification_service.py

@@ -0,0 +1,420 @@
+"""Notification service — dispatches in-app + external notifications.
+
+Entry point: `notify(event, account_id, payload, db)`.
+Retry engine: `retry_failed_notifications(db)` called by APScheduler.
+"""
+import logging
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any, Optional
+
+import httpx
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.core.email import EmailService
+from app.models.notification import Notification
+from app.models.notification_config import NotificationConfig
+from app.models.notification_log import NotificationLog
+from app.models.user import User
+
+logger = logging.getLogger(__name__)
+
+# Exponential backoff schedule (seconds): 30s, 2m, 10m
+_RETRY_DELAYS = [30, 120, 600]
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+async def notify(
+    event: str,
+    account_id: uuid.UUID,
+    payload: dict[str, Any],
+    db: AsyncSession,
+    target_user_ids: Optional[list[uuid.UUID]] = None,
+) -> None:
+    """Main entry point — create in-app notifications + route to external channels.
+
+    IMPORTANT: This function does NOT commit or rollback. The caller owns the transaction.
+    In-app notifications are added to the session (flushed, not committed).
+    External channel delivery is fire-and-forget — failures are logged, not raised.
+    """
+    try:
+        recipients = await _resolve_recipients(account_id, target_user_ids, db)
+
+        title = _build_notification_title(event, payload)
+        body = _build_notification_body(event, payload)
+        link = _build_notification_link(event, payload)
+
+        # Create in-app notification for each recipient
+        for user in recipients:
+            notification = Notification(
+                account_id=account_id,
+                user_id=user.id,
+                event=event,
+                title=title,
+                body=body,
+                link=link,
+            )
+            db.add(notification)
+
+        await db.flush()
+
+        # Route to active external channels (fire-and-forget per channel)
+        configs = await _get_active_configs(account_id, event, db)
+        for config in configs:
+            try:
+                await _deliver_to_channel(config, event, payload, db)
+            except Exception:
+                logger.exception(
+                    "External delivery failed for config=%s event=%s", config.id, event
+                )
+    except Exception:
+        logger.exception("Failed to process notification event=%s account=%s", event, account_id)
+
+
+async def retry_failed_notifications(db: AsyncSession) -> int:
+    """Retry failed notification deliveries. Called by APScheduler."""
+    now = datetime.now(timezone.utc)
+    result = await db.execute(
+        select(NotificationLog)
+        .where(NotificationLog.status == "retrying")
+        .where(NotificationLog.next_retry_at <= now)
+    )
+    logs = result.scalars().all()
+
+    if not logs:
+        return 0
+
+    logger.info("Retrying %d failed notification deliveries", len(logs))
+
+    for log in logs:
+        # Load the config for this log entry
+        config_result = await db.execute(
+            select(NotificationConfig).where(NotificationConfig.id == log.notification_config_id)
+        )
+        config = config_result.scalar_one_or_none()
+        if not config or not config.is_active:
+            log.status = "exhausted"
+            log.last_error = "Config disabled or deleted"
+            continue
+
+        try:
+            await _attempt_delivery(config, log.event, log.payload)
+            log.status = "sent"
+            log.delivered_at = datetime.now(timezone.utc)
+            log.last_error = None
+            logger.info("Retry succeeded for log=%s event=%s", log.id, log.event)
+        except Exception as exc:
+            log.retry_count += 1
+            log.last_error = str(exc)[:1000]
+
+            if log.retry_count >= log.max_retries:
+                log.status = "exhausted"
+                logger.warning(
+                    "Notification exhausted after %d retries: log=%s event=%s",
+                    log.retry_count, log.id, log.event,
+                )
+            else:
+                delay = _RETRY_DELAYS[min(log.retry_count, len(_RETRY_DELAYS) - 1)]
+                log.next_retry_at = datetime.now(timezone.utc) + timedelta(seconds=delay)
+                logger.info(
+                    "Notification retry %d/%d scheduled in %ds: log=%s",
+                    log.retry_count, log.max_retries, delay, log.id,
+                )
+
+    await db.commit()
+    return len(logs)
+
+
+async def send_test_notification(
+    config: NotificationConfig,
+) -> tuple[bool, str]:
+    """Send a test message through a channel config. Returns (success, message)."""
+    event = "test"
+    payload: dict[str, Any] = {}
+    try:
+        await _attempt_delivery(config, event, payload)
+        return True, "Test notification sent successfully"
+    except Exception as exc:
+        return False, f"Delivery failed: {exc}"
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+async def _get_active_configs(
+    account_id: uuid.UUID,
+    event: str,
+    db: AsyncSession,
+) -> list[NotificationConfig]:
+    """Get configs where channel is active and event is enabled."""
+    result = await db.execute(
+        select(NotificationConfig)
+        .where(NotificationConfig.account_id == account_id)
+        .where(NotificationConfig.is_active.is_(True))
+    )
+    configs = result.scalars().all()
+    # Filter to configs where this event is enabled
+    return [
+        c for c in configs
+        if c.events_enabled and c.events_enabled.get(event, False)
+    ]
+
+
+async def _resolve_recipients(
+    account_id: uuid.UUID,
+    target_user_ids: Optional[list[uuid.UUID]],
+    db: AsyncSession,
+) -> list[User]:
+    """Resolve notification recipients. Defaults to team admins + account owners + admins."""
+    if target_user_ids:
+        result = await db.execute(
+            select(User)
+            .where(User.id.in_(target_user_ids))
+            .where(User.account_id == account_id)  # enforce tenant boundary
+            .where(User.is_active.is_(True))
+        )
+        return list(result.scalars().all())
+
+    # Default: account owners, admins, and team admins
+    result = await db.execute(
+        select(User)
+        .where(User.account_id == account_id)
+        .where(User.is_active.is_(True))
+    )
+    users = result.scalars().all()
+    return [
+        u for u in users
+        if u.account_role in ("owner", "admin") or u.is_team_admin
+    ]
+
+
+async def _deliver_to_channel(
+    config: NotificationConfig,
+    event: str,
+    payload: dict[str, Any],
+    db: AsyncSession,
+) -> None:
+    """Attempt delivery and create a NotificationLog entry."""
+    log = NotificationLog(
+        notification_config_id=config.id,
+        event=event,
+        payload=payload,
+    )
+
+    try:
+        await _attempt_delivery(config, event, payload)
+        log.status = "sent"
+        log.delivered_at = datetime.now(timezone.utc)
+    except Exception as exc:
+        log.status = "retrying"
+        log.retry_count = 0
+        log.last_error = str(exc)[:1000]
+        log.next_retry_at = datetime.now(timezone.utc) + timedelta(seconds=_RETRY_DELAYS[0])
+        logger.warning(
+            "Notification delivery failed (will retry): config=%s event=%s error=%s",
+            config.id, event, exc,
+        )
+
+    db.add(log)
+
+
+async def _attempt_delivery(
+    config: NotificationConfig,
+    event: str,
+    payload: dict[str, Any],
+) -> None:
+    """Dispatch to the appropriate channel. Raises on failure."""
+    if config.channel == "email":
+        await _send_email(config, event, payload)
+    elif config.channel == "slack_webhook":
+        if not config.webhook_url:
+            raise ValueError("Slack webhook URL not configured")
+        await _send_slack_message(config.webhook_url, event, payload)
+    elif config.channel == "teams_webhook":
+        if not config.webhook_url:
+            raise ValueError("Teams webhook URL not configured")
+        await _send_teams_message(config.webhook_url, event, payload)
+    else:
+        raise ValueError(f"Unknown channel: {config.channel}")
+
+
+async def _send_email(
+    config: NotificationConfig,
+    event: str,
+    payload: dict[str, Any],
+) -> None:
+    """Send notification via email using EmailService."""
+    title = _build_notification_title(event, payload)
+    body = _build_notification_body(event, payload)
+    link = _build_notification_link(event, payload)
+
+    full_link = None
+    if link and settings.FRONTEND_URL:
+        full_link = f"{settings.FRONTEND_URL.rstrip('/')}{link}"
+
+    recipients = config.email_addresses or []
+    if not recipients:
+        raise ValueError("No email addresses configured for email channel")
+
+    failures = []
+    for email in recipients:
+        success = await EmailService.send_notification_email(
+            to_email=email,
+            title=title,
+            body=body,
+            link_url=full_link,
+        )
+        if not success:
+            failures.append(email)
+    if failures:
+        raise RuntimeError(f"Failed to send notification email to: {', '.join(failures)}")
+
+
+async def _send_slack_message(
+    webhook_url: str,
+    event: str,
+    payload: dict[str, Any],
+) -> None:
+    """POST notification to Slack incoming webhook."""
+    title = _build_notification_title(event, payload)
+    body = _build_notification_body(event, payload)
+    link = _build_notification_link(event, payload)
+
+    blocks: list[dict[str, Any]] = [
+        {
+            "type": "header",
+            "text": {"type": "plain_text", "text": f"\U0001f514 {title}", "emoji": True},
+        },
+        {
+            "type": "section",
+            "text": {"type": "mrkdwn", "text": body},
+        },
+    ]
+
+    if link and settings.FRONTEND_URL:
+        full_url = f"{settings.FRONTEND_URL.rstrip('/')}{link}"
+        blocks.append({
+            "type": "actions",
+            "elements": [{
+                "type": "button",
+                "text": {"type": "plain_text", "text": "Open in ResolutionFlow", "emoji": True},
+                "url": full_url,
+            }],
+        })
+
+    slack_payload = {"blocks": blocks}
+
+    async with httpx.AsyncClient(timeout=10) as client:
+        resp = await client.post(webhook_url, json=slack_payload)
+        if resp.status_code != 200 or resp.text.strip() != "ok":
+            raise RuntimeError(
+                f"Slack webhook failed (status={resp.status_code}): {resp.text[:200]}"
+            )
+
+
+async def _send_teams_message(
+    webhook_url: str,
+    event: str,
+    payload: dict[str, Any],
+) -> None:
+    """POST notification to Microsoft Teams incoming webhook (Adaptive Card)."""
+    title = _build_notification_title(event, payload)
+    body = _build_notification_body(event, payload)
+    link = _build_notification_link(event, payload)
+
+    card_body: list[dict[str, Any]] = [
+        {"type": "TextBlock", "text": title, "weight": "Bolder", "size": "Medium"},
+        {"type": "TextBlock", "text": body, "wrap": True},
+    ]
+
+    actions: list[dict[str, Any]] = []
+    if link and settings.FRONTEND_URL:
+        full_url = f"{settings.FRONTEND_URL.rstrip('/')}{link}"
+        actions.append({
+            "type": "Action.OpenUrl",
+            "title": "Open in ResolutionFlow",
+            "url": full_url,
+        })
+
+    teams_payload = {
+        "type": "message",
+        "attachments": [{
+            "contentType": "application/vnd.microsoft.card.adaptive",
+            "content": {
+                "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+                "type": "AdaptiveCard",
+                "version": "1.4",
+                "body": card_body,
+                "actions": actions,
+            },
+        }],
+    }
+
+    async with httpx.AsyncClient(timeout=10) as client:
+        resp = await client.post(webhook_url, json=teams_payload)
+        if resp.status_code not in (200, 202):
+            raise RuntimeError(
+                f"Teams webhook returned {resp.status_code}: {resp.text[:200]}"
+            )
+
+
+# ---------------------------------------------------------------------------
+# Content builders
+# ---------------------------------------------------------------------------
+
+def _build_notification_title(event: str, payload: dict[str, Any]) -> str:
+    """Human-readable title per event type."""
+    titles = {
+        "session.escalated": "Session escalated by {engineer_name}",
+        "session.high_priority": "High-priority session started: {ticket_number}",
+        "proposal.pending": "New flow proposal: {title}",
+        "proposal.approved": "Flow proposal approved: {title}",
+        "knowledge_gap.detected": "Knowledge gap detected: {gap_type}",
+        "test": "Test Notification from ResolutionFlow",
+    }
+    template = titles.get(event, f"Notification: {event}")
+    try:
+        return template.format(**payload)
+    except KeyError:
+        return template
+
+
+def _build_notification_body(event: str, payload: dict[str, Any]) -> str:
+    """Body text per event type."""
+    bodies = {
+        "session.escalated": "Engineer {engineer_name} has escalated a FlowPilot session and needs assistance.",
+        "session.high_priority": "A new high-priority troubleshooting session has been started for ticket {ticket_number}.",
+        "proposal.pending": "A new flow proposal \"{title}\" is awaiting review in the review queue.",
+        "proposal.approved": "The flow proposal \"{title}\" has been approved and is ready for use.",
+        "knowledge_gap.detected": "A {gap_type} knowledge gap has been identified. Review recommended.",
+        "test": "This is a test notification to verify your notification channel is working correctly.",
+    }
+    template = bodies.get(event, f"Event: {event}")
+    try:
+        return template.format(**payload)
+    except KeyError:
+        return template
+
+
+def _build_notification_link(event: str, payload: dict[str, Any]) -> Optional[str]:
+    """In-app link per event type. Returns path (no host)."""
+    links: dict[str, str] = {
+        "session.escalated": "/pilot/{session_id}",
+        "session.high_priority": "/pilot/{session_id}",
+        "proposal.pending": "/review-queue",
+        "proposal.approved": "/review-queue",
+        "knowledge_gap.detected": "/analytics/flowpilot",
+    }
+    template = links.get(event)
+    if template is None:
+        return None
+    try:
+        return template.format(**payload)
+    except KeyError:
+        return template