diff --git a/alembic/versions/2026_09_06_notification_project.py b/alembic/versions/2026_09_06_notification_project.py
new file mode 100644
index 000000000..9ecc81667
--- /dev/null
+++ b/alembic/versions/2026_09_06_notification_project.py
@@ -0,0 +1,36 @@
+"""notifications: scope a notification to its project
+
+Revision ID: c4b8e1f70a92
+Revises: a91d47c30fb2
+Create Date: 2026-09-06
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "c4b8e1f70a92"
+down_revision: str | None = "a91d47c30fb2"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("notifications", sa.Column("project_id", sa.Uuid(), nullable=True))
+    op.create_index("ix_notifications_project_id", "notifications", ["project_id"])
+    op.execute(
+        """
+        UPDATE notifications n
+        SET project_id = s.project_id
+        FROM scans s
+        WHERE n.notification_metadata->>'scan_id' IS NOT NULL
+          AND s.id = (n.notification_metadata->>'scan_id')::uuid
+        """
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_notifications_project_id", table_name="notifications")
+    op.drop_column("notifications", "project_id")
diff --git a/api/app/api/v1/notifications.py b/api/app/api/v1/notifications.py
index 155a976f3..0fbf7b6ce 100644
--- a/api/app/api/v1/notifications.py
+++ b/api/app/api/v1/notifications.py
@@ -1,6 +1,7 @@
 from typing import Annotated
+from uuid import UUID
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 from fastapi_pagination import Page
 from fastapi_pagination.ext.sqlalchemy import paginate
 from sqlalchemy import func, select
@@ -23,17 +24,36 @@ router = APIRouter(
     tags=["notifications"],
 )
 
+ProjectScope = Annotated[
+    UUID | None,
+    Query(description="Only this project's notifications, plus global ones"),
+]
+
+
+def _in_scope(query, project_id: UUID | None):
+    if project_id is None:
+        return query
+    return query.where(
+        (Notification.project_id == project_id) | Notification.project_id.is_(None)
+    )
+
 
 @router.get("/stats", response_model=NotificationStats)
 async def get_notification_stats(
     _current_user: CurrentUser,
     session: Annotated[AsyncSession, Depends(get_session)],
+    project_id: ProjectScope = None,
 ):
-    total_result = await session.execute(select(func.count(Notification.id)))
+    total_result = await session.execute(
+        _in_scope(select(func.count(Notification.id)), project_id)
+    )
     total = total_result.scalar_one()
 
     unread_result = await session.execute(
-        select(func.count(Notification.id)).where(Notification.is_read.is_(False))
+        _in_scope(
+            select(func.count(Notification.id)).where(Notification.is_read.is_(False)),
+            project_id,
+        )
     )
 
     unread = unread_result.scalar_one()
@@ -45,8 +65,11 @@ async def get_notification_stats(
 async def list_notifications(
     _current_user: CurrentUser,
     session: Annotated[AsyncSession, Depends(get_session)],
+    project_id: ProjectScope = None,
 ):
-    query = select(Notification).order_by(Notification.created_at.desc())
+    query = _in_scope(select(Notification), project_id).order_by(
+        Notification.created_at.desc()
+    )
     return await paginate(session, query)
 
 
@@ -54,12 +77,11 @@ async def list_notifications(
 async def list_unread_notifications(
     _current_user: CurrentUser,
     session: Annotated[AsyncSession, Depends(get_session)],
+    project_id: ProjectScope = None,
 ):
-    query = (
-        select(Notification)
-        .where(Notification.is_read.is_(False))
-        .order_by(Notification.created_at.desc())
-    )
+    query = _in_scope(
+        select(Notification).where(Notification.is_read.is_(False)), project_id
+    ).order_by(Notification.created_at.desc())
     return await paginate(session, query)
 
 
@@ -105,8 +127,11 @@ async def mark_notification_as_read(
 async def mark_all_notifications_as_read(
     _current_user: CurrentUser,
     session: Annotated[AsyncSession, Depends(get_session)],
+    project_id: ProjectScope = None,
 ):
-    count = await NotificationManager.mark_all_as_read(session=session)
+    count = await NotificationManager.mark_all_as_read(
+        session=session, project_id=project_id
+    )
 
     return {
         "success": True,
@@ -137,8 +162,9 @@ async def delete_notification(
 async def clear_all_notifications(
     _current_user: CurrentUser,
     session: Annotated[AsyncSession, Depends(get_session)],
+    project_id: ProjectScope = None,
 ):
-    await NotificationManager.clear_all(session=session)
+    await NotificationManager.clear_all(session=session, project_id=project_id)
 
 
 if settings.DEBUG:
diff --git a/api/app/services/scan.py b/api/app/services/scan.py
index bbdc26136..bf0a6abee 100644
--- a/api/app/services/scan.py
+++ b/api/app/services/scan.py
@@ -17,7 +17,6 @@ from app.services.scan_context import ScanContextService
 from app.services.scan_engine import ScanEngineService, stage_effects
 from app.services.target import TargetService
 from shared.config import BaseAppSettings
-from shared.definitions.notifications import scan_cancelled
 from shared.definitions.rescan import ASSET_SEED_STAGE, rescan_label
 from shared.enums.api_key import APIProvider
 from shared.enums.scan import SCAN_LIVE_STATUSES, ScanActivityStatus, ScanStatus
@@ -52,7 +51,6 @@ from shared.models.subdomain import Subdomain
 from shared.models.target import Target
 from shared.services.celery_dispatch import revoke_scan_tasks
 from shared.services.launch_plan import AdHocEngine, plan_label
-from shared.services.notification import NotificationManager
 from shared.services.orchestrator.events import ScanEventPublisher
 from shared.services.scan_factory import build_scan_row
 from shared.services.scan_resolve import (
@@ -1122,19 +1120,6 @@ class ScanService:
         return self._to_read(scan)
 
     async def _announce_cancelled(self, scan: Scan) -> None:
-        target_value = (scan.execution_config or {}).get("target_value", "")
-        payload = scan_cancelled(str(scan.id), target_value, scan.engine_name)
-        try:
-            await NotificationManager.publish(
-                session=self.session,
-                type=payload["type"],
-                severity=payload["severity"],
-                title=payload["title"],
-                message=payload["message"],
-                metadata=payload.get("metadata"),
-            )
-        except Exception:
-            logger.warning("cancel notification dispatch failed", exc_info=True)
         try:
             ScanEventPublisher(
                 BaseAppSettings().redis_url,
diff --git a/frontend/src/lib/api/notifications.ts b/frontend/src/lib/api/notifications.ts
index 32d4abfdd..1f625df35 100644
--- a/frontend/src/lib/api/notifications.ts
+++ b/frontend/src/lib/api/notifications.ts
@@ -2,28 +2,40 @@ import { api } from './client';
 import type { Notification, NotificationStats } from '$lib/types/notification';
 import type { PaginatedResponse } from '$lib/types/pagination';
 
-export const notificationsApi = {
-	list: (page = 1, size = 20): Promise<PaginatedResponse<Notification>> => {
-		return api.get<PaginatedResponse<Notification>>(`/notifications?page=${page}&size=${size}`);
-	},
+const scope = (projectId?: string) => (projectId ? `&project_id=${projectId}` : '');
 
-	listUnread: (page = 1, size = 20): Promise<PaginatedResponse<Notification>> => {
+export const notificationsApi = {
+	list: (page = 1, size = 20, projectId?: string): Promise<PaginatedResponse<Notification>> => {
 		return api.get<PaginatedResponse<Notification>>(
-			`/notifications/unread?page=${page}&size=${size}`
+			`/notifications?page=${page}&size=${size}${scope(projectId)}`
 		);
 	},
 
-	stats: (): Promise<NotificationStats> => {
-		return api.get<NotificationStats>('/notifications/stats');
+	listUnread: (
+		page = 1,
+		size = 20,
+		projectId?: string
+	): Promise<PaginatedResponse<Notification>> => {
+		return api.get<PaginatedResponse<Notification>>(
+			`/notifications/unread?page=${page}&size=${size}${scope(projectId)}`
+		);
+	},
+
+	stats: (projectId?: string): Promise<NotificationStats> => {
+		return api.get<NotificationStats>(
+			`/notifications/stats${projectId ? `?project_id=${projectId}` : ''}`
+		);
 	},
 
 	markAsRead: (id: number): Promise<{ success: boolean; message: string }> => {
 		return api.patch<{ success: boolean; message: string }>(`/notifications/${id}/read`, {});
 	},
 
-	markAllAsRead: (): Promise<{ success: boolean; message: string; count: number }> => {
+	markAllAsRead: (
+		projectId?: string
+	): Promise<{ success: boolean; message: string; count: number }> => {
 		return api.post<{ success: boolean; message: string; count: number }>(
-			'/notifications/read-all',
+			`/notifications/read-all${projectId ? `?project_id=${projectId}` : ''}`,
 			{}
 		);
 	},
@@ -32,7 +44,7 @@ export const notificationsApi = {
 		return api.delete(`/notifications/${id}`);
 	},
 
-	clearAll: (): Promise<void> => {
-		return api.delete('/notifications');
+	clearAll: (projectId?: string): Promise<void> => {
+		return api.delete(`/notifications${projectId ? `?project_id=${projectId}` : ''}`);
 	}
 };
diff --git a/frontend/src/lib/components/layout/notifications-menu.svelte b/frontend/src/lib/components/layout/notifications-menu.svelte
index 86649ed26..192086caa 100644
--- a/frontend/src/lib/components/layout/notifications-menu.svelte
+++ b/frontend/src/lib/components/layout/notifications-menu.svelte
@@ -65,6 +65,14 @@
 		return counts;
 	});
 
+	const presentTypes = $derived(NOTIFICATION_TYPES.filter((t) => typeCounts[t] > 0));
+
+	$effect(() => {
+		if (selectedFilter !== 'all' && !presentTypes.includes(selectedFilter)) {
+			selectedFilter = 'all';
+		}
+	});
+
 	function onPopoverChange(open: boolean) {
 		popoverOpen = open;
 		if (open) tab = notificationStore.unreadCount > 0 ? 'unread' : 'all';
@@ -116,7 +124,7 @@
 		try {
 			await notificationStore.deleteNotification(id);
 		} catch {
-			toast.error("Couldn't delete notification — try again");
+			toast.error('Notification could not be deleted');
 		}
 	};
 
@@ -124,7 +132,7 @@
 		try {
 			await notificationStore.markAllAsRead();
 		} catch {
-			toast.error("Couldn't mark notifications as read — try again");
+			toast.error('Notifications could not be marked read');
 		}
 	};
 
@@ -134,7 +142,7 @@
 			await notificationStore.clearAll();
 			clearAllOpen = false;
 		} catch {
-			toast.error("Couldn't clear notifications — try again");
+			toast.error('Notifications could not be cleared');
 		} finally {
 			clearing = false;
 		}
@@ -156,7 +164,7 @@
 		goto(ROUTES.settings('notifications'));
 	};
 
-	const retryLoad = () => notificationStore.loadNotifications();
+	const retryLoad = () => notificationStore.loadNotifications(notificationStore.projectId);
 </script>
 
 {#snippet skeletonRows(count: number)}
@@ -271,7 +279,7 @@
 						<Empty.Media variant="icon">
 							<TriangleAlert class="text-destructive" />
 						</Empty.Media>
-						<Empty.Title class="text-sm">Couldn't load notifications</Empty.Title>
+						<Empty.Title class="text-sm">Notifications could not be loaded</Empty.Title>
 					</Empty.Header>
 					<Empty.Content>
 						<Button variant="outline" size="sm" onclick={retryLoad}>Retry</Button>
@@ -288,12 +296,12 @@
 							{/if}
 						</Empty.Media>
 						<Empty.Title class="text-sm">
-							{tab === 'unread' ? "You're all caught up" : 'No notifications yet'}
+							{tab === 'unread' ? 'Nothing unread' : 'No notifications yet'}
 						</Empty.Title>
 						<Empty.Description class="text-xs">
 							{tab === 'unread'
-								? 'New alerts and scan results land here.'
-								: 'Scan results, findings and system events will appear here.'}
+								? 'Every notification has been read.'
+								: 'Scan results, findings and system events appear here.'}
 						</Empty.Description>
 					</Empty.Header>
 				</Empty.Root>
@@ -397,16 +405,14 @@
 						{typeCounts.all}
 					</span>
 				</Tabs.Trigger>
-				{#each NOTIFICATION_TYPES as type (type)}
+				{#each presentTypes as type (type)}
 					{@const TypeIcon = getTypeIcon(type)}
 					<Tabs.Trigger value={type} class="flex-none text-xs">
 						<TypeIcon class="size-3" />
 						{NOTIFICATION_TYPE_LABELS[type]}
-						{#if typeCounts[type] > 0}
-							<span class="font-mono text-[10px] text-muted-foreground tabular-nums">
-								{typeCounts[type]}
-							</span>
-						{/if}
+						<span class="font-mono text-[10px] text-muted-foreground tabular-nums">
+							{typeCounts[type]}
+						</span>
 					</Tabs.Trigger>
 				{/each}
 			</Tabs.List>
@@ -431,10 +437,8 @@
 							<Empty.Media variant="icon">
 								<TriangleAlert class="text-destructive" />
 							</Empty.Media>
-							<Empty.Title>Couldn't load notifications</Empty.Title>
-							<Empty.Description
-								>Something went wrong fetching your notifications.</Empty.Description
-							>
+							<Empty.Title>Notifications could not be loaded</Empty.Title>
+							<Empty.Description>The request did not complete. Retry below.</Empty.Description>
 						</Empty.Header>
 						<Empty.Content>
 							<Button
@@ -479,7 +483,7 @@
 <DeleteConfirmationDialog
 	bind:open={clearAllOpen}
 	title="Clear all notifications?"
-	description={`This permanently deletes all ${notificationStore.totalCount} notifications, including any you haven't triaged. This can't be undone.`}
+	description={`All ${notificationStore.totalCount} notifications in this project are deleted, including unread ones.`}
 	confirmLabel="Clear all"
 	isDeleting={clearing}
 	onOpenChange={(o) => (clearAllOpen = o)}
diff --git a/frontend/src/lib/components/notifications/notification-toasts.svelte b/frontend/src/lib/components/notifications/notification-toasts.svelte
index fa4467e37..28d0a4854 100644
--- a/frontend/src/lib/components/notifications/notification-toasts.svelte
+++ b/frontend/src/lib/components/notifications/notification-toasts.svelte
@@ -3,6 +3,10 @@
 	import { goto } from '$app/navigation';
 	import { toast } from 'svelte-sonner';
 	import { onMount } from 'svelte';
+	import type { MessageLevel } from '$lib/types/message-level';
+
+	const INTERRUPTS: MessageLevel[] = ['warning', 'error'];
+	const DURATION: Partial<Record<MessageLevel, number>> = { warning: 6000, error: 10000 };
 
 	function isSameOrigin(url: string): boolean {
 		try {
@@ -14,34 +18,28 @@
 
 	onMount(() => {
 		const unsubscribe = notificationStore.subscribeToToasts((notification) => {
+			if (!INTERRUPTS.includes(notification.severity)) return;
+
 			const metadata = notification.notification_metadata;
+			const show = notification.severity === 'error' ? toast.error : toast.warning;
 
-			const toastFunction =
-				{
-					success: toast.success,
-					error: toast.error,
-					warning: toast.warning,
-					info: toast.info
-				}[notification.severity] || toast;
-
-			toastFunction(notification.title, {
+			show(notification.title, {
 				description: notification.message,
-				action: metadata?.action_label
+				action: metadata?.url
 					? {
-							label: metadata.action_label,
+							label: metadata.action_label ?? 'View',
 							onClick: () => {
 								void notificationStore.markAsRead(notification.id);
-								if (metadata.url) {
-									if (metadata.open_new_tab || !isSameOrigin(metadata.url)) {
-										window.open(metadata.url, '_blank', 'noopener,noreferrer');
-									} else {
-										void goto(metadata.url);
-									}
+								const url = metadata.url as string;
+								if (metadata.open_new_tab || !isSameOrigin(url)) {
+									window.open(url, '_blank', 'noopener,noreferrer');
+								} else {
+									void goto(url);
 								}
 							}
 						}
 					: undefined,
-				duration: 5000
+				duration: DURATION[notification.severity] ?? 5000
 			});
 		});
 
diff --git a/frontend/src/lib/stores/notifications.svelte.ts b/frontend/src/lib/stores/notifications.svelte.ts
index 95da0d833..3c98b75ca 100644
--- a/frontend/src/lib/stores/notifications.svelte.ts
+++ b/frontend/src/lib/stores/notifications.svelte.ts
@@ -10,15 +10,19 @@ interface NotificationState {
 	totalCount: number;
 	isLoading: boolean;
 	hasLoaded: boolean;
+	projectId: string | undefined;
 	error: Error | null;
 }
 
+const MAX_INBOX = 200;
+
 const state = $state<NotificationState>({
 	notifications: [],
 	unreadCount: 0,
 	totalCount: 0,
 	isLoading: false,
 	hasLoaded: false,
+	projectId: undefined,
 	error: null
 });
 
@@ -46,6 +50,10 @@ export const notificationStore = {
 		return state.hasLoaded;
 	},
 
+	get projectId() {
+		return state.projectId;
+	},
+
 	get error() {
 		return state.error;
 	},
@@ -55,15 +63,16 @@ export const notificationStore = {
 		return () => toastCallbacks.delete(callback);
 	},
 
-	async loadNotifications(page = 1, size = 50) {
+	async loadNotifications(projectId?: string, page = 1, size = 50) {
 		state.isLoading = true;
 		state.error = null;
+		state.projectId = projectId;
 
 		try {
-			const response = await notificationsApi.list(page, size);
+			const response = await notificationsApi.list(page, size, projectId);
 			state.notifications = response.items;
 
-			const stats = await notificationsApi.stats();
+			const stats = await notificationsApi.stats(projectId);
 			state.unreadCount = stats.unread;
 			state.totalCount = stats.total;
 
@@ -81,8 +90,12 @@ export const notificationStore = {
 		state.error = null;
 
 		try {
-			const stats = await notificationsApi.stats();
-			const response = await notificationsApi.list(1, stats.total);
+			const stats = await notificationsApi.stats(state.projectId);
+			const response = await notificationsApi.list(
+				1,
+				Math.min(Math.max(stats.total, 1), MAX_INBOX),
+				state.projectId
+			);
 			state.notifications = response.items;
 			state.unreadCount = stats.unread;
 			state.totalCount = stats.total;
@@ -96,7 +109,7 @@ export const notificationStore = {
 
 	async updateStats() {
 		try {
-			const stats = await notificationsApi.stats();
+			const stats = await notificationsApi.stats(state.projectId);
 			state.unreadCount = stats.unread;
 		} catch (error) {
 			console.error('[Notifications] Failed to update stats:', error);
@@ -119,7 +132,7 @@ export const notificationStore = {
 
 	async markAllAsRead() {
 		try {
-			const result = await notificationsApi.markAllAsRead();
+			const result = await notificationsApi.markAllAsRead(state.projectId);
 
 			state.notifications.forEach((n) => (n.is_read = true));
 			state.unreadCount = 0;
@@ -153,7 +166,7 @@ export const notificationStore = {
 
 	async clearAll() {
 		try {
-			await notificationsApi.clearAll();
+			await notificationsApi.clearAll(state.projectId);
 			state.notifications = [];
 			state.unreadCount = 0;
 			state.totalCount = 0;
@@ -188,11 +201,15 @@ export const notificationStore = {
 		state.totalCount = 0;
 		state.isLoading = false;
 		state.hasLoaded = false;
+		state.projectId = undefined;
 		state.error = null;
 	},
 
 	handleNewNotification(notification: Notification, fromSSE: boolean = true) {
-		state.notifications = [notification, ...state.notifications];
+		if (notification.project_id && notification.project_id !== state.projectId) return;
+		if (state.notifications.some((n) => n.id === notification.id)) return;
+
+		state.notifications = [notification, ...state.notifications].slice(0, MAX_INBOX);
 
 		if (!notification.is_read) {
 			state.unreadCount++;
diff --git a/frontend/src/lib/types/notification.ts b/frontend/src/lib/types/notification.ts
index 156c51582..a0573e744 100644
--- a/frontend/src/lib/types/notification.ts
+++ b/frontend/src/lib/types/notification.ts
@@ -34,6 +34,7 @@ export interface NotificationMetadata {
 
 export interface Notification {
 	id: number;
+	project_id?: string | null;
 	type: NotificationType;
 	severity: NotificationSeverity;
 	title: string;
diff --git a/frontend/src/routes/(app)/+layout.svelte b/frontend/src/routes/(app)/+layout.svelte
index 9e30b5009..700304aac 100644
--- a/frontend/src/routes/(app)/+layout.svelte
+++ b/frontend/src/routes/(app)/+layout.svelte
@@ -8,7 +8,7 @@
 	import { notificationStore } from '$lib/stores/notifications.svelte';
 	import { sseStore } from '$lib/stores/sse.svelte';
 	import { liveScans } from '$lib/stores/live-scans.svelte';
-	import { onMount } from 'svelte';
+	import { onMount, untrack } from 'svelte';
 	import AppSidebar from '$lib/components/layout/app-sidebar.svelte';
 	import TopBar from '$lib/components/layout/top-bar.svelte';
 	import NotificationToasts from '$lib/components/notifications/notification-toasts.svelte';
@@ -71,11 +71,13 @@
 	});
 
 	$effect(() => {
-		if (auth.isAuthenticated && !auth.isLoading) {
-			if (!notificationStore.hasLoaded && !notificationStore.isLoading) {
-				notificationStore.loadNotifications();
-			}
-		}
+		const projectId = projectsStore.activeProject?.id;
+		if (!auth.isAuthenticated || auth.isLoading || !projectId) return;
+		untrack(() => {
+			if (notificationStore.isLoading) return;
+			if (notificationStore.hasLoaded && notificationStore.projectId === projectId) return;
+			notificationStore.loadNotifications(projectId);
+		});
 	});
 
 	$effect(() => {
diff --git a/shared/definitions/notifications.py b/shared/definitions/notifications.py
index d9fa5c971..b249e8cf1 100644
--- a/shared/definitions/notifications.py
+++ b/shared/definitions/notifications.py
@@ -1,3 +1,7 @@
+"""Notification templates: one interrupt per event, never a copy of the activity log."""
+
+from dataclasses import dataclass, field
+
 from shared.definitions.vulnerabilities import (
     ALERT_SEVERITIES,
     SEVERITY_LABELS,
@@ -7,28 +11,21 @@ from shared.definitions.vulnerabilities import (
 from shared.enums.notification import NotificationSeverity, NotificationType
 
 
-def whois_enrichment_complete(
-    success: int,
-    failed: int,
-    total: int,
-) -> dict:
-    if failed == 0:
-        return {
-            "type": NotificationType.TARGET,
-            "severity": NotificationSeverity.SUCCESS,
-            "title": "WHOIS Enrichment Complete",
-            "message": (
-                f"WHOIS lookup completed for {success} "
-                f"{'target' if success == 1 else 'targets'}."
-            ),
-        }
+def _count(n: int, singular: str, plural: str) -> str:
+    return f"{n:,} {singular if n == 1 else plural}"
+
+
+def whois_enrichment_incomplete(success: int, failed: int, total: int) -> dict | None:
+    if not failed:
+        return None
     return {
         "type": NotificationType.TARGET,
         "severity": NotificationSeverity.WARNING,
-        "title": "WHOIS Enrichment Complete",
+        "title": "WHOIS lookup failed",
         "message": (
-            f"WHOIS lookup completed: {success} succeeded, "
-            f"{failed} failed out of {total} targets."
+            f"WHOIS could not be resolved for {failed} of {total} "
+            f"{'target' if total == 1 else 'targets'}"
+            f"{f'; {success} succeeded' if success else ''}."
         ),
     }
 
@@ -37,38 +34,26 @@ def whois_enrichment_failed(error: str) -> dict:
     return {
         "type": NotificationType.TARGET,
         "severity": NotificationSeverity.ERROR,
-        "title": "WHOIS Enrichment Failed",
-        "message": f"WHOIS enrichment task failed: {error}",
+        "title": "WHOIS enrichment failed",
+        "message": f"WHOIS enrichment failed. {error}",
     }
 
 
-def ripestat_enrichment_complete(
+def ripestat_enrichment_incomplete(
     success: int, failed: int, skipped: int, total: int
-) -> dict:
-    if failed == 0 and skipped == 0:
-        severity = NotificationSeverity.SUCCESS
-        title = "BGP Enrichment Complete"
-        message = f"Successfully enriched {success} targets with BGP data."
-    elif success > 0:
-        severity = NotificationSeverity.WARNING
-        title = "BGP Enrichment Partially Complete"
-        message = (
-            f"BGP enrichment finished: {success} succeeded, "
-            f"{failed} failed, {skipped} skipped out of {total} targets."
-        )
-    else:
-        severity = NotificationSeverity.ERROR
-        title = "BGP Enrichment Failed"
-        message = (
-            f"BGP enrichment could not complete: "
-            f"{failed} failed, {skipped} skipped out of {total} targets."
-        )
-
+) -> dict | None:
+    if not failed:
+        return None
     return {
         "type": NotificationType.TARGET,
-        "severity": severity,
-        "title": title,
-        "message": message,
+        "severity": NotificationSeverity.WARNING,
+        "title": "BGP enrichment failed",
+        "message": (
+            f"BGP data could not be resolved for {failed} of {total} "
+            f"{'target' if total == 1 else 'targets'}"
+            f"{f'; {success} succeeded' if success else ''}"
+            f"{f', {skipped} had nothing to look up' if skipped else ''}."
+        ),
     }
 
 
@@ -76,195 +61,220 @@ def ripestat_enrichment_failed(error: str) -> dict:
     return {
         "type": NotificationType.TARGET,
         "severity": NotificationSeverity.ERROR,
-        "title": "BGP Enrichment Failed",
-        "message": f"BGP enrichment task failed: {error}",
+        "title": "BGP enrichment failed",
+        "message": f"BGP enrichment failed. {error}",
     }
 
 
-_SECONDS_PER_MINUTE = 60
-_MINUTES_PER_HOUR = 60
-
-
-def _scan_meta(scan_id: str) -> dict:
-    return {"scan_id": str(scan_id), "url": f"/scans/{scan_id}"}
-
-
-def _scan_duration(seconds: float | None) -> str:
-    if not seconds or seconds < 0:
-        return ""
-    seconds = int(seconds)
-    if seconds < _SECONDS_PER_MINUTE:
-        return f" in {seconds}s"
-    minutes, rem = divmod(seconds, _SECONDS_PER_MINUTE)
-    if minutes < _MINUTES_PER_HOUR:
-        return f" in {minutes}m {rem}s" if rem else f" in {minutes}m"
-    hours, minutes = divmod(minutes, _MINUTES_PER_HOUR)
-    return f" in {hours}h {minutes}m"
-
-
-def scan_started(scan_id: str, target: str, engine: str) -> dict:
-    return {
-        "type": NotificationType.SCAN,
-        "severity": NotificationSeverity.INFO,
-        "title": "Scan Started",
-        "message": f"Scan of {target} started with engine {engine}.",
-        "metadata": _scan_meta(scan_id),
-    }
+def _scan_meta(scan_id: str, tab: str | None = None) -> dict:
+    url = f"/scans/{scan_id}" + (f"?tab={tab}" if tab else "")
+    return {"scan_id": str(scan_id), "url": url}
 
 
 _SCAN_COUNT_LABELS = {
-    "subdomains_found": ("subdomain", "subdomains"),
-    "ips_found": ("IP", "IPs"),
-    "open_ports_found": ("open port", "open ports"),
+    "subdomains_found": ("host", "hosts"),
+    "ips_found": ("address", "addresses"),
+    "open_ports_found": ("service", "services"),
     "http_assets_found": ("HTTP service", "HTTP services"),
-    "vulnerabilities_found": ("vulnerability", "vulnerabilities"),
+    "vulnerabilities_found": ("finding", "findings"),
     "endpoints_found": ("endpoint", "endpoints"),
 }
 
 
 def scan_count_summary(counts: dict) -> str:
     parts = [
-        f"{n} {singular if n == 1 else plural}"
+        _count(n, singular, plural)
         for col, (singular, plural) in _SCAN_COUNT_LABELS.items()
         if (n := counts.get(col, 0))
     ]
-    return ", ".join(parts) if parts else "no assets"
-
-
-def stage_count_summary(counts: dict) -> str:
-    parts = [
-        f"{n} {key.replace('_', ' ')}"
-        for key, n in counts.items()
-        if isinstance(n, int) and n and key != "excluded"
-    ]
     return ", ".join(parts) if parts else "no results"
 
 
-def scan_completed(
-    scan_id: str,
-    target: str,
-    engine: str,
-    counts: dict,
-    duration_seconds: float | None,
-) -> dict:
-    return {
-        "type": NotificationType.SCAN,
-        "severity": NotificationSeverity.SUCCESS,
-        "title": "Scan Completed",
-        "message": (
-            f"Scan of {target} ({engine}) completed"
-            f"{_scan_duration(duration_seconds)} — {scan_count_summary(counts)}."
-        ),
-        "metadata": _scan_meta(scan_id),
-    }
+_STAGE_COUNT_LABELS: dict[str, tuple[str, str]] = {
+    "active": ("resolving host", "resolving hosts"),
+    "addresses": ("address", "addresses"),
+    "alive": ("responsive host", "responsive hosts"),
+    "answered": ("answered", "answered"),
+    "bgp": ("BGP record", "BGP records"),
+    "cdn": ("CDN-fronted address", "CDN-fronted addresses"),
+    "checked": ("service checked", "services checked"),
+    "checks": ("check run", "checks run"),
+    "cloud": ("cloud-hosted address", "cloud-hosted addresses"),
+    "dns_records": ("DNS record", "DNS records"),
+    "edge_only": ("CDN edge address", "CDN edge addresses"),
+    "endpoints": ("endpoint", "endpoints"),
+    "endpoints_new": ("new endpoint", "new endpoints"),
+    "endpoints_probed": ("endpoint requested", "endpoints requested"),
+    "enriched": ("address enriched", "addresses enriched"),
+    "fingerprinted": ("service identified", "services identified"),
+    "http_assets": ("web service", "web services"),
+    "ips": ("address", "addresses"),
+    "known_ports": ("known service", "known services"),
+    "new": ("new", "new"),
+    "open_ports": ("open service", "open services"),
+    "probed": ("host probed", "hosts probed"),
+    "ptr": ("PTR record", "PTR records"),
+    "scanned": ("address scanned", "addresses scanned"),
+    "screenshots": ("screenshot", "screenshots"),
+    "skipped": ("skipped", "skipped"),
+    "subdomains": ("host", "hosts"),
+    "targets": ("target", "targets"),
+    "vulnerabilities": ("finding", "findings"),
+    "waf": ("firewall identified", "firewalls identified"),
+    "web_services": ("web service", "web services"),
+    "whois": ("WHOIS record", "WHOIS records"),
+}
 
 
-def scan_new_subdomains(scan_id: str, target: str, count: int) -> dict:
-    return {
-        "type": NotificationType.SCAN,
-        "severity": NotificationSeverity.INFO,
-        "title": "New Subdomains Discovered",
-        "message": (
-            f"{count} new {'subdomain' if count == 1 else 'subdomains'} "
-            f"discovered on {target}."
-        ),
-        "metadata": _scan_meta(scan_id),
-    }
+def stage_count_summary(counts: dict) -> str:
+    """Label every figure a stage reports. An unlabelled key is omitted, never printed raw."""
+    parts = [
+        _count(n, *_STAGE_COUNT_LABELS[key])
+        for key, n in counts.items()
+        if isinstance(n, int) and n and key in _STAGE_COUNT_LABELS
+    ]
+    return ", ".join(parts) if parts else "no results"
 
 
-def scan_new_services(scan_id: str, target: str, count: int, sensitive: int) -> dict:
-    detail = (
-        f" {sensitive} of them administrative or datastore "
-        f"{'port' if sensitive == 1 else 'ports'}."
-        if sensitive
-        else ""
+@dataclass(frozen=True)
+class ScanDeltas:
+    baseline: bool = False
+    new_hosts: int = 0
+    new_services: int = 0
+    sensitive_services: int = 0
+    new_vulnerabilities: int = 0
+    vulnerability_counts: dict[str, int] = field(default_factory=dict)
+    kev: int = 0
+    dropped_hosts: int = 0
+
+    @property
+    def critical(self) -> int:
+        return self.vulnerability_counts.get(Severity.CRITICAL.value, 0)
+
+    @property
+    def severe(self) -> int:
+        return sum(self.vulnerability_counts.get(s, 0) for s in ALERT_SEVERITIES)
+
+    def worth_reporting(self, counts: dict) -> bool:
+        if not self.baseline:
+            return any(counts.get(col) for col in _SCAN_COUNT_LABELS)
+        return bool(
+            self.new_hosts
+            or self.new_services
+            or self.new_vulnerabilities
+            or self.dropped_hosts
+        )
+
+
+def _severity_phrase(counts: dict) -> str:
+    return ", ".join(
+        f"{counts[name]} {SEVERITY_LABELS[name].lower()}"
+        for name in SEVERITY_ORDER
+        if counts.get(name)
     )
+
+
+def _digest_title(target: str, deltas: ScanDeltas) -> str:
+    if deltas.critical:
+        head = _count(deltas.critical, "critical finding", "critical findings")
+    elif deltas.kev:
+        head = _count(
+            deltas.kev, "exploited vulnerability", "exploited vulnerabilities"
+        )
+    elif deltas.severe or deltas.sensitive_services:
+        head = "New exposure"
+    elif not deltas.baseline:
+        return f"First scan of {target}"
+    elif deltas.new_vulnerabilities:
+        head = _count(deltas.new_vulnerabilities, "new finding", "new findings")
+    elif deltas.new_hosts or deltas.new_services:
+        head = "New assets"
+    else:
+        head = "Partial coverage"
+    return f"{head} on {target}"
+
+
+def _digest_body(counts: dict, deltas: ScanDeltas) -> str:
+    if not deltas.baseline:
+        body = f"No earlier run to compare against. This run found {scan_count_summary(counts)}."
+    else:
+        detail = _severity_phrase(deltas.vulnerability_counts)
+        parts = [
+            text
+            for text, n in (
+                (_count(deltas.new_hosts, "new host", "new hosts"), deltas.new_hosts),
+                (
+                    _count(deltas.new_services, "new service", "new services"),
+                    deltas.new_services,
+                ),
+                (
+                    _count(deltas.new_vulnerabilities, "new finding", "new findings")
+                    + (f" ({detail})" if detail else ""),
+                    deltas.new_vulnerabilities,
+                ),
+            )
+            if n
+        ]
+        if parts:
+            body = ", ".join(parts) + "."
+        else:
+            body = "Nothing new since the previous run."
+
+    if deltas.sensitive_services:
+        n = deltas.sensitive_services
+        body += (
+            f" {n} {'is' if n == 1 else 'are'} on an administrative or datastore port."
+        )
+    if deltas.kev:
+        body += (
+            f" {deltas.kev} {'is' if deltas.kev == 1 else 'are'} "
+            f"known to be exploited in the wild."
+        )
+    if deltas.dropped_hosts:
+        body += (
+            f" Testing stopped on {_count(deltas.dropped_hosts, 'host', 'hosts')} "
+            f"after repeated errors, so coverage there is partial."
+        )
+    return body
+
+
+def scan_digest(
+    scan_id: str, target: str, counts: dict, deltas: ScanDeltas
+) -> dict | None:
+    """One row per run; None when there is nothing to say."""
+    if not deltas.worth_reporting(counts):
+        return None
+
+    if deltas.critical or deltas.kev:
+        severity = NotificationSeverity.ERROR
+    elif deltas.severe or deltas.sensitive_services or deltas.dropped_hosts:
+        severity = NotificationSeverity.WARNING
+    else:
+        severity = NotificationSeverity.SUCCESS
+
+    if deltas.new_vulnerabilities or deltas.dropped_hosts:
+        tab = "vulnerabilities"
+    elif deltas.sensitive_services or deltas.new_services:
+        tab = "services"
+    else:
+        tab = None
+
     return {
-        "type": NotificationType.SCAN,
-        "severity": NotificationSeverity.WARNING
-        if sensitive
-        else NotificationSeverity.INFO,
-        "title": "New Services Exposed",
-        "message": (
-            f"{count} {'service' if count == 1 else 'services'} on {target} "
-            f"{'was' if count == 1 else 'were'} not open at the previous scan.{detail}"
-        ),
-        "metadata": _scan_meta(scan_id),
+        "type": NotificationType.VULNERABILITY
+        if deltas.new_vulnerabilities
+        else NotificationType.SCAN,
+        "severity": severity,
+        "title": _digest_title(target, deltas),
+        "message": _digest_body(counts, deltas),
+        "metadata": _scan_meta(scan_id, tab),
     }
 
 
 def scan_failed(scan_id: str, target: str, engine: str, error: str) -> dict:
     return {
         "type": NotificationType.SCAN,
-        "severity": NotificationSeverity.ERROR,
-        "title": "Scan Failed",
-        "message": f"Scan of {target} ({engine}) failed: {error[:300]}",
+        "severity": NotificationSeverity.WARNING,
+        "title": f"Scan failed on {target}",
+        "message": f"The {engine} run did not finish: {error[:300]}",
         "metadata": _scan_meta(scan_id),
     }
-
-
-def scan_cancelled(scan_id: str, target: str, engine: str) -> dict:
-    return {
-        "type": NotificationType.SCAN,
-        "severity": NotificationSeverity.WARNING,
-        "title": "Scan Cancelled",
-        "message": f"Scan of {target} ({engine}) was cancelled.",
-        "metadata": _scan_meta(scan_id),
-    }
-
-
-def _severity_phrase(counts: dict) -> str:
-    parts = [
-        f"{counts[name]} {SEVERITY_LABELS[name].lower()}"
-        for name in SEVERITY_ORDER
-        if counts.get(name)
-    ]
-    return ", ".join(parts)
-
-
-def scan_new_vulnerabilities(
-    scan_id: str, target: str, counts: dict, kev: int, total: int
-) -> dict:
-    severe = sum(counts.get(name, 0) for name in ALERT_SEVERITIES)
-    if counts.get(Severity.CRITICAL.value) or kev:
-        severity = NotificationSeverity.ERROR
-    elif severe:
-        severity = NotificationSeverity.WARNING
-    else:
-        severity = NotificationSeverity.INFO
-    detail = _severity_phrase(counts)
-    kev_note = (
-        f" {kev} {'is' if kev == 1 else 'are'} known to be exploited in the wild."
-        if kev
-        else ""
-    )
-    return {
-        "type": NotificationType.VULNERABILITY,
-        "severity": severity,
-        "title": "New Vulnerabilities Found",
-        "message": (
-            f"{total} new {'finding' if total == 1 else 'findings'} on {target}"
-            f"{f' — {detail}' if detail else ''}.{kev_note}"
-        ),
-        "metadata": {
-            **_scan_meta(scan_id),
-            "url": f"/scans/{scan_id}?tab=vulnerabilities",
-        },
-    }
-
-
-def scan_vulnerability_coverage(scan_id: str, target: str, dropped: int) -> dict:
-    return {
-        "type": NotificationType.VULNERABILITY,
-        "severity": NotificationSeverity.WARNING,
-        "title": "Vulnerability Scan Incomplete",
-        "message": (
-            f"The scan of {target} stopped testing {dropped} "
-            f"{'host' if dropped == 1 else 'hosts'} after repeated errors. "
-            f"Coverage for {'that host' if dropped == 1 else 'those hosts'} is partial."
-        ),
-        "metadata": {
-            **_scan_meta(scan_id),
-            "url": f"/scans/{scan_id}?tab=vulnerabilities",
-        },
-    }
diff --git a/shared/models/notification.py b/shared/models/notification.py
index 41f4c112b..87c4bf86a 100644
--- a/shared/models/notification.py
+++ b/shared/models/notification.py
@@ -1,3 +1,4 @@
+import uuid
 from datetime import datetime, timedelta
 
 from pydantic import BaseModel, Field, field_validator
@@ -48,6 +49,7 @@ class Notification(NotificationBase, table=True):
     __tablename__ = "notifications"
 
     id: int = SQLField(default=None, primary_key=True, index=True)
+    project_id: uuid.UUID | None = SQLField(default=None, index=True)
     notification_metadata: dict = SQLField(
         default_factory=dict, sa_column=Column(JSONB)
     )
@@ -65,6 +67,7 @@ class NotificationCreate(NotificationBase):
 
 class NotificationRead(NotificationBase):
     id: int
+    project_id: uuid.UUID | None = None
     notification_metadata: dict
     is_read: bool
     created_at: datetime
diff --git a/shared/services/notification.py b/shared/services/notification.py
index 046029332..922678cd3 100644
--- a/shared/services/notification.py
+++ b/shared/services/notification.py
@@ -1,3 +1,5 @@
+import uuid
+
 from sqlalchemy import delete, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -20,6 +22,7 @@ class NotificationManager:
         title: str,
         message: str,
         metadata: NotificationMetadata | dict | None = None,
+        project_id: uuid.UUID | str | None = None,
         commit: bool = True,
     ) -> Notification:
         if isinstance(metadata, NotificationMetadata):
@@ -36,6 +39,7 @@ class NotificationManager:
             title=title[:200],
             message=message,
             notification_metadata=metadata_dict,
+            project_id=uuid.UUID(str(project_id)) if project_id else None,
         )
 
         session.add(notification)
@@ -51,6 +55,9 @@ class NotificationManager:
             event_type=SSEEventType.NOTIFICATION,
             data={
                 "id": notification.id,
+                "project_id": str(notification.project_id)
+                if notification.project_id
+                else None,
                 "type": notification.type.value,
                 "severity": notification.severity.value,
                 "title": notification.title,
@@ -92,13 +99,16 @@ class NotificationManager:
     @staticmethod
     async def mark_all_as_read(
         session: AsyncSession,
+        project_id: uuid.UUID | None = None,
         commit: bool = True,
     ) -> int:
-        result = await session.execute(
-            update(Notification)
-            .where(Notification.is_read.is_(False))
-            .values(is_read=True)
-        )
+        stmt = update(Notification).where(Notification.is_read.is_(False))
+        if project_id is not None:
+            stmt = stmt.where(
+                (Notification.project_id == project_id)
+                | Notification.project_id.is_(None)
+            )
+        result = await session.execute(stmt.values(is_read=True))
 
         if commit:
             await session.commit()
@@ -123,9 +133,16 @@ class NotificationManager:
     @staticmethod
     async def clear_all(
         session: AsyncSession,
+        project_id: uuid.UUID | None = None,
         commit: bool = True,
     ) -> int:
-        result = await session.execute(delete(Notification))
+        stmt = delete(Notification)
+        if project_id is not None:
+            stmt = stmt.where(
+                (Notification.project_id == project_id)
+                | Notification.project_id.is_(None)
+            )
+        result = await session.execute(stmt)
 
         if commit:
             await session.commit()
diff --git a/shared/services/notification_sync.py b/shared/services/notification_sync.py
index e6d0231f9..5e09a1983 100644
--- a/shared/services/notification_sync.py
+++ b/shared/services/notification_sync.py
@@ -1,4 +1,5 @@
 import logging
+import uuid
 from datetime import timedelta
 
 from sqlalchemy.orm import Session
@@ -14,6 +15,11 @@ logger = logging.getLogger(__name__)
 NOTIFICATION_EXPIRY_DAYS = 7
 
 
+def single_project(rows) -> uuid.UUID | None:
+    ids = {r.project_id for r in rows if getattr(r, "project_id", None)}
+    return next(iter(ids)) if len(ids) == 1 else None
+
+
 class SyncNotificationPublisher:
     def __init__(self, redis_url: str) -> None:
         self._event_publisher = SyncEventPublisher(redis_url)
@@ -26,6 +32,7 @@ class SyncNotificationPublisher:
         title: str,
         message: str,
         metadata: NotificationMetadata | dict | None = None,
+        project_id: uuid.UUID | str | None = None,
     ) -> Notification:
         if isinstance(metadata, NotificationMetadata):
             metadata_dict = metadata.model_dump(exclude_none=True)
@@ -43,10 +50,9 @@ class SyncNotificationPublisher:
             title=title[:200],
             message=message,
             notification_metadata=metadata_dict,
+            project_id=uuid.UUID(str(project_id)) if project_id else None,
             created_at=now,
-            expires_at=(now + timedelta(days=NOTIFICATION_EXPIRY_DAYS)).replace(
-                tzinfo=None
-            ),
+            expires_at=now + timedelta(days=NOTIFICATION_EXPIRY_DAYS),
         )
 
         session.add(notification)
@@ -58,6 +64,9 @@ class SyncNotificationPublisher:
             event_type=SSEEventType.NOTIFICATION,
             data={
                 "id": notification.id,
+                "project_id": str(notification.project_id)
+                if notification.project_id
+                else None,
                 "type": notification.type.value,
                 "severity": notification.severity.value,
                 "title": notification.title,
diff --git a/worker/app/celery.py b/worker/app/celery.py
index bfe8f77ae..8367a6144 100644
--- a/worker/app/celery.py
+++ b/worker/app/celery.py
@@ -118,6 +118,7 @@ celery_app.autodiscover_tasks(
         "app.tasks.vuln_templates",
         "app.tasks.endpoints",
         "app.tasks.reports",
+        "app.tasks.notifications",
     ]
 )
 
@@ -129,6 +130,7 @@ SCHEDULE_TICK_SECONDS = 60.0
 IP_RANGE_REFRESH_SECONDS = 7 * 24 * 60 * 60.0
 TEMPLATE_SYNC_SECONDS = 24 * 60 * 60.0
 REPORT_CLEANUP_SECONDS = 24 * 60 * 60.0
+NOTIFICATION_CLEANUP_SECONDS = 6 * 60 * 60.0
 
 celery_app.conf.beat_schedule = {
     "scan-schedule-tick": {
@@ -147,6 +149,10 @@ celery_app.conf.beat_schedule = {
         "task": "app.tasks.reports.cleanup",
         "schedule": REPORT_CLEANUP_SECONDS,
     },
+    "notification-cleanup": {
+        "task": "app.tasks.notifications.cleanup",
+        "schedule": NOTIFICATION_CLEANUP_SECONDS,
+    },
 }
 
 
diff --git a/worker/app/orchestrator/finalize.py b/worker/app/orchestrator/finalize.py
index e565d580f..7cb2eabb9 100644
--- a/worker/app/orchestrator/finalize.py
+++ b/worker/app/orchestrator/finalize.py
@@ -1,18 +1,14 @@
 """Aggregate stage outcomes into the final scan status + terminal notification/event."""
 
-from sqlalchemy import Integer, String, bindparam, func, select, text
+from sqlalchemy import Integer, String, bindparam, select, text
 from sqlalchemy.dialects.postgresql import ARRAY
 from sqlalchemy.orm import Session
 
 from shared.definitions.notifications import (
-    scan_cancelled,
-    scan_completed,
+    ScanDeltas,
     scan_count_summary,
+    scan_digest,
     scan_failed,
-    scan_new_services,
-    scan_new_subdomains,
-    scan_new_vulnerabilities,
-    scan_vulnerability_coverage,
 )
 from shared.definitions.ports import SENSITIVE_PORTS
 from shared.definitions.vulnerabilities import SUPPRESSED_STATES, CoverageStatus
@@ -21,7 +17,6 @@ from shared.enums.scan import SCAN_TERMINAL_STATUSES, ScanScope, ScanStatus
 from shared.logging import get_logger
 from shared.models.scan import Scan
 from shared.models.scan_activity import ScanActivity
-from shared.models.subdomain import Subdomain
 from shared.models.vulnerability import VulnerabilityCoverage
 from shared.services.activity_log import ActivityLogService
 from shared.services.notification_sync import SyncNotificationPublisher
@@ -38,8 +33,13 @@ logger = get_logger(__name__)
 
 
 def _notify(
-    notifier: SyncNotificationPublisher, session: Session, payload: dict
+    notifier: SyncNotificationPublisher,
+    session: Session,
+    scan: Scan,
+    payload: dict | None,
 ) -> None:
+    if payload is None:
+        return
     try:
         notifier.publish(
             session=session,
@@ -48,6 +48,7 @@ def _notify(
             title=payload["title"],
             message=payload["message"],
             metadata=payload.get("metadata"),
+            project_id=scan.project_id,
         )
     except Exception:
         logger.warning("scan notification dispatch failed", exc_info=True)
@@ -79,60 +80,70 @@ def _log_cancelled(activity_log: ActivityLogService, scan: Scan) -> None:
     )
 
 
-def _count_new_subdomains(session: Session, scan: Scan) -> int:
-    """Subdomain names this scan was the first to discover for its target."""
-    rn = (
-        func.row_number()
-        .over(
-            partition_by=[Subdomain.target_id, Subdomain.name],
-            order_by=[Subdomain.discovered_at.asc(), Subdomain.scan_id.asc()],
-        )
-        .label("rn")
-    )
-    firsts = (
-        select(Subdomain.scan_id.label("scan_id"), rn)
-        .where(Subdomain.target_id == scan.target_id)
-        .subquery()
-    )
-    return (
-        session.execute(
-            select(func.count()).where(firsts.c.rn == 1, firsts.c.scan_id == scan.id)
-        ).scalar_one()
-        or 0
-    )
+# an earlier census run of this target; a focused rescan is never a baseline
+_HOST_BASELINE_SQL = """
+SELECT EXISTS (
+    SELECT 1 FROM subdomains b
+    JOIN scans bs ON bs.id = b.scan_id AND bs.scope = 'full'
+                 AND bs.id <> :sid AND bs.started_at < :started
+    WHERE b.target_id = :tid
+)
+"""
 
+_SERVICE_BASELINE_SQL = """
+SELECT EXISTS (
+    SELECT 1 FROM ports b
+    JOIN scans bs ON bs.id = b.scan_id AND bs.scope = 'full'
+                 AND bs.id <> :sid AND bs.started_at < :started
+    WHERE b.target_id = :tid
+)
+"""
+
+_VULN_BASELINE_SQL = """
+SELECT EXISTS (
+    SELECT 1 FROM vulnerabilities b
+    JOIN scans bs ON bs.id = b.scan_id AND bs.scope = 'full'
+                 AND bs.id <> :sid AND bs.started_at < :started
+    WHERE b.target_id = :tid
+)
+"""
+
+_NEW_SUBDOMAINS_SQL = """
+WITH seen AS (
+    SELECT DISTINCT b.name FROM subdomains b
+    JOIN scans bs ON bs.id = b.scan_id AND bs.scope = 'full'
+                 AND bs.id <> :sid AND bs.started_at < :started
+    WHERE b.target_id = :tid
+)
+SELECT count(*) AS total
+FROM subdomains p
+WHERE p.scan_id = :sid
+  AND NOT EXISTS (SELECT 1 FROM seen s WHERE s.name = p.name)
+"""
 
 _NEW_SERVICES_SQL = """
+WITH seen AS (
+    SELECT DISTINCT b.ip, b.number FROM ports b
+    JOIN scans bs ON bs.id = b.scan_id AND bs.scope = 'full'
+                 AND bs.id <> :sid AND bs.started_at < :started
+    WHERE b.target_id = :tid
+)
 SELECT count(*) AS total,
        count(*) FILTER (WHERE p.number = ANY(:sensitive_ports)) AS sensitive
 FROM ports p
 WHERE p.scan_id = :sid
-  AND EXISTS (
-      SELECT 1 FROM ports b
-      WHERE b.target_id = p.target_id AND b.scan_id <> :sid
-        AND b.discovered_at < p.discovered_at
-  )
   AND NOT EXISTS (
-      SELECT 1 FROM ports e
-      WHERE e.target_id = p.target_id AND e.ip = p.ip AND e.number = p.number
-        AND e.scan_id <> :sid AND e.discovered_at < p.discovered_at
+      SELECT 1 FROM seen s WHERE s.ip = p.ip AND s.number = p.number
   )
 """
 
-
-def _count_new_services(session: Session, scan: Scan) -> tuple[int, int]:
-    """Ports this scan is the first to see for its target, and how many are sensitive."""
-    row = session.execute(
-        text(_NEW_SERVICES_SQL).bindparams(
-            bindparam("sid", scan.id),
-            bindparam("sensitive_ports", SENSITIVE_PORTS, type_=ARRAY(Integer)),
-        )
-    ).one()
-    return int(row.total or 0), int(row.sensitive or 0)
-
-
-# findings this scan is the first to report for its target, by severity
 _NEW_VULNS_SQL = """
+WITH seen AS (
+    SELECT DISTINCT b.fingerprint FROM vulnerabilities b
+    JOIN scans bs ON bs.id = b.scan_id AND bs.scope = 'full'
+                 AND bs.id <> :sid AND bs.started_at < :started
+    WHERE b.target_id = :tid
+)
 SELECT v.severity AS severity,
        count(*) AS total,
        count(*) FILTER (WHERE v.is_kev) AS kev
@@ -143,25 +154,46 @@ WHERE v.scan_id = :sid
       WHERE t.target_id = v.target_id AND t.fingerprint = v.fingerprint
         AND t.state = ANY(:suppressed)
   )
-  AND EXISTS (
-      SELECT 1 FROM vulnerabilities b
-      WHERE b.target_id = v.target_id AND b.scan_id <> :sid
-        AND b.discovered_at < v.discovered_at
-  )
-  AND NOT EXISTS (
-      SELECT 1 FROM vulnerabilities e
-      WHERE e.target_id = v.target_id AND e.fingerprint = v.fingerprint
-        AND e.scan_id <> :sid AND e.discovered_at < v.discovered_at
-  )
+  AND NOT EXISTS (SELECT 1 FROM seen s WHERE s.fingerprint = v.fingerprint)
 GROUP BY v.severity
 """
 
 
+def _has_baseline(session: Session, scan: Scan, sql: str) -> bool:
+    return bool(session.execute(text(sql).bindparams(*_scope(scan))).scalar_one())
+
+
+def _scope(scan: Scan) -> list:
+    return [
+        bindparam("sid", scan.id),
+        bindparam("tid", scan.target_id),
+        bindparam("started", scan.started_at or utc_now()),
+    ]
+
+
+def _count_new_subdomains(session: Session, scan: Scan) -> int:
+    return int(
+        session.execute(
+            text(_NEW_SUBDOMAINS_SQL).bindparams(*_scope(scan))
+        ).scalar_one()
+        or 0
+    )
+
+
+def _count_new_services(session: Session, scan: Scan) -> tuple[int, int]:
+    row = session.execute(
+        text(_NEW_SERVICES_SQL).bindparams(
+            *_scope(scan),
+            bindparam("sensitive_ports", SENSITIVE_PORTS, type_=ARRAY(Integer)),
+        )
+    ).one()
+    return int(row.total or 0), int(row.sensitive or 0)
+
+
 def _count_new_vulnerabilities(session: Session, scan: Scan) -> tuple[dict, int, int]:
-    """New findings for this target, by severity, excluding anything a reviewer set aside."""
     rows = session.execute(
         text(_NEW_VULNS_SQL).bindparams(
-            bindparam("sid", scan.id),
+            *_scope(scan),
             bindparam("suppressed", list(SUPPRESSED_STATES), type_=ARRAY(String)),
         )
     ).all()
@@ -184,45 +216,42 @@ def _dropped_hosts(session: Session, scan: Scan) -> int:
     return sum(len(entry or []) for entry in rows)
 
 
-def _notify_deltas(
-    notifier: SyncNotificationPublisher, session: Session, scan: Scan, target: str
-) -> None:
-    """Announce what this scan found that the previous one did not."""
+def _guard(fn, fallback):
     try:
-        new_subs = _count_new_subdomains(session, scan)
-        if new_subs > 0:
-            _notify(
-                notifier, session, scan_new_subdomains(str(scan.id), target, new_subs)
-            )
+        return fn()
     except Exception:
-        logger.warning("new-subdomain notification failed", exc_info=True)
-    try:
-        new_services, sensitive = _count_new_services(session, scan)
-        if new_services > 0:
-            _notify(
-                notifier,
-                session,
-                scan_new_services(str(scan.id), target, new_services, sensitive),
-            )
-    except Exception:
-        logger.warning("new-service notification failed", exc_info=True)
-    try:
-        counts, kev, total = _count_new_vulnerabilities(session, scan)
-        if total > 0:
-            _notify(
-                notifier,
-                session,
-                scan_new_vulnerabilities(str(scan.id), target, counts, kev, total),
-            )
-        dropped = _dropped_hosts(session, scan)
-        if dropped > 0:
-            _notify(
-                notifier,
-                session,
-                scan_vulnerability_coverage(str(scan.id), target, dropped),
-            )
-    except Exception:
-        logger.warning("vulnerability notification failed", exc_info=True)
+        logger.warning("scan delta measurement failed", exc_info=True)
+        return fallback
+
+
+def _measure(session: Session, scan: Scan) -> ScanDeltas:
+    """Each dimension is only compared where an earlier census run covered it."""
+    hosts, services, vulns = (
+        _guard(lambda sql=sql: _has_baseline(session, scan, sql), False)
+        for sql in (_HOST_BASELINE_SQL, _SERVICE_BASELINE_SQL, _VULN_BASELINE_SQL)
+    )
+    new_services, sensitive = (
+        _guard(lambda: _count_new_services(session, scan), (0, 0))
+        if services
+        else (0, 0)
+    )
+    vuln_counts, kev, new_vulns = (
+        _guard(lambda: _count_new_vulnerabilities(session, scan), ({}, 0, 0))
+        if vulns
+        else ({}, 0, 0)
+    )
+    return ScanDeltas(
+        baseline=hosts or services or vulns,
+        new_hosts=_guard(lambda: _count_new_subdomains(session, scan), 0)
+        if hosts
+        else 0,
+        new_services=new_services,
+        sensitive_services=sensitive,
+        new_vulnerabilities=new_vulns,
+        vulnerability_counts=vuln_counts,
+        kev=kev,
+        dropped_hosts=_guard(lambda: _dropped_hosts(session, scan), 0),
+    )
 
 
 def finalize_scan_run(session: Session, scan: Scan, *, redis_url: str) -> None:
@@ -287,11 +316,6 @@ def finalize_scan_run(session: Session, scan: Scan, *, redis_url: str) -> None:
         _log_cancelled(activity_log, scan)
         session.commit()
         events.scan_cancelled(status=status)
-        _notify(
-            notifier,
-            session,
-            scan_cancelled(str(scan.id), target_value, scan.engine_name),
-        )
         return
 
     if status == ScanStatus.COMPLETED.value:
@@ -311,11 +335,11 @@ def finalize_scan_run(session: Session, scan: Scan, *, redis_url: str) -> None:
             _notify(
                 notifier,
                 session,
-                scan_completed(
-                    str(scan.id), target_value, scan.engine_name, counts, duration
+                scan,
+                scan_digest(
+                    str(scan.id), target_value, counts, _measure(session, scan)
                 ),
             )
-        _notify_deltas(notifier, session, scan, target_value)
         return
 
     activity_log.log(
@@ -333,6 +357,7 @@ def finalize_scan_run(session: Session, scan: Scan, *, redis_url: str) -> None:
     _notify(
         notifier,
         session,
+        scan,
         scan_failed(
             str(scan.id), target_value, scan.engine_name, scan.error or "unknown error"
         ),
diff --git a/worker/app/tasks/notifications.py b/worker/app/tasks/notifications.py
new file mode 100644
index 000000000..30a858575
--- /dev/null
+++ b/worker/app/tasks/notifications.py
@@ -0,0 +1,24 @@
+"""Drop notifications past their expiry so the inbox stays a working set."""
+
+from celery import shared_task
+from sqlalchemy import delete
+
+from app.database import get_sync_session
+from shared.logging import get_logger
+from shared.models.notification import Notification
+from shared.utils.datetime import utc_now
+
+logger = get_logger(__name__)
+
+
+@shared_task(name="app.tasks.notifications.cleanup")
+def cleanup() -> dict:
+    with get_sync_session() as session:
+        result = session.execute(
+            delete(Notification).where(Notification.expires_at < utc_now())
+        )
+        session.commit()
+    removed = result.rowcount or 0
+    if removed:
+        logger.info("expired notifications removed", removed=removed)
+    return {"removed": removed}
diff --git a/worker/app/tasks/reports.py b/worker/app/tasks/reports.py
index 089d5e0b8..629affc71 100644
--- a/worker/app/tasks/reports.py
+++ b/worker/app/tasks/reports.py
@@ -211,6 +211,7 @@ def _notify(session, report: Report, *, ok: bool) -> None:
             NotificationSeverity.SUCCESS if ok else NotificationSeverity.ERROR,
             "Report ready" if ok else "Report failed",
             body,
+            project_id=report.project_id,
         )
     except Exception:
         logger.debug("report notification skipped", exc_info=True)
diff --git a/worker/app/tasks/ripestat.py b/worker/app/tasks/ripestat.py
index 4b7d05819..428003694 100644
--- a/worker/app/tasks/ripestat.py
+++ b/worker/app/tasks/ripestat.py
@@ -14,8 +14,8 @@ from app.celery import celery_app
 from app.config import settings
 from app.database import get_sync_session
 from shared.definitions.notifications import (
-    ripestat_enrichment_complete,
     ripestat_enrichment_failed,
+    ripestat_enrichment_incomplete,
 )
 from shared.enums.activity import ActivityEvent, ActivityLevel
 from shared.enums.target import TargetType
@@ -24,7 +24,10 @@ from shared.logging import get_logger
 from shared.models.target import Target
 from shared.services.activity_log import ActivityLogService
 from shared.services.bgp_summary import write_bgp_summary_for_target
-from shared.services.notification_sync import SyncNotificationPublisher
+from shared.services.notification_sync import (
+    SyncNotificationPublisher,
+    single_project,
+)
 from tools.ripestat.client import RIPEStatRateLimitError
 from tools.ripestat.service import RIPEStatLookupError, RIPEStatService
 
@@ -81,19 +84,21 @@ def enrich_targets_bgp(target_ids: list[str]) -> dict:
                 failed += 1
 
         total = success + failed + skipped
-        template = ripestat_enrichment_complete(
+        template = ripestat_enrichment_incomplete(
             success=success,
             failed=failed,
             skipped=skipped,
             total=total,
         )
-        notifier.publish(
-            session=session,
-            type=template["type"],
-            severity=template["severity"],
-            title=template["title"],
-            message=template["message"],
-        )
+        if template:
+            notifier.publish(
+                session=session,
+                type=template["type"],
+                severity=template["severity"],
+                title=template["title"],
+                message=template["message"],
+                project_id=single_project(targets),
+            )
 
         logger.info(
             "RIPEstat enrichment complete: %d success, %d failed, %d skipped",
@@ -149,8 +154,8 @@ def _enrich_target(
             target.bgp_status = TaskStatus.SUCCESS
             activity.log(
                 event=ActivityEvent.TARGET_ENRICHMENT_BGP_COMPLETED,
-                title="BGP enrichment completed.",
-                description=f"{count} lookups performed",
+                title=f"BGP enrichment completed · {target.target_value}",
+                description=f"{count} {'lookup' if count == 1 else 'lookups'} performed",
                 level=ActivityLevel.SUCCESS,
                 target_id=target.id,
                 project_id=target.project_id,
@@ -159,7 +164,7 @@ def _enrich_target(
             target.bgp_status = TaskStatus.FAILED
             activity.log(
                 event=ActivityEvent.TARGET_ENRICHMENT_BGP_FAILED,
-                title="BGP enrichment failed.",
+                title=f"BGP enrichment failed · {target.target_value}",
                 level=ActivityLevel.ERROR,
                 target_id=target.id,
                 project_id=target.project_id,
diff --git a/worker/app/tasks/scan.py b/worker/app/tasks/scan.py
index 180f79e35..ed46c399a 100644
--- a/worker/app/tasks/scan.py
+++ b/worker/app/tasks/scan.py
@@ -7,13 +7,11 @@ from celery import shared_task
 from app.config import settings
 from app.database import get_sync_session
 from app.orchestrator import build_canvas, finalize_scan_run, run_stage
-from shared.definitions.notifications import scan_started
 from shared.enums.activity import ActivityEvent, ActivityLevel
-from shared.enums.scan import SCAN_TERMINAL_STATUSES, ScanScope, ScanStatus
+from shared.enums.scan import SCAN_TERMINAL_STATUSES, ScanStatus
 from shared.logging import get_logger
 from shared.models.scan import Scan
 from shared.services.activity_log import ActivityLogService
-from shared.services.notification_sync import SyncNotificationPublisher
 from shared.services.orchestrator.events import ScanEventPublisher
 from shared.utils.datetime import utc_now
 from stages.registry import get_stage
@@ -24,20 +22,6 @@ logger = get_logger(__name__)
 _DISPATCHED_TASK_IDS = 2
 
 
-def _send_notif(redis_url: str, session, payload: dict) -> None:
-    try:
-        SyncNotificationPublisher(redis_url).publish(
-            session=session,
-            type=payload["type"],
-            severity=payload["severity"],
-            title=payload["title"],
-            message=payload["message"],
-            metadata=payload.get("metadata"),
-        )
-    except Exception:
-        logger.warning("scan notification dispatch failed", exc_info=True)
-
-
 @shared_task(bind=True, name="app.tasks.scan.run_scan", max_retries=0)
 def run_scan(self, scan_id: str) -> dict:
     """Orchestrator entrypoint: claim RUNNING, dispatch the stage canvas, then notify."""
@@ -88,12 +72,6 @@ def run_scan(self, scan_id: str) -> dict:
             )
             session.commit()
             events.scan_started(status=scan.status, engine=scan.engine_name)
-            if scan.scope != ScanScope.FOCUSED.value:
-                _send_notif(
-                    redis_url,
-                    session,
-                    scan_started(scan_id, target_value, scan.engine_name),
-                )
         return {"dispatched": True, "task_id": result.id}
 
 
diff --git a/worker/app/tasks/whois.py b/worker/app/tasks/whois.py
index 8d7a47a45..b5cd3fa1f 100644
--- a/worker/app/tasks/whois.py
+++ b/worker/app/tasks/whois.py
@@ -4,15 +4,18 @@ from app.celery import celery_app
 from app.config import settings
 from app.database import get_sync_session
 from shared.definitions.notifications import (
-    whois_enrichment_complete,
     whois_enrichment_failed,
+    whois_enrichment_incomplete,
 )
 from shared.enums.activity import ActivityEvent, ActivityLevel
 from shared.enums.task_status import TaskStatus
 from shared.logging import get_logger
 from shared.models.target import Target
 from shared.services.activity_log import ActivityLogService
-from shared.services.notification_sync import SyncNotificationPublisher
+from shared.services.notification_sync import (
+    SyncNotificationPublisher,
+    single_project,
+)
 from shared.utils.datetime import utc_now
 from shared.utils.validation import normalize_query
 from tools.whois.service import WhoisService
@@ -79,7 +82,7 @@ def perform_whois_lookups(target_ids: list[str]) -> dict:  # noqa: PLR0915
                     success_count += 1
                     activity.log(
                         event=ActivityEvent.TARGET_ENRICHMENT_WHOIS_COMPLETED,
-                        title="WHOIS query completed.",
+                        title=f"WHOIS lookup completed · {target.target_value}",
                         level=ActivityLevel.SUCCESS,
                         target_id=target.id,
                         project_id=target.project_id,
@@ -99,7 +102,7 @@ def perform_whois_lookups(target_ids: list[str]) -> dict:  # noqa: PLR0915
                     failed_count += 1
                     activity.log(
                         event=ActivityEvent.TARGET_ENRICHMENT_WHOIS_FAILED,
-                        title="WHOIS query failed.",
+                        title=f"WHOIS lookup failed · {target.target_value}",
                         description=error_msg,
                         level=ActivityLevel.ERROR,
                         target_id=target.id,
@@ -108,16 +111,18 @@ def perform_whois_lookups(target_ids: list[str]) -> dict:  # noqa: PLR0915
                 session.commit()
 
         total = success_count + failed_count
-        template = whois_enrichment_complete(
+        template = whois_enrichment_incomplete(
             success=success_count, failed=failed_count, total=total
         )
-        notifier.publish(
-            session=session,
-            type=template["type"],
-            severity=template["severity"],
-            title=template["title"],
-            message=template["message"],
-        )
+        if template:
+            notifier.publish(
+                session=session,
+                type=template["type"],
+                severity=template["severity"],
+                title=template["title"],
+                message=template["message"],
+                project_id=single_project(targets),
+            )
 
         return {"success": success_count, "failed": failed_count, "total": total}
 
@@ -141,7 +146,7 @@ def perform_whois_lookups(target_ids: list[str]) -> dict:  # noqa: PLR0915
                 target.updated_at = utc_now()
                 activity.log(
                     event=ActivityEvent.TARGET_ENRICHMENT_WHOIS_FAILED,
-                    title=f"WHOIS failed for {target.target_value}",
+                    title=f"WHOIS lookup failed · {target.target_value}",
                     description=str(e)[:1000],
                     level=ActivityLevel.ERROR,
                     target_id=target.id,