diff --git a/apps/web/src/lib/components/datasets/DatasetTabs.svelte b/apps/web/src/lib/components/datasets/DatasetTabs.svelte
new file mode 100644
index 00000000..4c84d4ef
--- /dev/null
+++ b/apps/web/src/lib/components/datasets/DatasetTabs.svelte
@@ -0,0 +1,30 @@
+<script lang="ts">
+	import { resolve } from '$app/paths';
+
+	let { datasetId, active }: { datasetId: string; active: 'dashboard' | 'alerts' } = $props();
+
+	const CONTROL_GROUP_CLASS =
+		'dark:border-dark-border dark:bg-dark-subtle grid w-full grid-cols-2 gap-0.5 rounded-md border border-gray-200 bg-gray-50 p-1 sm:w-fit';
+	const CONTROL_BUTTON_CLASS =
+		'flex min-h-7 items-center justify-center rounded px-2.5 py-0.5 text-center text-xs font-medium transition-colors focus:ring-2 focus:ring-blue-500 focus:outline-none';
+	const CONTROL_BUTTON_INACTIVE_CLASS =
+		'text-gray-700 hover:text-gray-900 dark:text-gray-300 dark:hover:text-gray-100';
+	const CONTROL_BUTTON_ACTIVE_CLASS = 'bg-blue-600 text-white shadow-sm';
+</script>
+
+<nav aria-label="Dataset views" class={CONTROL_GROUP_CLASS}>
+	<a
+		href={resolve('/datasets/[dataset]', { dataset: datasetId })}
+		class={`${CONTROL_BUTTON_CLASS} ${
+			active === 'dashboard' ? CONTROL_BUTTON_ACTIVE_CLASS : CONTROL_BUTTON_INACTIVE_CLASS
+		}`}
+		aria-current={active === 'dashboard' ? 'page' : undefined}>Dashboard</a
+	>
+	<a
+		href={resolve('/datasets/[dataset]/alerts', { dataset: datasetId })}
+		class={`${CONTROL_BUTTON_CLASS} ${
+			active === 'alerts' ? CONTROL_BUTTON_ACTIVE_CLASS : CONTROL_BUTTON_INACTIVE_CLASS
+		}`}
+		aria-current={active === 'alerts' ? 'page' : undefined}>Alerts</a
+	>
+</nav>
diff --git a/apps/web/src/lib/components/netflow/DatasetDashboardPage.svelte b/apps/web/src/lib/components/netflow/DatasetDashboardPage.svelte
index 5d48cd11..82b80322 100644
--- a/apps/web/src/lib/components/netflow/DatasetDashboardPage.svelte
+++ b/apps/web/src/lib/components/netflow/DatasetDashboardPage.svelte
@@ -1,5 +1,6 @@
 <script lang="ts">
 	import { onMount } from 'svelte';
+	import DatasetTabs from '$lib/components/datasets/DatasetTabs.svelte';
 	import PrimaryFilters from '$lib/components/filters/PrimaryFilters.svelte';
 	import NetflowDashboard from '$lib/components/netflow/NetflowDashboard.svelte';
 	import IPChart from '$lib/components/charts/IPChart.svelte';
@@ -374,6 +375,7 @@
 	<h1 class="px-1 text-2xl font-semibold text-gray-900 dark:text-gray-100">
 		{props.title ?? props.dataset}
 	</h1>
+	<DatasetTabs datasetId={props.dataset} active="dashboard" />
 
 	<PrimaryFilters
 		{startDate}
@@ -388,7 +390,6 @@
 		on:scopeChange={handleScopeChange}
 		on:resetView={handleResetView}
 	/>
-
 	<div role="list" aria-label="Reorderable charts" class="flex flex-col gap-2">
 		{#each chartOrder as chartId (chartId)}
 			<section
diff --git a/apps/web/src/lib/server/alerts.ts b/apps/web/src/lib/server/alerts.ts
new file mode 100644
index 00000000..a7e98926
--- /dev/null
+++ b/apps/web/src/lib/server/alerts.ts
@@ -0,0 +1,454 @@
+import { env as privateEnv } from '$env/dynamic/private';
+import type Database from 'better-sqlite3';
+import type {
+	AlertFeedAddress,
+	AlertHorizon,
+	AlertSort,
+	AlertsFeedResponse,
+	AlertTail
+} from '$lib/types/types';
+
+type QueryParam = string | number | boolean | null | Uint8Array;
+
+type SqliteClient = Database.Database;
+
+type LocalDbIdentity = {
+	device: number;
+	inode: number;
+};
+
+type LocalDbCacheEntry = {
+	db: SqliteClient;
+	identity: LocalDbIdentity;
+};
+
+type FeedMetaRow = {
+	key: string;
+	value: string;
+};
+
+type LatestWindowRow = {
+	windowStart: number;
+	windowEnd: number;
+	addressCount: number;
+	processedAt: number;
+};
+
+type AddressRow = AlertFeedAddress & {
+	totalAddresses: number;
+};
+
+const HORIZON_SECONDS: Record<AlertHorizon, number> = {
+	'1h': 60 * 60,
+	'6h': 6 * 60 * 60,
+	'24h': 24 * 60 * 60,
+	'7d': 7 * 24 * 60 * 60
+};
+const DEFAULT_HORIZON: AlertHorizon = '24h';
+const DEFAULT_LIMIT = 100;
+const MAX_LIMIT = 500;
+const REQUIRED_TABLES = ['alerts', 'feed_meta', 'windows'] as const;
+const REQUIRED_META_KEYS = [
+	'schema_version',
+	'dataset_id',
+	'threshold_high',
+	'threshold_low',
+	'max_per_tail'
+] as const;
+
+const localDbCache = new Map<string, LocalDbCacheEntry>();
+
+export type AlertsFeedOptions = {
+	platform?: App.Platform;
+	tail?: AlertTail;
+	horizon?: AlertHorizon;
+	sort?: AlertSort;
+	limit?: number;
+};
+
+function absentFeed(horizonSeconds: number): AlertsFeedResponse {
+	return {
+		feed: { present: false },
+		horizonSeconds,
+		totalAddresses: 0,
+		addresses: []
+	};
+}
+
+function getEnv(name: string): string | undefined {
+	return globalThis.process?.env?.[name]?.trim() || privateEnv[name]?.trim() || undefined;
+}
+
+function shouldUseD1(platform?: App.Platform): boolean {
+	return getEnv('ATLANTIS_DB_DRIVER') !== 'sqlite' && Boolean(platform?.env.DB);
+}
+
+async function resolvePath(value: string): Promise<string> {
+	if (value === ':memory:') {
+		return value;
+	}
+
+	const path = await import('node:path');
+	return path.isAbsolute(value) ? value : path.resolve(process.cwd(), value);
+}
+
+async function discoverLocalSqlitePaths(): Promise<string[]> {
+	const configured = getEnv('LOCAL_SQLITE_PATH') ?? getEnv('DATABASE_PATH');
+	if (configured) {
+		return [await resolvePath(configured)];
+	}
+
+	const fs = await import('node:fs/promises');
+	const path = await import('node:path');
+	const roots = [path.resolve(process.cwd(), 'data'), path.resolve(process.cwd(), '../../data')];
+	const dbPaths = new Set<string>();
+
+	for (const root of roots) {
+		let entries: import('node:fs').Dirent[];
+		try {
+			entries = await fs.readdir(root, { withFileTypes: true });
+		} catch {
+			continue;
+		}
+
+		for (const entry of entries) {
+			if (!entry.isDirectory()) {
+				continue;
+			}
+
+			const dbPath = path.join(root, entry.name, 'netflow.sqlite');
+			try {
+				const stat = await fs.stat(dbPath);
+				if (stat.isFile()) {
+					dbPaths.add(dbPath);
+				}
+			} catch {
+				// Not every data directory is a web dataset.
+			}
+		}
+	}
+
+	return [...dbPaths].sort();
+}
+
+async function openReadonlyClient(dbPath: string): Promise<SqliteClient> {
+	const betterSqlite3 = await import(/* @vite-ignore */ 'better-sqlite3');
+	const sqlite = new betterSqlite3.default(dbPath, { readonly: true, fileMustExist: true });
+	sqlite.pragma('query_only = ON');
+	sqlite.pragma('busy_timeout = 60000');
+	return sqlite;
+}
+
+async function resolveDatasetDirectory(datasetId: string): Promise<string | null> {
+	const path = await import('node:path');
+	for (const dbPath of await discoverLocalSqlitePaths()) {
+		if (dbPath === ':memory:') {
+			continue;
+		}
+
+		let db: SqliteClient | undefined;
+		try {
+			db = await openReadonlyClient(dbPath);
+			const row = db.prepare('SELECT id FROM datasets WHERE id = ? LIMIT 1').get(datasetId) as
+				| { id: string }
+				| undefined;
+			if (row) {
+				return path.dirname(dbPath);
+			}
+		} catch {
+			// A broken candidate cannot contain a usable local dataset.
+		} finally {
+			db?.close();
+		}
+	}
+
+	return null;
+}
+
+async function localDbIdentity(dbPath: string): Promise<LocalDbIdentity> {
+	const fs = await import('node:fs/promises');
+	const stat = await fs.stat(dbPath);
+	return { device: stat.dev, inode: stat.ino };
+}
+
+function sameLocalDbIdentity(left: LocalDbIdentity, right: LocalDbIdentity): boolean {
+	return left.device === right.device && left.inode === right.inode;
+}
+
+function evictLocalDb(dbPath: string): void {
+	const existing = localDbCache.get(dbPath);
+	if (!existing) {
+		return;
+	}
+
+	localDbCache.delete(dbPath);
+	existing.db.close();
+}
+
+async function createLocalDb(dbPath: string): Promise<LocalDbCacheEntry> {
+	for (let attempt = 0; attempt < 3; attempt += 1) {
+		const identityBeforeOpen = await localDbIdentity(dbPath);
+		const db = await openReadonlyClient(dbPath);
+		try {
+			const identityAfterOpen = await localDbIdentity(dbPath);
+			if (sameLocalDbIdentity(identityBeforeOpen, identityAfterOpen)) {
+				return { db, identity: identityAfterOpen };
+			}
+		} catch (error) {
+			db.close();
+			throw error;
+		}
+		db.close();
+	}
+
+	throw new Error(`Local alerts database kept changing while opening: ${dbPath}`);
+}
+
+async function getLocalDb(dbPath: string): Promise<SqliteClient> {
+	try {
+		const identity = await localDbIdentity(dbPath);
+		const existing = localDbCache.get(dbPath);
+		if (existing && sameLocalDbIdentity(existing.identity, identity)) {
+			return existing.db;
+		}
+
+		evictLocalDb(dbPath);
+		const entry = await createLocalDb(dbPath);
+		localDbCache.set(dbPath, entry);
+		return entry.db;
+	} catch (error) {
+		evictLocalDb(dbPath);
+		throw error;
+	}
+}
+
+function clampLimit(limit: number | undefined): number {
+	if (limit === undefined || !Number.isFinite(limit)) {
+		return DEFAULT_LIMIT;
+	}
+	return Math.min(MAX_LIMIT, Math.max(1, Math.trunc(limit)));
+}
+
+function readFeedMetadata(
+	db: SqliteClient,
+	datasetId: string
+): { thresholdHigh: number; thresholdLow: number } {
+	const tables = db
+		.prepare(
+			`SELECT name FROM sqlite_master WHERE type = 'table' AND name IN (${REQUIRED_TABLES.map(() => '?').join(', ')})`
+		)
+		.all(...REQUIRED_TABLES) as { name: string }[];
+	if (new Set(tables.map((row) => row.name)).size !== REQUIRED_TABLES.length) {
+		throw new Error('Alerts database is missing required tables');
+	}
+	db.prepare(
+		`SELECT
+			window_start,
+			window_end,
+			member_files,
+			address_count,
+			alert_count,
+			alpha_min,
+			alpha_max,
+			alpha_median,
+			processed_at
+		FROM windows
+		LIMIT 0`
+	).all();
+	db.prepare(
+		`SELECT window_start, address, alpha, tail, rank, r2, prefix_levels
+		FROM alerts
+		LIMIT 0`
+	).all();
+
+	const rows = db
+		.prepare(
+			`SELECT key, value FROM feed_meta WHERE key IN (${REQUIRED_META_KEYS.map(() => '?').join(', ')})`
+		)
+		.all(...REQUIRED_META_KEYS) as FeedMetaRow[];
+	const metadata = new Map(rows.map((row) => [row.key, row.value]));
+	const thresholdHigh = Number(metadata.get('threshold_high'));
+	const thresholdLow = Number(metadata.get('threshold_low'));
+	const maxPerTail = Number(metadata.get('max_per_tail'));
+
+	if (
+		metadata.get('schema_version') !== '1' ||
+		metadata.get('dataset_id') !== datasetId ||
+		!Number.isFinite(thresholdHigh) ||
+		!Number.isFinite(thresholdLow) ||
+		!Number.isSafeInteger(maxPerTail) ||
+		maxPerTail < 1
+	) {
+		throw new Error('Alerts database metadata is invalid');
+	}
+
+	return { thresholdHigh, thresholdLow };
+}
+
+function readAddresses(
+	db: SqliteClient,
+	options: Pick<AlertsFeedOptions, 'tail' | 'sort' | 'limit'>,
+	horizonStart: number,
+	thresholdHigh: number,
+	thresholdLow: number
+): { totalAddresses: number; addresses: AlertFeedAddress[] } {
+	const whereTail = options.tail ? 'AND tail = ?' : '';
+	const sortOrder =
+		options.sort === 'recent'
+			? 'lastSeen DESC, severity DESC, address ASC'
+			: 'severity DESC, address ASC';
+	const limit = clampLimit(options.limit);
+	const params: QueryParam[] = [thresholdHigh, thresholdLow, horizonStart];
+	if (options.tail) {
+		params.push(options.tail);
+	}
+	params.push(limit);
+
+	const rows = db
+		.prepare(
+			`
+				WITH scoped AS (
+					SELECT
+						address,
+						alpha,
+						tail,
+						rank,
+						r2,
+						window_start,
+						CASE tail
+							WHEN 'high' THEN alpha - ?
+							ELSE ? - alpha
+						END AS severity
+					FROM alerts
+					WHERE window_start >= ?
+				${whereTail}
+				),
+				ranked AS (
+					SELECT
+						address,
+						alpha,
+						tail,
+						r2,
+						window_start,
+						severity,
+						ROW_NUMBER() OVER (
+							PARTITION BY address
+							ORDER BY severity DESC, window_start DESC, rank ASC
+						) AS peak_rank,
+						ROW_NUMBER() OVER (
+							PARTITION BY address
+							ORDER BY window_start DESC, rank ASC
+						) AS recent_rank,
+						MAX(window_start) OVER (PARTITION BY address) AS last_seen,
+						COUNT(*) OVER (PARTITION BY address) AS times_flagged
+					FROM scoped
+				),
+				enriched AS (
+					SELECT
+						ranked.*,
+						MAX(CASE WHEN recent_rank = 1 THEN alpha END)
+							OVER (PARTITION BY address) AS latest_alpha
+					FROM ranked
+				)
+				SELECT
+					address,
+					tail,
+					alpha AS peakAlpha,
+					window_start AS peakWindowStart,
+					r2 AS peakR2,
+					latest_alpha AS latestAlpha,
+					last_seen AS lastSeen,
+					-- Novelty is judged against the whole retained history, not the
+					-- horizon or tail filter: "new" means never flagged before at all.
+					(SELECT MIN(history.window_start)
+						FROM alerts AS history
+						WHERE history.address = enriched.address) AS firstSeen,
+					times_flagged AS timesFlagged,
+					COUNT(*) OVER () AS totalAddresses
+				FROM enriched
+				WHERE peak_rank = 1
+				ORDER BY ${sortOrder}
+				LIMIT ?
+			`
+		)
+		.all(...params) as AddressRow[];
+
+	return {
+		totalAddresses: rows[0]?.totalAddresses ?? 0,
+		addresses: rows.map((row) => ({
+			address: row.address,
+			tail: row.tail,
+			peakAlpha: row.peakAlpha,
+			peakWindowStart: row.peakWindowStart,
+			peakR2: row.peakR2,
+			latestAlpha: row.latestAlpha,
+			lastSeen: row.lastSeen,
+			firstSeen: row.firstSeen,
+			timesFlagged: row.timesFlagged
+		}))
+	};
+}
+
+export async function getAlertsFeedForDataset(
+	datasetId: string,
+	options: AlertsFeedOptions = {}
+): Promise<AlertsFeedResponse> {
+	const horizonSeconds = HORIZON_SECONDS[options.horizon ?? DEFAULT_HORIZON];
+	if (shouldUseD1(options.platform)) {
+		return absentFeed(horizonSeconds);
+	}
+
+	let alertsDbPath: string | undefined;
+	try {
+		const datasetDirectory = await resolveDatasetDirectory(datasetId);
+		if (!datasetDirectory) {
+			return absentFeed(horizonSeconds);
+		}
+
+		const path = await import('node:path');
+		alertsDbPath = path.join(datasetDirectory, 'alerts.sqlite');
+		const db = await getLocalDb(alertsDbPath);
+		const { thresholdHigh, thresholdLow } = readFeedMetadata(db, datasetId);
+		const latestWindow = db
+			.prepare(
+				`
+					SELECT
+						window_start AS windowStart,
+						window_end AS windowEnd,
+						address_count AS addressCount,
+						processed_at AS processedAt
+					FROM windows
+					ORDER BY window_start DESC
+					LIMIT 1
+				`
+			)
+			.get() as LatestWindowRow | undefined;
+		const result = latestWindow
+			? readAddresses(
+					db,
+					options,
+					latestWindow.windowEnd - horizonSeconds,
+					thresholdHigh,
+					thresholdLow
+				)
+			: { totalAddresses: 0, addresses: [] };
+
+		return {
+			feed: {
+				present: true,
+				latestWindowStart: latestWindow?.windowStart ?? null,
+				latestWindowEnd: latestWindow?.windowEnd ?? null,
+				latestAddressCount: latestWindow?.addressCount ?? null,
+				latestProcessedAt: latestWindow?.processedAt ?? null,
+				thresholds: { high: thresholdHigh, low: thresholdLow }
+			},
+			horizonSeconds,
+			...result
+		};
+	} catch {
+		if (alertsDbPath) {
+			evictLocalDb(alertsDbPath);
+		}
+		return absentFeed(horizonSeconds);
+	}
+}
diff --git a/apps/web/src/lib/types/types.ts b/apps/web/src/lib/types/types.ts
index b12d518f..07ff8547 100644
--- a/apps/web/src/lib/types/types.ts
+++ b/apps/web/src/lib/types/types.ts
@@ -11,6 +11,43 @@ export interface DatasetSummariesResponse {
 	error: string | null;
 }
 
+export type AlertTail = 'high' | 'low';
+
+export type AlertHorizon = '1h' | '6h' | '24h' | '7d';
+
+export type AlertSort = 'extreme' | 'recent';
+
+export interface AlertFeedAddress {
+	address: string;
+	tail: AlertTail;
+	peakAlpha: number;
+	peakWindowStart: number;
+	peakR2: number;
+	/** Alpha of the address's most recent crossing within the horizon. */
+	latestAlpha: number;
+	lastSeen: number;
+	firstSeen: number;
+	timesFlagged: number;
+}
+
+export type AlertFeedStatus =
+	| { present: false }
+	| {
+			present: true;
+			latestWindowStart: number | null;
+			latestWindowEnd: number | null;
+			latestAddressCount: number | null;
+			latestProcessedAt: number | null;
+			thresholds: { high: number; low: number };
+	  };
+
+export interface AlertsFeedResponse {
+	feed: AlertFeedStatus;
+	horizonSeconds: number;
+	totalAddresses: number;
+	addresses: AlertFeedAddress[];
+}
+
 export type CoverageState = 'complete' | 'partial' | 'unknown';
 
 export interface BucketCoverage {
diff --git a/apps/web/src/routes/api/alerts/+server.ts b/apps/web/src/routes/api/alerts/+server.ts
new file mode 100644
index 00000000..b918ce15
--- /dev/null
+++ b/apps/web/src/routes/api/alerts/+server.ts
@@ -0,0 +1,74 @@
+import { json } from '@sveltejs/kit';
+import type { RequestHandler } from './$types';
+import { getAlertsFeedForDataset } from '$lib/server/alerts';
+import { getRequestedDataset } from '$lib/server/datasets';
+import type { AlertHorizon, AlertsFeedResponse, AlertSort, AlertTail } from '$lib/types/types';
+
+type ErrorResponse = {
+	data: null;
+	error: string;
+};
+
+function parseInteger(value: string | null): number | undefined | null {
+	if (value === null) {
+		return undefined;
+	}
+	const normalized = value.trim();
+	if (!/^-?\d+$/.test(normalized)) {
+		return null;
+	}
+
+	const parsed = Number(normalized);
+	return Number.isSafeInteger(parsed) ? parsed : null;
+}
+
+function errorResponse(message: string, status: number): Response {
+	const response: ErrorResponse = { data: null, error: message };
+	return json(response, { status });
+}
+
+export const GET: RequestHandler = async ({ url, platform }) => {
+	const tailParam = url.searchParams.get('tail');
+	if (tailParam !== null && tailParam !== 'high' && tailParam !== 'low') {
+		return errorResponse('Invalid tail parameter', 400);
+	}
+	const tail = tailParam as AlertTail | null;
+
+	const horizonParam = url.searchParams.get('horizon') ?? '24h';
+	if (
+		horizonParam !== '1h' &&
+		horizonParam !== '6h' &&
+		horizonParam !== '24h' &&
+		horizonParam !== '7d'
+	) {
+		return errorResponse('Invalid horizon parameter', 400);
+	}
+	const horizon = horizonParam as AlertHorizon;
+
+	const sortParam = url.searchParams.get('sort') ?? 'extreme';
+	if (sortParam !== 'extreme' && sortParam !== 'recent') {
+		return errorResponse('Invalid sort parameter', 400);
+	}
+	const sort = sortParam as AlertSort;
+
+	const parsedLimit = parseInteger(url.searchParams.get('limit'));
+	if (parsedLimit === null) {
+		return errorResponse('Invalid limit parameter', 400);
+	}
+	const limit = Math.min(500, Math.max(1, parsedLimit ?? 100));
+
+	try {
+		const dataset = await getRequestedDataset(url, platform);
+		const response: AlertsFeedResponse = await getAlertsFeedForDataset(dataset, {
+			platform,
+			tail: tail ?? undefined,
+			horizon,
+			sort,
+			limit
+		});
+		return json(response);
+	} catch (error) {
+		console.error('Failed to load alerts feed:', error);
+		return errorResponse('Failed to load alerts feed', 500);
+	}
+};
diff --git a/apps/web/src/routes/datasets/[dataset]/alerts/+page.svelte b/apps/web/src/routes/datasets/[dataset]/alerts/+page.svelte
new file mode 100644
index 00000000..f5cb353c
--- /dev/null
+++ b/apps/web/src/routes/datasets/[dataset]/alerts/+page.svelte
@@ -0,0 +1,570 @@
+<script lang="ts">
+	import { afterNavigate } from '$app/navigation';
+	import { untrack } from 'svelte';
+	import type { PageProps } from './$types';
+	import DatasetTabs from '$lib/components/datasets/DatasetTabs.svelte';
+	import type { AlertHorizon, AlertsFeedResponse, AlertSort, AlertTail } from '$lib/types/types';
+
+	type TailSelection = AlertTail;
+	type ErrorResponse = { data: null; error: string };
+
+	const PAGE_SIZE = 100;
+	const MAX_LIMIT = 500;
+	const REFRESH_INTERVAL_MS = 30_000;
+	const LIVE_WINDOW_AGE_MS = 15 * 60_000;
+	const NEW_ADDRESS_AGE_SECONDS = 15 * 60;
+	const CONTROL_GROUP_CLASS =
+		'dark:border-dark-border dark:bg-dark-subtle grid w-full gap-0.5 rounded-md border border-gray-200 bg-gray-50 p-1 sm:w-fit';
+	const CONTROL_BUTTON_CLASS =
+		'flex min-h-7 items-center justify-center rounded px-2.5 py-0.5 text-center text-xs font-medium transition-colors focus:ring-2 focus:ring-blue-500 focus:outline-none';
+	const CONTROL_BUTTON_INACTIVE_CLASS =
+		'text-gray-700 hover:text-gray-900 dark:text-gray-300 dark:hover:text-gray-100';
+	const CONTROL_BUTTON_ACTIVE_CLASS = 'bg-blue-600 text-white shadow-sm';
+	const TAIL_OPTIONS: Array<{ value: TailSelection; label: string; title: string }> = [
+		{
+			value: 'high',
+			label: 'High α',
+			title: 'Isolated addresses in sparse regions of address space'
+		},
+		{ value: 'low', label: 'Low α', title: 'Addresses deep inside dense clusters' }
+	];
+	const HORIZON_OPTIONS: Array<{ value: AlertHorizon; label: string }> = [
+		{ value: '1h', label: '1h' },
+		{ value: '6h', label: '6h' },
+		{ value: '24h', label: '24h' },
+		{ value: '7d', label: '7d' }
+	];
+	const SORT_OPTIONS: Array<{ value: AlertSort; label: string }> = [
+		{ value: 'extreme', label: 'Most extreme' },
+		{ value: 'recent', label: 'Recent' }
+	];
+	const timeFormatter = new Intl.DateTimeFormat(undefined, {
+		hour: '2-digit',
+		minute: '2-digit'
+	});
+	const dateTimeFormatter = new Intl.DateTimeFormat(undefined, {
+		dateStyle: 'medium',
+		timeStyle: 'short'
+	});
+	const countFormatter = new Intl.NumberFormat();
+
+	let { data }: PageProps = $props();
+	let activeDataset = $state(untrack(() => data.selectedDataset));
+	let feedResponse = $state<AlertsFeedResponse>(untrack(() => data.alerts));
+	let selectedTail = $state<TailSelection>('high');
+	let selectedHorizon = $state<AlertHorizon>('24h');
+	let selectedSort = $state<AlertSort>('extreme');
+	let limit = $state(PAGE_SIZE);
+	let now = $state(Date.now());
+	let loading = $state(false);
+	let loadingMore = $state(false);
+	let fetchError = $state('');
+	let copied = $state(false);
+	let copyError = $state('');
+	let requestGeneration = 0;
+
+	const selectedDatasetLabel = $derived(
+		data.datasets.find((dataset) => dataset.datasetId === data.selectedDataset)?.label ??
+			data.selectedDataset
+	);
+	const feedCommand = $derived(`netflow-db feed ${data.selectedDataset}`);
+	const canShowMore = $derived(
+		feedResponse.addresses.length < feedResponse.totalAddresses && limit < MAX_LIMIT
+	);
+	const statusText = $derived.by(() => {
+		if (!feedResponse.feed.present) {
+			return 'Feed not running';
+		}
+		if (
+			feedResponse.feed.latestWindowStart === null ||
+			feedResponse.feed.latestWindowEnd === null ||
+			feedResponse.feed.latestAddressCount === null
+		) {
+			return 'Feed idle · no windows processed';
+		}
+
+		const windowStart = feedResponse.feed.latestWindowStart;
+		const windowEnd = feedResponse.feed.latestWindowEnd;
+		const windowAge = now - windowEnd * 1000;
+		if (windowAge >= 0 && windowAge < LIVE_WINDOW_AGE_MS) {
+			return `Live · last window ${formatTime(windowStart)}–${formatTime(windowEnd)} · ${countFormatter.format(feedResponse.feed.latestAddressCount)} addresses scored`;
+		}
+
+		return `Feed idle · last window ${dateTimeFormatter.format(new Date(windowEnd * 1000))}`;
+	});
+
+	function formatTime(timestamp: number): string {
+		return timeFormatter.format(new Date(timestamp * 1000));
+	}
+
+	function formatRelativeTime(timestamp: number): string {
+		const ageMs = Math.max(0, now - timestamp * 1000);
+		if (ageMs < 60_000) {
+			return 'just now';
+		}
+		if (ageMs < 60 * 60_000) {
+			return `${Math.floor(ageMs / 60_000)} min ago`;
+		}
+		if (ageMs < 24 * 60 * 60_000) {
+			return `${Math.floor(ageMs / (60 * 60_000))} h ago`;
+		}
+		return dateTimeFormatter.format(new Date(timestamp * 1000));
+	}
+
+	// "New" = the address's first appearance in the entire retained history
+	// (firstSeen is retention-wide, not horizon-scoped) is within the last few
+	// windows — it just entered the anomalous set for the first time.
+	function isNewAddress(firstSeen: number): boolean {
+		return (
+			feedResponse.feed.present &&
+			feedResponse.feed.latestWindowStart !== null &&
+			firstSeen >= feedResponse.feed.latestWindowStart - NEW_ADDRESS_AGE_SECONDS
+		);
+	}
+
+	function feedUrl(
+		tail: TailSelection,
+		horizon: AlertHorizon,
+		sort: AlertSort,
+		requestLimit: number
+	): string {
+		const params = [
+			`dataset=${encodeURIComponent(data.selectedDataset)}`,
+			`tail=${tail}`,
+			`horizon=${horizon}`,
+			`sort=${sort}`,
+			`limit=${requestLimit}`
+		];
+		return `/api/alerts?${params.join('&')}`;
+	}
+
+	async function requestFeed(
+		tail: TailSelection,
+		horizon: AlertHorizon,
+		sort: AlertSort,
+		requestLimit: number
+	): Promise<AlertsFeedResponse> {
+		const response = await fetch(feedUrl(tail, horizon, sort, requestLimit));
+		const payload = (await response.json()) as AlertsFeedResponse | ErrorResponse;
+		if (!response.ok || 'error' in payload) {
+			throw new Error('error' in payload ? payload.error : 'Failed to load alerts feed');
+		}
+		return payload;
+	}
+
+	async function loadFeed(
+		tail: TailSelection,
+		horizon: AlertHorizon,
+		sort: AlertSort,
+		requestLimit: number,
+		showLoading: boolean
+	): Promise<boolean> {
+		const generation = ++requestGeneration;
+		if (showLoading) {
+			loading = true;
+		}
+		try {
+			const nextResponse = await requestFeed(tail, horizon, sort, requestLimit);
+			if (generation !== requestGeneration) {
+				return false;
+			}
+
+			feedResponse = nextResponse;
+			fetchError = '';
+			return true;
+		} catch (error) {
+			if (generation === requestGeneration) {
+				fetchError = error instanceof Error ? error.message : 'Failed to load alerts feed';
+			}
+			return false;
+		} finally {
+			if (generation === requestGeneration && showLoading) {
+				loading = false;
+			}
+		}
+	}
+
+	async function selectTail(tail: TailSelection): Promise<void> {
+		if (tail === selectedTail) {
+			return;
+		}
+
+		selectedTail = tail;
+		limit = PAGE_SIZE;
+		await loadFeed(tail, selectedHorizon, selectedSort, limit, true);
+	}
+
+	async function selectHorizon(horizon: AlertHorizon): Promise<void> {
+		if (horizon === selectedHorizon) {
+			return;
+		}
+
+		selectedHorizon = horizon;
+		limit = PAGE_SIZE;
+		await loadFeed(selectedTail, horizon, selectedSort, limit, true);
+	}
+
+	async function selectSort(sort: AlertSort): Promise<void> {
+		if (sort === selectedSort) {
+			return;
+		}
+
+		selectedSort = sort;
+		limit = PAGE_SIZE;
+		await loadFeed(selectedTail, selectedHorizon, sort, limit, true);
+	}
+
+	async function refreshFeed(): Promise<void> {
+		if (loading || loadingMore) {
+			return;
+		}
+		await loadFeed(selectedTail, selectedHorizon, selectedSort, limit, false);
+	}
+
+	async function showMore(): Promise<void> {
+		if (!canShowMore || loading || loadingMore) {
+			return;
+		}
+
+		const nextLimit = Math.min(MAX_LIMIT, limit + PAGE_SIZE);
+		loadingMore = true;
+		try {
+			if (await loadFeed(selectedTail, selectedHorizon, selectedSort, nextLimit, false)) {
+				limit = nextLimit;
+			}
+		} finally {
+			loadingMore = false;
+		}
+	}
+
+	async function copyCommand(): Promise<void> {
+		try {
+			await navigator.clipboard.writeText(feedCommand);
+			copied = true;
+			copyError = '';
+		} catch {
+			copied = false;
+			copyError = 'Could not copy the command';
+		}
+	}
+
+	afterNavigate(() => {
+		if (activeDataset === data.selectedDataset) {
+			return;
+		}
+
+		requestGeneration += 1;
+		activeDataset = data.selectedDataset;
+		feedResponse = data.alerts;
+		selectedTail = 'high';
+		selectedHorizon = '24h';
+		selectedSort = 'extreme';
+		limit = PAGE_SIZE;
+		now = Date.now();
+		loading = false;
+		loadingMore = false;
+		fetchError = '';
+		copied = false;
+		copyError = '';
+	});
+
+	// The initial page comes from +page.ts. This one effect only owns polling and tab visibility.
+	$effect(() => {
+		let interval: ReturnType<typeof setInterval> | undefined;
+		const stopTimer = () => {
+			if (interval !== undefined) {
+				clearInterval(interval);
+				interval = undefined;
+			}
+		};
+		const startTimer = () => {
+			stopTimer();
+			if (document.visibilityState === 'visible') {
+				interval = setInterval(() => {
+					now = Date.now();
+					void refreshFeed();
+				}, REFRESH_INTERVAL_MS);
+			}
+		};
+		const handleVisibilityChange = () => {
+			if (document.visibilityState === 'visible') {
+				now = Date.now();
+				void refreshFeed();
+			}
+			startTimer();
+		};
+
+		document.addEventListener('visibilitychange', handleVisibilityChange);
+		startTimer();
+
+		return () => {
+			stopTimer();
+			document.removeEventListener('visibilitychange', handleVisibilityChange);
+		};
+	});
+</script>
+
+<svelte:head>
+	<title>Singularity alerts | ATLANTIS</title>
+	<meta
+		name="description"
+		content="Continuous Singularity alerts for anomalous network addresses"
+	/>
+</svelte:head>
+
+<main class="mx-auto flex max-w-[95vw] flex-col gap-4 px-4 py-8 sm:px-2 lg:px-4">
+	<header>
+		<h1 class="text-2xl font-semibold text-gray-900 dark:text-gray-100">Singularity alerts</h1>
+		<p class="mt-1 text-sm text-gray-600 dark:text-gray-300">{selectedDatasetLabel}</p>
+		<p class="mt-2 text-sm text-gray-500 dark:text-gray-400" aria-live="polite">
+			{statusText}
+		</p>
+	</header>
+	<DatasetTabs datasetId={data.selectedDataset} active="alerts" />
+
+	{#if fetchError}
+		<div
+			class="rounded-lg border border-red-200 bg-red-50 p-4 text-red-700 dark:border-red-900 dark:bg-red-950 dark:text-red-400"
+			role="alert"
+		>
+			{fetchError}
+		</div>
+	{/if}
+
+	{#if !feedResponse.feed.present}
+		<section
+			class="dark:border-dark-border dark:bg-dark-surface rounded-lg border bg-white p-4 text-gray-600 shadow-sm dark:text-gray-400"
+		>
+			<h2 class="font-medium text-gray-900 dark:text-gray-100">The alert feed is not running</h2>
+			<p class="mt-1 text-sm">Start it for this dataset with:</p>
+			<div
+				class="dark:bg-dark-subtle mt-3 flex flex-col gap-2 rounded-md bg-gray-100 p-3 sm:flex-row sm:items-center sm:justify-between"
+			>
+				<code class="overflow-x-auto font-mono text-sm text-gray-900 dark:text-gray-100"
+					>{feedCommand}</code
+				>
+				<button
+					type="button"
+					onclick={copyCommand}
+					class="shrink-0 rounded bg-blue-600 px-3 py-1 text-sm text-white hover:bg-blue-700 focus:ring-2 focus:ring-blue-500 focus:ring-offset-2 focus:outline-none"
+				>
+					{copied ? 'Copied' : 'Copy'}
+				</button>
+			</div>
+			{#if copyError}
+				<p class="mt-2 text-sm text-red-600 dark:text-red-400" role="alert">{copyError}</p>
+			{/if}
+		</section>
+	{:else}
+		<div class="flex flex-col gap-3 lg:flex-row lg:items-end lg:justify-between">
+			<div class="flex flex-col gap-3 sm:flex-row sm:flex-wrap sm:items-end">
+				<div class="flex flex-col gap-1">
+					<span class="text-xs font-medium text-gray-500 dark:text-gray-400">Alpha</span>
+					<div
+						class={`${CONTROL_GROUP_CLASS} grid-cols-2`}
+						role="group"
+						aria-label="Filter alerts by alpha tail"
+					>
+						{#each TAIL_OPTIONS as option (option.value)}
+							<button
+								type="button"
+								onclick={() => selectTail(option.value)}
+								title={option.title}
+								class={`${CONTROL_BUTTON_CLASS} ${
+									selectedTail === option.value
+										? CONTROL_BUTTON_ACTIVE_CLASS
+										: CONTROL_BUTTON_INACTIVE_CLASS
+								}`}
+								aria-pressed={selectedTail === option.value}
+							>
+								{option.label}
+							</button>
+						{/each}
+					</div>
+				</div>
+
+				<div class="flex flex-col gap-1">
+					<span class="text-xs font-medium text-gray-500 dark:text-gray-400">Horizon</span>
+					<div
+						class={`${CONTROL_GROUP_CLASS} grid-cols-4`}
+						role="group"
+						aria-label="Select alert horizon"
+					>
+						{#each HORIZON_OPTIONS as option (option.value)}
+							<button
+								type="button"
+								onclick={() => selectHorizon(option.value)}
+								class={`${CONTROL_BUTTON_CLASS} ${
+									selectedHorizon === option.value
+										? CONTROL_BUTTON_ACTIVE_CLASS
+										: CONTROL_BUTTON_INACTIVE_CLASS
+								}`}
+								aria-pressed={selectedHorizon === option.value}
+							>
+								{option.label}
+							</button>
+						{/each}
+					</div>
+				</div>
+
+				<div class="flex flex-col gap-1">
+					<span class="text-xs font-medium text-gray-500 dark:text-gray-400">Sort</span>
+					<div
+						class={`${CONTROL_GROUP_CLASS} grid-cols-2`}
+						role="group"
+						aria-label="Sort alert addresses"
+					>
+						{#each SORT_OPTIONS as option (option.value)}
+							<button
+								type="button"
+								onclick={() => selectSort(option.value)}
+								class={`${CONTROL_BUTTON_CLASS} ${
+									selectedSort === option.value
+										? CONTROL_BUTTON_ACTIVE_CLASS
+										: CONTROL_BUTTON_INACTIVE_CLASS
+								}`}
+								aria-pressed={selectedSort === option.value}
+							>
+								{option.label}
+							</button>
+						{/each}
+					</div>
+				</div>
+
+				{#if feedResponse.feed.present && feedResponse.feed.thresholds}
+					<div class="flex flex-col gap-1">
+						<span class="text-xs font-medium text-gray-500 dark:text-gray-400">Thresholds</span>
+						<p
+							class="flex min-h-7 items-center pb-1 text-sm text-gray-500 tabular-nums dark:text-gray-400"
+							title="The feed records an address when its alpha crosses either calibrated bound"
+						>
+							α ≥ {feedResponse.feed.thresholds.high} · α ≤ {feedResponse.feed.thresholds.low}
+						</p>
+					</div>
+				{/if}
+			</div>
+		</div>
+
+		{#if feedResponse.addresses.length === 0}
+			<div
+				class="dark:border-dark-border dark:bg-dark-surface rounded-lg border bg-white px-4 py-3 text-sm text-gray-500 shadow-sm dark:text-gray-400"
+			>
+				No anomalous addresses in the last {selectedHorizon}.
+			</div>
+		{:else}
+			<section
+				class={`dark:border-dark-border dark:bg-dark-surface overflow-x-auto rounded-lg border bg-white shadow-sm transition-opacity ${loading ? 'opacity-60' : ''}`}
+				aria-label="Anomalous addresses"
+				aria-busy={loading}
+			>
+				<table class="w-full text-sm">
+					<thead>
+						<tr
+							class="dark:border-dark-border border-b text-xs font-medium text-gray-500 dark:text-gray-400"
+						>
+							<th class="w-full px-4 py-2 text-left font-medium">Address</th>
+							<th
+								class={`px-4 py-2 text-right font-medium whitespace-nowrap ${
+									selectedSort === 'extreme' ? 'text-gray-900 dark:text-gray-100' : ''
+								}`}
+							>
+								Peak α
+							</th>
+							<th
+								class={`px-4 py-2 text-right font-medium whitespace-nowrap ${
+									selectedSort === 'recent' ? 'text-gray-900 dark:text-gray-100' : ''
+								}`}
+							>
+								Latest α
+							</th>
+							<th class="px-4 py-2 text-right font-medium">First seen</th>
+							<th class="px-4 py-2 text-right font-medium">Last seen</th>
+							<th class="px-4 py-2 text-right font-medium" title="Windows flagged in this horizon">
+								Flagged
+							</th>
+							<th class="px-4 py-2 text-right font-medium">r²</th>
+						</tr>
+					</thead>
+					<tbody class="dark:divide-dark-border divide-y divide-gray-100">
+						{#each feedResponse.addresses as alert (alert.address)}
+							<tr>
+								<td class="px-4 py-3">
+									<div class="flex min-w-0 items-center gap-2">
+										<span class="truncate font-mono text-gray-900 dark:text-gray-100">
+											{alert.address}
+										</span>
+										{#if isNewAddress(alert.firstSeen)}
+											<span
+												title={`First flagged ${formatRelativeTime(alert.firstSeen)} — never seen in the retained history before that`}
+												class="shrink-0 rounded-full bg-amber-100 px-2 py-0.5 text-xs font-medium text-amber-700 dark:bg-amber-950 dark:text-amber-300"
+											>
+												new
+											</span>
+										{/if}
+									</div>
+								</td>
+								<td
+									class={`px-4 py-3 text-right tabular-nums ${
+										selectedSort === 'extreme'
+											? 'font-medium text-gray-900 dark:text-gray-100'
+											: 'text-gray-400 dark:text-gray-500'
+									}`}
+								>
+									{alert.peakAlpha.toFixed(3)}
+								</td>
+								<td
+									class={`px-4 py-3 text-right tabular-nums ${
+										selectedSort === 'recent'
+											? 'font-medium text-gray-900 dark:text-gray-100'
+											: 'text-gray-400 dark:text-gray-500'
+									}`}
+								>
+									{alert.latestAlpha.toFixed(3)}
+								</td>
+								<td
+									title={dateTimeFormatter.format(new Date(alert.firstSeen * 1000))}
+									class="px-4 py-3 text-right whitespace-nowrap text-gray-500 tabular-nums dark:text-gray-400"
+								>
+									{formatRelativeTime(alert.firstSeen)}
+								</td>
+								<td
+									title={dateTimeFormatter.format(new Date(alert.lastSeen * 1000))}
+									class="px-4 py-3 text-right whitespace-nowrap text-gray-500 tabular-nums dark:text-gray-400"
+								>
+									{formatRelativeTime(alert.lastSeen)}
+								</td>
+								<td
+									title={`flagged in ${countFormatter.format(alert.timesFlagged)} windows in this horizon`}
+									class="px-4 py-3 text-right text-gray-600 tabular-nums dark:text-gray-300"
+								>
+									{countFormatter.format(alert.timesFlagged)}×
+								</td>
+								<td class="px-4 py-3 text-right text-gray-500 tabular-nums dark:text-gray-400">
+									{alert.peakR2.toFixed(2)}
+								</td>
+							</tr>
+						{/each}
+					</tbody>
+				</table>
+			</section>
+		{/if}
+
+		<div class="flex flex-col items-center gap-2 pt-1">
+			{#if canShowMore}
+				<button
+					type="button"
+					onclick={showMore}
+					disabled={loadingMore}
+					class="dark:border-dark-border dark:bg-dark-surface rounded-md border border-gray-300 bg-white px-4 py-2 text-sm font-medium text-gray-700 hover:bg-gray-50 focus:ring-2 focus:ring-blue-500 focus:outline-none disabled:cursor-not-allowed disabled:opacity-60 dark:text-gray-300 dark:hover:bg-gray-800"
+				>
+					{loadingMore ? 'Loading…' : 'Show more'}
+				</button>
+			{/if}
+			<p class="text-xs text-gray-500 dark:text-gray-400">
+				showing {countFormatter.format(feedResponse.addresses.length)} of {countFormatter.format(
+					feedResponse.totalAddresses
+				)}
+			</p>
+		</div>
+	{/if}
+</main>
diff --git a/apps/web/src/routes/datasets/[dataset]/alerts/+page.ts b/apps/web/src/routes/datasets/[dataset]/alerts/+page.ts
new file mode 100644
index 00000000..0389df5f
--- /dev/null
+++ b/apps/web/src/routes/datasets/[dataset]/alerts/+page.ts
@@ -0,0 +1,35 @@
+import { error } from '@sveltejs/kit';
+import type { PageLoad } from './$types';
+import { loadDatasetSummariesFromFetch } from '$lib/datasets';
+import type { AlertsFeedResponse } from '$lib/types/types';
+
+type ErrorResponse = {
+	data: null;
+	error: string;
+};
+
+export const load: PageLoad = async ({ fetch, params }) => {
+	const datasets = await loadDatasetSummariesFromFetch(fetch);
+	const selectedDataset = params.dataset;
+	if (!datasets.some((dataset) => dataset.datasetId === selectedDataset)) {
+		throw error(404, `Unknown dataset '${selectedDataset}'`);
+	}
+
+	try {
+		const response = await fetch(
+			`/api/alerts?dataset=${encodeURIComponent(selectedDataset)}&tail=high&horizon=24h&sort=extreme&limit=100`
+		);
+		const alerts = (await response.json()) as AlertsFeedResponse | ErrorResponse;
+		if (!response.ok || 'error' in alerts) {
+			throw new Error('error' in alerts ? alerts.error : 'Failed to load alerts feed');
+		}
+
+		return {
+			datasets,
+			selectedDataset,
+			alerts
+		};
+	} catch (err) {
+		throw error(500, err instanceof Error ? err.message : 'Failed to load alerts feed');
+	}
+};
diff --git a/apps/web/tests/lib/server/alerts.test.ts b/apps/web/tests/lib/server/alerts.test.ts
new file mode 100644
index 00000000..6eeb6a31
--- /dev/null
+++ b/apps/web/tests/lib/server/alerts.test.ts
@@ -0,0 +1,327 @@
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import Database from 'better-sqlite3';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+const ALERT_SCHEMA = `
+	CREATE TABLE feed_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+	CREATE TABLE windows (
+		window_start INTEGER PRIMARY KEY,
+		window_end INTEGER NOT NULL,
+		member_files INTEGER NOT NULL,
+		address_count INTEGER NOT NULL,
+		alert_count INTEGER NOT NULL,
+		alpha_min REAL,
+		alpha_max REAL,
+		alpha_median REAL,
+		processed_at INTEGER NOT NULL
+	);
+	CREATE TABLE alerts (
+		window_start INTEGER NOT NULL REFERENCES windows(window_start) ON DELETE CASCADE,
+		address TEXT NOT NULL,
+		alpha REAL NOT NULL,
+		tail TEXT NOT NULL CHECK (tail IN ('high', 'low')),
+		rank INTEGER NOT NULL,
+		r2 REAL NOT NULL,
+		prefix_levels INTEGER NOT NULL,
+		PRIMARY KEY (window_start, tail, rank)
+	);
+	CREATE INDEX alerts_address ON alerts(address, window_start);
+`;
+
+type Fixture = {
+	directory: string;
+	netflowPath: string;
+	alertsPath: string;
+};
+
+async function loadAlertsModule() {
+	vi.resetModules();
+	return import('../../../src/lib/server/alerts');
+}
+
+function createDatasetFixture(datasetId = 'alpha'): Fixture {
+	const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'alerts-test-'));
+	const netflowPath = path.join(directory, 'netflow.sqlite');
+	const db = new Database(netflowPath);
+	db.exec(`
+		CREATE TABLE datasets (
+			id TEXT PRIMARY KEY NOT NULL,
+			label TEXT NOT NULL,
+			default_start_date TEXT NOT NULL,
+			source_mode TEXT DEFAULT 'static' NOT NULL,
+			discovery_mode TEXT DEFAULT 'static' NOT NULL,
+			sort_order INTEGER DEFAULT 0 NOT NULL
+		);
+	`);
+	db.prepare(
+		`INSERT INTO datasets (
+			id, label, default_start_date, source_mode, discovery_mode, sort_order
+		) VALUES (?, ?, '2025-03-01', 'static', 'live', 0)`
+	).run(datasetId, 'Alpha Label');
+	db.close();
+
+	return {
+		directory,
+		netflowPath,
+		alertsPath: path.join(directory, 'alerts.sqlite')
+	};
+}
+
+function seedAlertsDatabase(fixture: Fixture, windowCount = 2): void {
+	const db = new Database(fixture.alertsPath);
+	db.exec(ALERT_SCHEMA);
+	const insertMeta = db.prepare('INSERT INTO feed_meta (key, value) VALUES (?, ?)');
+	for (const [key, value] of [
+		['schema_version', '1'],
+		['dataset_id', 'alpha'],
+		['threshold_high', '3.5'],
+		['threshold_low', '0.4'],
+		['max_per_tail', '25']
+	] as const) {
+		insertMeta.run(key, value);
+	}
+
+	const insertWindow = db.prepare(`
+		INSERT INTO windows (
+			window_start,
+			window_end,
+			member_files,
+			address_count,
+			alert_count,
+			alpha_min,
+			alpha_max,
+			alpha_median,
+			processed_at
+		) VALUES (?, ?, 3, ?, ?, NULL, NULL, NULL, ?)
+	`);
+	const seedWindows = db.transaction(() => {
+		for (let index = 0; index < windowCount; index += 1) {
+			const windowStart = 1_700_000_000 + index * 300;
+			const isLatestFixtureWindow = index === 1;
+			insertWindow.run(
+				windowStart,
+				windowStart + 300,
+				48_000 + index,
+				isLatestFixtureWindow ? 3 : index === 0 ? 1 : 0,
+				windowStart + 320
+			);
+		}
+	});
+	seedWindows();
+
+	if (windowCount >= 1) {
+		db.prepare(
+			`INSERT INTO alerts (
+				window_start, address, alpha, tail, rank, r2, prefix_levels
+			) VALUES (?, '9.9.9.9', 0.21, 'low', 1, 0.91, 24)`
+		).run(1_700_000_000);
+	}
+	if (windowCount >= 2) {
+		const insertAlert = db.prepare(`
+			INSERT INTO alerts (
+				window_start, address, alpha, tail, rank, r2, prefix_levels
+			) VALUES (?, ?, ?, ?, ?, ?, 24)
+		`);
+		const latestWindowStart = 1_700_000_300;
+		insertAlert.run(latestWindowStart, '1.1.1.2', 3.7, 'high', 2, 0.94);
+		insertAlert.run(latestWindowStart, '2.2.2.2', 0.2, 'low', 1, 0.89);
+		insertAlert.run(latestWindowStart, '1.1.1.1', 3.9, 'high', 1, 0.98);
+	}
+	db.close();
+}
+
+describe('alerts server helper', () => {
+	const originalCwd = process.cwd();
+
+	afterEach(() => {
+		process.chdir(originalCwd);
+		vi.unstubAllEnvs();
+	});
+
+	it('returns feed metadata and severity-ordered address aggregates', async () => {
+		const fixture = createDatasetFixture();
+		seedAlertsDatabase(fixture);
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+		const alerts = await loadAlertsModule();
+
+		await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({
+			feed: {
+				present: true,
+				latestWindowStart: 1_700_000_300,
+				latestWindowEnd: 1_700_000_600,
+				latestAddressCount: 48_001,
+				latestProcessedAt: 1_700_000_620,
+				thresholds: { high: 3.5, low: 0.4 }
+			},
+			horizonSeconds: 86_400,
+			totalAddresses: 4,
+			addresses: [
+				{
+					address: '1.1.1.1',
+					tail: 'high',
+					peakAlpha: 3.9,
+					peakWindowStart: 1_700_000_300,
+					peakR2: 0.98,
+					latestAlpha: 3.9,
+					lastSeen: 1_700_000_300,
+					firstSeen: 1_700_000_300,
+					timesFlagged: 1
+				},
+				{
+					address: '1.1.1.2',
+					tail: 'high',
+					peakAlpha: 3.7,
+					peakWindowStart: 1_700_000_300,
+					peakR2: 0.94,
+					latestAlpha: 3.7,
+					lastSeen: 1_700_000_300,
+					firstSeen: 1_700_000_300,
+					timesFlagged: 1
+				},
+				{
+					address: '2.2.2.2',
+					tail: 'low',
+					peakAlpha: 0.2,
+					peakWindowStart: 1_700_000_300,
+					peakR2: 0.89,
+					latestAlpha: 0.2,
+					lastSeen: 1_700_000_300,
+					firstSeen: 1_700_000_300,
+					timesFlagged: 1
+				},
+				{
+					address: '9.9.9.9',
+					tail: 'low',
+					peakAlpha: 0.21,
+					peakWindowStart: 1_700_000_000,
+					peakR2: 0.91,
+					latestAlpha: 0.21,
+					lastSeen: 1_700_000_000,
+					firstSeen: 1_700_000_000,
+					timesFlagged: 1
+				}
+			]
+		});
+	});
+
+	it('returns absent when a discovered dataset has no alerts database', async () => {
+		const workspace = fs.mkdtempSync(path.join(os.tmpdir(), 'alerts-discovery-'));
+		const datasetDirectory = path.join(workspace, 'data', 'alpha');
+		fs.mkdirSync(datasetDirectory, { recursive: true });
+		const fixture = createDatasetFixture();
+		fs.renameSync(fixture.netflowPath, path.join(datasetDirectory, 'netflow.sqlite'));
+		process.chdir(workspace);
+		const alerts = await loadAlertsModule();
+
+		await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({
+			feed: { present: false },
+			horizonSeconds: 86_400,
+			totalAddresses: 0,
+			addresses: []
+		});
+	});
+
+	it('does not throw when a configured dataset database has no sibling alerts database', async () => {
+		const fixture = createDatasetFixture();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+		const alerts = await loadAlertsModule();
+
+		await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({
+			feed: { present: false },
+			horizonSeconds: 86_400,
+			totalAddresses: 0,
+			addresses: []
+		});
+	});
+
+	it('opens a feed file that appears after an earlier absent result', async () => {
+		const fixture = createDatasetFixture();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+		const alerts = await loadAlertsModule();
+
+		await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toMatchObject({
+			feed: { present: false }
+		});
+		seedAlertsDatabase(fixture);
+		await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toMatchObject({
+			feed: { present: true },
+			totalAddresses: 4,
+			addresses: expect.arrayContaining([
+				expect.objectContaining({ address: '1.1.1.1' }),
+				expect.objectContaining({ address: '9.9.9.9' })
+			])
+		});
+	});
+
+	it('evicts a cached feed handle when its file disappears', async () => {
+		const fixture = createDatasetFixture();
+		seedAlertsDatabase(fixture);
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+		const alerts = await loadAlertsModule();
+
+		await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toMatchObject({
+			feed: { present: true }
+		});
+		fs.unlinkSync(fixture.alertsPath);
+		await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({
+			feed: { present: false },
+			horizonSeconds: 86_400,
+			totalAddresses: 0,
+			addresses: []
+		});
+	});
+
+	it('filters rows by tail before returning address aggregates', async () => {
+		const fixture = createDatasetFixture();
+		seedAlertsDatabase(fixture);
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+		const alerts = await loadAlertsModule();
+
+		const result = await alerts.getAlertsFeedForDataset('alpha', { tail: 'high' });
+		expect(result.totalAddresses).toBe(2);
+		expect(result.addresses).toMatchObject([
+			{ address: '1.1.1.1', tail: 'high' },
+			{ address: '1.1.1.2', tail: 'high' }
+		]);
+	});
+
+	it('sorts recent addresses by last seen before severity', async () => {
+		const fixture = createDatasetFixture();
+		seedAlertsDatabase(fixture);
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+		const alerts = await loadAlertsModule();
+
+		const result = await alerts.getAlertsFeedForDataset('alpha', { sort: 'recent' });
+		expect(result.addresses.map(({ address }) => address)).toEqual([
+			'1.1.1.1',
+			'1.1.1.2',
+			'2.2.2.2',
+			'9.9.9.9'
+		]);
+	});
+
+	it('clamps address limits to the inclusive range from 1 through 500', async () => {
+		const fixture = createDatasetFixture();
+		seedAlertsDatabase(fixture, 600);
+		const db = new Database(fixture.alertsPath);
+		const insertAlert = db.prepare(`
+			INSERT INTO alerts (
+				window_start, address, alpha, tail, rank, r2, prefix_levels
+			) VALUES (?, ?, 3.6, 'high', 1, 0.8, 24)
+		`);
+		for (let index = 2; index < 600; index += 1) {
+			insertAlert.run(1_700_000_000 + index * 300, `address-${index}`);
+		}
+		db.close();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+		const alerts = await loadAlertsModule();
+
+		const lower = await alerts.getAlertsFeedForDataset('alpha', { horizon: '7d', limit: 0 });
+		const upper = await alerts.getAlertsFeedForDataset('alpha', { horizon: '7d', limit: 999 });
+		expect(lower.addresses).toHaveLength(1);
+		expect(upper.addresses).toHaveLength(500);
+		expect(upper.totalAddresses).toBe(602);
+	});
+});
diff --git a/apps/web/tests/routes/api-alerts.test.ts b/apps/web/tests/routes/api-alerts.test.ts
new file mode 100644
index 00000000..95868a4e
--- /dev/null
+++ b/apps/web/tests/routes/api-alerts.test.ts
@@ -0,0 +1,326 @@
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import Database from 'better-sqlite3';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { getRequestedDataset } from '$lib/server/datasets';
+import { GET } from '../../src/routes/api/alerts/+server';
+
+vi.mock('$lib/server/datasets', () => ({
+	getRequestedDataset: vi.fn()
+}));
+
+const LATEST_WINDOW_START = 200_000;
+const LATEST_WINDOW_END = 200_300;
+const ALERT_SCHEMA = `
+	CREATE TABLE feed_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+	CREATE TABLE windows (
+		window_start INTEGER PRIMARY KEY,
+		window_end INTEGER NOT NULL,
+		member_files INTEGER NOT NULL,
+		address_count INTEGER NOT NULL,
+		alert_count INTEGER NOT NULL,
+		alpha_min REAL,
+		alpha_max REAL,
+		alpha_median REAL,
+		processed_at INTEGER NOT NULL
+	);
+	CREATE TABLE alerts (
+		window_start INTEGER NOT NULL REFERENCES windows(window_start) ON DELETE CASCADE,
+		address TEXT NOT NULL,
+		alpha REAL NOT NULL,
+		tail TEXT NOT NULL CHECK (tail IN ('high', 'low')),
+		rank INTEGER NOT NULL,
+		r2 REAL NOT NULL,
+		prefix_levels INTEGER NOT NULL,
+		PRIMARY KEY (window_start, tail, rank)
+	);
+	CREATE INDEX alerts_address ON alerts(address, window_start);
+`;
+
+type Fixture = {
+	directory: string;
+	netflowPath: string;
+	alertsPath: string;
+};
+
+const fixtureDirectories: string[] = [];
+
+function createFixture(withAlerts = true): Fixture {
+	const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'api-alerts-test-'));
+	fixtureDirectories.push(directory);
+	const netflowPath = path.join(directory, 'netflow.sqlite');
+	const db = new Database(netflowPath);
+	db.exec(`
+		CREATE TABLE datasets (
+			id TEXT PRIMARY KEY NOT NULL,
+			label TEXT NOT NULL,
+			default_start_date TEXT NOT NULL,
+			source_mode TEXT DEFAULT 'static' NOT NULL,
+			discovery_mode TEXT DEFAULT 'static' NOT NULL,
+			sort_order INTEGER DEFAULT 0 NOT NULL
+		);
+	`);
+	db.prepare(
+		`INSERT INTO datasets (
+			id, label, default_start_date, source_mode, discovery_mode, sort_order
+		) VALUES ('alpha', 'Alpha Label', '2025-03-01', 'static', 'live', 0)`
+	).run();
+	db.close();
+
+	const fixture = {
+		directory,
+		netflowPath,
+		alertsPath: path.join(directory, 'alerts.sqlite')
+	};
+	if (withAlerts) {
+		seedAlerts(fixture);
+	}
+	return fixture;
+}
+
+function seedAlerts(fixture: Fixture): void {
+	const db = new Database(fixture.alertsPath);
+	db.exec(ALERT_SCHEMA);
+	const insertMeta = db.prepare('INSERT INTO feed_meta (key, value) VALUES (?, ?)');
+	for (const [key, value] of [
+		['schema_version', '1'],
+		['dataset_id', 'alpha'],
+		['threshold_high', '2.0'],
+		['threshold_low', '0.3'],
+		['max_per_tail', '25']
+	] as const) {
+		insertMeta.run(key, value);
+	}
+
+	const insertWindow = db.prepare(`
+		INSERT INTO windows (
+			window_start,
+			window_end,
+			member_files,
+			address_count,
+			alert_count,
+			alpha_min,
+			alpha_max,
+			alpha_median,
+			processed_at
+		) VALUES (?, ?, 1, 1000, 0, NULL, NULL, NULL, ?)
+	`);
+	for (const windowStart of [113_899, 196_600, 196_700, 199_700, LATEST_WINDOW_START]) {
+		const windowEnd = windowStart === LATEST_WINDOW_START ? LATEST_WINDOW_END : windowStart + 300;
+		insertWindow.run(windowStart, windowEnd, windowEnd + 20);
+	}
+
+	const insertAlert = db.prepare(`
+		INSERT INTO alerts (
+			window_start, address, alpha, tail, rank, r2, prefix_levels
+		) VALUES (?, ?, ?, ?, ?, ?, 24)
+	`);
+	insertAlert.run(113_899, 'outside-24h', 12, 'high', 1, 0.5);
+	// history for high-address outside every horizon: firstSeen must be
+	// retention-wide (113_899) while in-horizon aggregates ignore this row.
+	insertAlert.run(113_899, 'high-address', 2.05, 'high', 2, 0.5);
+	insertAlert.run(196_600, 'old-high', 6, 'high', 1, 0.6);
+	insertAlert.run(196_700, 'repeat', 2.5, 'high', 1, 0.7);
+	insertAlert.run(199_700, 'repeat', 4.5, 'high', 1, 0.91);
+	insertAlert.run(199_700, 'cross-tail', 2.1, 'high', 2, 0.65);
+	insertAlert.run(199_700, 'low-address', 0, 'low', 1, 0.87);
+	insertAlert.run(LATEST_WINDOW_START, 'repeat', 2.2, 'high', 1, 0.8);
+	insertAlert.run(LATEST_WINDOW_START, 'high-address', 2.8, 'high', 2, 0.93);
+	insertAlert.run(LATEST_WINDOW_START, 'cross-tail', 0.1, 'low', 1, 0.95);
+	db.close();
+}
+
+function eventFor(query = '') {
+	return {
+		url: new URL(`http://localhost/api/alerts${query}`),
+		platform: undefined
+	} as never;
+}
+
+async function getJson(query = ''): Promise<{ response: Response; payload: unknown }> {
+	const response = await GET(eventFor(query));
+	return { response, payload: await response.json() };
+}
+
+describe('/api/alerts GET', () => {
+	beforeEach(() => {
+		vi.mocked(getRequestedDataset).mockReset().mockResolvedValue('alpha');
+	});
+
+	afterEach(() => {
+		vi.unstubAllEnvs();
+		for (const directory of fixtureDirectories.splice(0)) {
+			fs.rmSync(directory, { recursive: true, force: true });
+		}
+	});
+
+	it('deduplicates addresses and selects the most severe row inside the horizon', async () => {
+		const fixture = createFixture();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+
+		const { response, payload } = await getJson('?dataset=alpha&horizon=1h');
+
+		expect(response.status).toBe(200);
+		expect(payload).toEqual({
+			feed: {
+				present: true,
+				latestWindowStart: LATEST_WINDOW_START,
+				latestWindowEnd: LATEST_WINDOW_END,
+				latestAddressCount: 1000,
+				latestProcessedAt: LATEST_WINDOW_END + 20,
+				thresholds: { high: 2, low: 0.3 }
+			},
+			horizonSeconds: 3600,
+			totalAddresses: 4,
+			addresses: [
+				{
+					address: 'repeat',
+					tail: 'high',
+					peakAlpha: 4.5,
+					peakWindowStart: 199_700,
+					peakR2: 0.91,
+					latestAlpha: 2.2,
+					lastSeen: LATEST_WINDOW_START,
+					firstSeen: 196_700,
+					timesFlagged: 3
+				},
+				{
+					address: 'high-address',
+					tail: 'high',
+					peakAlpha: 2.8,
+					peakWindowStart: LATEST_WINDOW_START,
+					peakR2: 0.93,
+					latestAlpha: 2.8,
+					lastSeen: LATEST_WINDOW_START,
+					// retention-wide, not horizon-scoped: the 113_899 history row
+					firstSeen: 113_899,
+					timesFlagged: 1
+				},
+				{
+					address: 'low-address',
+					tail: 'low',
+					peakAlpha: 0,
+					peakWindowStart: 199_700,
+					peakR2: 0.87,
+					latestAlpha: 0,
+					lastSeen: 199_700,
+					firstSeen: 199_700,
+					timesFlagged: 1
+				},
+				{
+					address: 'cross-tail',
+					tail: 'low',
+					peakAlpha: 0.1,
+					peakWindowStart: LATEST_WINDOW_START,
+					peakR2: 0.95,
+					latestAlpha: 0.1,
+					lastSeen: LATEST_WINDOW_START,
+					firstSeen: 199_700,
+					timesFlagged: 2
+				}
+			]
+		});
+	});
+
+	it('sorts by last seen and uses peak severity as the tiebreak', async () => {
+		const fixture = createFixture();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+
+		const { payload } = await getJson('?sort=recent&horizon=1h');
+
+		expect(payload).toMatchObject({ horizonSeconds: 3600, totalAddresses: 4 });
+		expect(
+			(payload as { addresses: Array<{ address: string }> }).addresses.map(({ address }) => address)
+		).toEqual(['repeat', 'high-address', 'cross-tail', 'low-address']);
+	});
+
+	it('anchors horizon filtering on the latest processed window end', async () => {
+		const fixture = createFixture();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+
+		const oneHour = await getJson('?horizon=1h');
+		const oneDay = await getJson('?horizon=24h');
+
+		expect((oneHour.payload as { addresses: Array<{ address: string }> }).addresses).toEqual(
+			expect.arrayContaining([expect.objectContaining({ address: 'repeat', firstSeen: 196_700 })])
+		);
+		expect(
+			(oneHour.payload as { addresses: Array<{ address: string }> }).addresses.some(
+				({ address }) => address === 'old-high'
+			)
+		).toBe(false);
+		expect(
+			(oneDay.payload as { addresses: Array<{ address: string }> }).addresses.map(
+				({ address }) => address
+			)
+		).toContain('old-high');
+		expect(
+			(oneDay.payload as { addresses: Array<{ address: string }> }).addresses.map(
+				({ address }) => address
+			)
+		).not.toContain('outside-24h');
+	});
+
+	it('filters rows by tail before aggregating addresses', async () => {
+		const fixture = createFixture();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+
+		const { payload } = await getJson('?tail=high&horizon=1h');
+		const result = payload as {
+			totalAddresses: number;
+			addresses: Array<{ address: string; tail: string; timesFlagged: number; peakAlpha: number }>;
+		};
+
+		expect(result.totalAddresses).toBe(3);
+		expect(result.addresses).toEqual(
+			expect.arrayContaining([
+				expect.objectContaining({
+					address: 'cross-tail',
+					tail: 'high',
+					peakAlpha: 2.1,
+					timesFlagged: 1
+				})
+			])
+		);
+		expect(result.addresses.every(({ tail }) => tail === 'high')).toBe(true);
+	});
+
+	it('applies the limit after counting all matching addresses', async () => {
+		const fixture = createFixture();
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+
+		const { payload } = await getJson('?horizon=24h&limit=2');
+		const result = payload as { totalAddresses: number; addresses: unknown[] };
+
+		expect(result.totalAddresses).toBe(5);
+		expect(result.addresses).toHaveLength(2);
+	});
+
+	it.each([
+		['?tail=middle', 'Invalid tail parameter'],
+		['?horizon=2h', 'Invalid horizon parameter'],
+		['?sort=alpha', 'Invalid sort parameter'],
+		['?limit=many', 'Invalid limit parameter']
+	])('rejects invalid query %s', async (query, message) => {
+		const { response, payload } = await getJson(query);
+
+		expect(response.status).toBe(400);
+		expect(payload).toEqual({ data: null, error: message });
+	});
+
+	it('returns an absent feed as a normal response', async () => {
+		const fixture = createFixture(false);
+		vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath);
+
+		const { response, payload } = await getJson('?horizon=7d');
+
+		expect(response.status).toBe(200);
+		expect(payload).toEqual({
+			feed: { present: false },
+			horizonSeconds: 604_800,
+			totalAddresses: 0,
+			addresses: []
+		});
+	});
+});
diff --git a/docs/agent/singularity-calibration.md b/docs/agent/singularity-calibration.md
new file mode 100644
index 00000000..8cc0c5db
--- /dev/null
+++ b/docs/agent/singularity-calibration.md
@@ -0,0 +1,227 @@
+# Rust Singularity conformance and threshold calibration
+
+## Summary
+
+I compared the Rust port with the prebuilt Haskell reference on 24 real five-minute windows and calibrated the Rust alert feed on 200 separate windows. The conformance verdict is PASS. Every compared address and `n_levels` value matched, and no numeric value violated the requested tolerance.
+
+The calibration supports `threshold_high = 2` and `threshold_low = 0.3`. At the feed's 20-per-tail cap, this pair produces a mean of 19.7 and median of 20.0 recorded alerts per window. The middle 80% spans 14.0 to 25.0.
+
+## Environment and extraction contract
+
+The Rust binary was `netflow-db 0.1.0`, built from commit `a30a51f41793ec88c99e07bf4bb2aa8078345192` with the repository's Rust 1.97.1 toolchain. Its SHA-256 was `a3258b36c29021a6056cdbb53633313d3200884ce8a36e61b62f8ec6e9f1cac6`. The prebuilt Haskell binary SHA-256 was `ea5b0ccca94355cafbebe6f3c2a9ca3e8441b2cb11406a93b8f67e4971f81d69`; I did not rebuild it. Capture reads used nfdump 1.7.6-release on Linux 6.18.35 x86_64.
+
+For each timestamp, the extractor reads both `cc_ir1_gw` and `oh_ir1_gw` files with `nfdump -q -o 'csv:%sa,%da'`, rejects fields containing `:`, combines source and destination addresses, and runs one locale-fixed external `sort -u`. The standard CSV probe reported `srcAddr` and `dstAddr` as columns 4 and 6. A 10,000-record validation found 5,149 unique IPv4 addresses in both the standard ten-column output and the custom two-column projection, with byte-identical sorted lists.
+
+The capture inventory contained 111,272 paired timestamps from 2025-06-01 00:00 through 2026-06-30 13:15. The selector excluded one cc-only and two oh-only timestamps. Filename hours are treated as America/Los_Angeles local time, matching the feed's configured timezone.
+
+## Part 1: Rust versus Haskell conformance
+
+### Method
+
+The deterministic selector chose 24 windows across all 13 available months. I kept all 24 candidates, instead of stopping at 20, so May and June 2026 remained represented. It used 03:00, 09:00, 15:00, and 21:00 hours six times each, with 16 weekday and 8 weekend windows. Both implementations received the exact same sorted address file. I joined results by address, compared `n_levels` with exact integer equality, and applied `abs(a-b) <= max(1e-9 * max(abs(a),abs(b)), 1e-12)` to alpha, intercept, and r2.
+
+The deviation columns below report `abs(a-b) / max(abs(a),abs(b),1e-12)`. Near zero, that display value can exceed 1e-9 while the absolute 1e-12 tolerance still passes. The A/I/R violations column removes that ambiguity.
+
+| Timestamp    | Addresses | Max rel alpha | Max rel intercept | Max rel r2 | A/I/R violations | Level mismatches | Rust/Haskell only |
+| ------------ | --------- | ------------- | ----------------- | ---------- | ---------------- | ---------------- | ----------------- |
+| 202506080355 | 296,779   | 7.270e-15     | 1.474e-10         | 6.947e-15  | 0/0/0            | 0                | 0/0               |
+| 202506250900 | 309,275   | 3.911e-15     | 4.524e-09         | 6.901e-15  | 0/0/0            | 0                | 0/0               |
+| 202507111555 | 299,657   | 3.817e-15     | 5.471e-10         | 6.190e-15  | 0/0/0            | 0                | 0/0               |
+| 202507272155 | 294,684   | 4.441e-15     | 6.094e-11         | 1.168e-14  | 0/0/0            | 0                | 0/0               |
+| 202508140300 | 302,023   | 5.717e-15     | 1.033e-10         | 8.630e-15  | 0/0/0            | 0                | 0/0               |
+| 202508290955 | 301,098   | 4.400e-15     | 1.066e-10         | 6.494e-15  | 0/0/0            | 0                | 0/0               |
+| 202509141555 | 291,695   | 4.422e-15     | 2.007e-10         | 7.696e-15  | 0/0/0            | 0                | 0/0               |
+| 202510022100 | 330,546   | 4.318e-15     | 1.182e-09         | 7.520e-15  | 0/0/0            | 0                | 0/0               |
+| 202510170355 | 294,497   | 7.737e-15     | 2.417e-10         | 8.917e-15  | 0/0/0            | 0                | 0/0               |
+| 202511020955 | 309,703   | 4.559e-15     | 7.636e-11         | 7.037e-15  | 0/0/0            | 0                | 0/0               |
+| 202511171555 | 316,316   | 7.450e-15     | 1.004e-10         | 7.095e-15  | 0/0/0            | 0                | 0/0               |
+| 202512082100 | 318,123   | 5.220e-15     | 7.459e-10         | 7.741e-15  | 0/0/0            | 0                | 0/0               |
+| 202512210355 | 290,362   | 4.610e-15     | 2.848e-09         | 8.392e-15  | 0/0/0            | 0                | 0/0               |
+| 202601090900 | 373,695   | 4.797e-15     | 1.187e-09         | 8.681e-15  | 0/0/0            | 0                | 0/0               |
+| 202601261500 | 321,101   | 5.734e-15     | 1.404e-08         | 7.690e-15  | 0/0/0            | 0                | 0/0               |
+| 202602082155 | 305,401   | 6.569e-15     | 1.324e-10         | 7.569e-15  | 0/0/0            | 0                | 0/0               |
+| 202602270300 | 311,010   | 5.386e-15     | 1.380e-10         | 9.091e-15  | 0/0/0            | 0                | 0/0               |
+| 202603160900 | 335,478   | 4.635e-15     | 1.410e-10         | 6.593e-15  | 0/0/0            | 0                | 0/0               |
+| 202603291555 | 329,895   | 5.980e-15     | 5.910e-11         | 8.342e-15  | 0/0/0            | 0                | 0/0               |
+| 202604172100 | 343,949   | 4.615e-15     | 1.824e-10         | 7.257e-15  | 0/0/0            | 0                | 0/0               |
+| 202605040300 | 348,968   | 4.265e-15     | 2.754e-10         | 6.557e-15  | 0/0/0            | 0                | 0/0               |
+| 202605170955 | 346,383   | 7.401e-15     | 1.073e-08         | 7.783e-15  | 0/0/0            | 0                | 0/0               |
+| 202606051555 | 338,304   | 5.021e-15     | 1.737e-10         | 7.039e-15  | 0/0/0            | 0                | 0/0               |
+| 202606222100 | 324,410   | 4.557e-15     | 1.269e-10         | 7.463e-15  | 0/0/0            | 0                | 0/0               |
+
+Overall maxima were 7.737e-15 for alpha, 1.404e-08 for intercept, and 1.168e-14 for r2. The total mismatch or tolerance-violation count was 0. The largest tested window was 202601090900 with 373,695 addresses.
+
+The largest displayed intercept ratio came from two near-zero values at 72.5.189.239: Rust -1.642975968608e-07, Haskell -1.642975945543e-07. Their absolute difference was 2.306e-15, versus an allowed 1.000e-12. This is the expected absolute-tolerance case, not a numerical failure.
+
+### Haskell wall-clock budget
+
+No window was skipped, replaced, sampled down, or timed out. The firm budget was 600 seconds per Haskell run. Observed Haskell time had a median of 11.7 seconds and a maximum of 16.0 seconds. The 299,472-address format probe took 9.1 seconds in Haskell and 3.7 seconds in Rust, so full-window checks were comfortably tractable.
+
+## Part 2: Rust threshold calibration
+
+### Sampling and statistics
+
+The calibration used 200 unique paired windows from 202506010000 through 202606301100. Coverage included 13 months and all 24 hours, with 8 or 9 samples per hour, 14 to 16 per month, 56 weekend windows, and 144 weekday windows. The selector chose the nearest available pair around evenly spaced date-bin midpoints and permuted requested hours with `(index * 5) % 24`. Percentiles use Type 7 linear interpolation at rank `(n - 1) * p`.
+
+All 200 windows completed. Rust emitted 0 non-finite alpha values across the sample.
+
+Across-window distribution of each per-window statistic:
+
+| Statistic   | Mean    | SD     | Min     | p10     | Median  | p90     | Max     |
+| ----------- | ------- | ------ | ------- | ------- | ------- | ------- | ------- |
+| n_addresses | 318,709 | 21,833 | 274,916 | 296,414 | 314,254 | 349,859 | 396,341 |
+| min         | 0.199   | 0.024  | 0.170   | 0.178   | 0.191   | 0.229   | 0.277   |
+| p0_1        | 0.364   | 0.010  | 0.348   | 0.353   | 0.362   | 0.379   | 0.396   |
+| p1          | 0.394   | 0.009  | 0.371   | 0.383   | 0.392   | 0.408   | 0.421   |
+| p5          | 0.439   | 0.009  | 0.406   | 0.425   | 0.439   | 0.449   | 0.458   |
+| median      | 0.511   | 0.002  | 0.508   | 0.509   | 0.511   | 0.514   | 0.519   |
+| p95         | 1.021   | 0.006  | 1.006   | 1.013   | 1.021   | 1.028   | 1.055   |
+| p99         | 1.190   | 0.004  | 1.174   | 1.184   | 1.190   | 1.195   | 1.202   |
+| p99_9       | 1.379   | 0.012  | 1.345   | 1.363   | 1.380   | 1.392   | 1.410   |
+| max         | 3.194   | 1.155  | 2.013   | 2.149   | 2.265   | 4.507   | 4.917   |
+
+Eight representative windows:
+
+| Timestamp    | Addresses | Min   | p0.1  | Median | p99.9 | Max   |
+| ------------ | --------- | ----- | ----- | ------ | ----- | ----- |
+| 202506010000 | 298,018   | 0.180 | 0.352 | 0.510  | 1.354 | 4.457 |
+| 202507272015 | 300,483   | 0.177 | 0.360 | 0.510  | 1.392 | 4.457 |
+| 202509222130 | 315,015   | 0.181 | 0.359 | 0.510  | 1.377 | 4.509 |
+| 202511161745 | 331,615   | 0.203 | 0.363 | 0.513  | 1.373 | 2.193 |
+| 202601131800 | 341,425   | 0.231 | 0.368 | 0.514  | 1.365 | 4.539 |
+| 202603091415 | 316,560   | 0.198 | 0.361 | 0.512  | 1.385 | 2.239 |
+| 202605051545 | 360,662   | 0.200 | 0.371 | 0.516  | 1.363 | 2.214 |
+| 202606301100 | 349,565   | 0.190 | 0.369 | 0.514  | 1.357 | 2.209 |
+
+### Tail stability and time effects
+
+The table compares overall spread with date, hour, weekend, and month groupings. Date uses Pearson correlation against capture time. Hour and month cells show the groups with the lowest and highest means. Weekday/weekend is shown in that order.
+
+| Metric | Overall SD | Overall p10 to p90 | Date r | Hourly mean low/high    | Weekday/weekend mean | Monthly mean low/high        |
+| ------ | ---------- | ------------------ | ------ | ----------------------- | -------------------- | ---------------------------- |
+| min    | 0.024      | 0.178 to 0.229     | 0.052  | 0:00 0.186; 2:00 0.225  | 0.201/0.193          | 2026-03 0.190; 2025-12 0.207 |
+| p0_1   | 0.010      | 0.353 to 0.379     | 0.423  | 2:00 0.359; 5:00 0.370  | 0.366/0.361          | 2025-08 0.356; 2026-05 0.374 |
+| p99_9  | 0.012      | 1.363 to 1.392     | -0.360 | 9:00 1.373; 21:00 1.386 | 1.377/1.383          | 2026-05 1.368; 2025-08 1.387 |
+| max    | 1.155      | 2.149 to 4.507     | -0.607 | 4:00 2.724; 18:00 3.663 | 3.247/3.057          | 2025-11 2.178; 2025-06 4.509 |
+
+The center of the alpha distribution is tight: the per-window median has mean 0.511 and SD 0.002. The p0.1 and p99.9 tails have SDs of 0.010 and 0.012. Min and max are much noisier because a single address controls each value. The table keeps network growth and calendar effects separate from that single-address churn instead of treating every max swing as a feed-wide shift.
+
+There is a measured date shift, but it is small in absolute alpha terms. Address count rose with date at r=0.697; median alpha also rose at r=0.719, while its SD stayed 0.002. The p0.1 mean rose from a monthly low of 0.356 in August 2025 to 0.374 in May 2026, with date r=0.423. The p99.9 mean moved the other way, from 1.387 in August 2025 to 1.368 in May 2026, with r=-0.360.
+
+Time-of-day and weekday effects were smaller. Hourly p0.1 means ranged from 0.359 to 0.370, and hourly p99.9 means ranged from 1.373 to 1.386. Weekday versus weekend means differed by 0.005 for p0.1 and 0.006 for p99.9. Max alpha was the unstable statistic: SD 1.155, p10 to p90 2.149 to 4.507, and date r=-0.607. That reflects individual recurring addresses entering or leaving the tail, not a broad distribution shift.
+
+High alpha marks sparse, isolated address-space regions. Low alpha marks addresses that remain inside dense prefix clusters. Both tails therefore remain in the recommendation.
+
+### Initial candidate grid
+
+Counts here use the requested strict comparisons. Capped values apply the feed's 20-per-tail limit before combining tails.
+
+| High | Low | Raw mean | Raw median | Capped mean | Capped median |
+| ---- | --- | -------- | ---------- | ----------- | ------------- |
+| 2.5  | 0.3 | 6.5      | 6.0        | 5.7         | 6.0           |
+| 2.5  | 0.5 | 129,919  | 128,360    | 20.4        | 20.0          |
+| 2.5  | 0.7 | 245,417  | 241,792    | 20.4        | 20.0          |
+| 3.0  | 0.3 | 6.5      | 6.0        | 5.7         | 6.0           |
+| 3.0  | 0.5 | 129,919  | 128,360    | 20.4        | 20.0          |
+| 3.0  | 0.7 | 245,417  | 241,792    | 20.4        | 20.0          |
+| 3.5  | 0.3 | 6.5      | 6.0        | 5.7         | 6.0           |
+| 3.5  | 0.5 | 129,919  | 128,360    | 20.4        | 20.0          |
+| 3.5  | 0.7 | 245,417  | 241,792    | 20.4        | 20.0          |
+
+The initial high range of 2.5 to 3.5 is too conservative on this feed. It contributes almost no high-tail volume, leaving the low tail to determine the result. A high threshold of 2.0 restores a useful sparse-address signal without pinning the high tail at its cap.
+
+### Recommendation
+
+Use `threshold_high = 2` and `threshold_low = 0.3` as fixed global constants. Before caps, high 2.0 produces mean/median counts of 14.7/14.0 and low 0.3 produces 6.1/5.0. Combined raw mean/median is 20.8/20.0. After the 20-per-tail caps, combined mean/median is 19.7/20.0.
+
+The saved grid uses strict `>` and `<` as requested. The Rust feed uses inclusive `>=` and `<=`. Exactly 0 sampled address scores landed on either recommended constant, so inclusive feed-visible mean/median remains 19.7/20.0.
+
+I prefer 2.0 and 0.3 because they are round, give both tails room to contribute, and put a typical window inside the requested 5 to 30 recorded-alert range. The p10 to p90 capped range is 14 to 25. One May 2026 window had 177 raw low-tail crossings and 194 raw combined crossings; the per-tail caps reduced it to 37 recorded alerts. None of the 200 selected windows was quiet enough to produce zero at this pair, but an empty or genuinely low-traffic window can still do so.
+
+### Repeat offenders
+
+I selected eight evenly spaced calibration windows and retained every address beyond the recommended strict thresholds.
+
+| Timestamp    | High | Low | Combined |
+| ------------ | ---- | --- | -------- |
+| 202506010000 | 5    | 4   | 9        |
+| 202507272015 | 17   | 6   | 23       |
+| 202509222130 | 16   | 6   | 22       |
+| 202511161745 | 15   | 6   | 21       |
+| 202601131800 | 18   | 4   | 22       |
+| 202603091415 | 17   | 5   | 22       |
+| 202605051545 | 19   | 7   | 26       |
+| 202606301100 | 13   | 6   | 19       |
+
+Pairwise overlap across all 28 window pairs:
+
+| Tail     | Mean Jaccard | Median Jaccard | Jaccard range  | Mean intersection | Intersection range |
+| -------- | ------------ | -------------- | -------------- | ----------------- | ------------------ |
+| high     | 0.008        | 0.000          | 0.000 to 0.050 | 0.2               | 0 to 1             |
+| low      | 0.487        | 0.444          | 0.300 to 0.833 | 3.5               | 3 to 5             |
+| combined | 0.102        | 0.101          | 0.065 to 0.148 | 3.7               | 3 to 5             |
+
+Addresses recurring in at least two of the eight windows, limited to the ten most frequent:
+
+| Tail | Address         | Windows | Min alpha | Max alpha |
+| ---- | --------------- | ------- | --------- | --------- |
+| low  | 224.0.0.1       | 8/8     | 0.267     | 0.273     |
+| low  | 224.0.0.13      | 8/8     | 0.289     | 0.296     |
+| low  | 224.0.0.2       | 8/8     | 0.267     | 0.273     |
+| low  | 72.4.181.18     | 5/8     | 0.215     | 0.231     |
+| high | 239.255.255.250 | 4/8     | 4.457     | 4.539     |
+| low  | 72.5.64.18      | 3/8     | 0.198     | 0.214     |
+| low  | 60.247.96.22    | 2/8     | 0.190     | 0.200     |
+
+The two tails behave differently. High-tail membership mostly churned, with mean Jaccard 0.008 and pairwise intersections of at most one address. The only recurring high address was 239.255.255.250, present in four of eight windows. The low tail had a stable core: 224.0.0.1, 224.0.0.2, and 224.0.0.13 appeared in all eight windows, and low-tail mean Jaccard was 0.487. The feed should therefore expect repeat multicast-style low alerts alongside a changing high tail.
+
+## How to reproduce
+
+All scripts, raw outputs, address lists, logs, and summaries are under `/tmp/singularity-calibration/`. Nothing from this run was written to tracked repository files. The Cargo build wrote only the permitted `target/` output.
+
+Key audit files:
+
+- `inventory/conformance_candidates.txt` and `inventory/calibration_timestamps.txt` contain the selected timestamps.
+- `inventory/*_coverage.csv` records dates, hours, day types, and both capture paths.
+- `data/conformance_per_window/` and `data/calibration_per_window/` contain every per-window CSV summary.
+- `data/conformance_windows.csv` and `data/calibration_windows.csv` are the consolidated tables.
+- `data/conformance_worst_deviations.csv` records the address and absolute difference behind each displayed maximum.
+- `data/threshold_grid.csv`, `data/threshold_tail_sensitivity.csv`, and `data/threshold_pair_sensitivity.csv` contain the threshold scan.
+- `data/repeat_*.csv` contains the repeat-offender sets and overlaps.
+- `work/conformance/` retains Rust and Haskell CSVs plus exact address lists. `work/calibration/` retains all Rust CSVs and address lists.
+
+Run these commands from a fresh `/tmp/singularity-calibration/` layout after restoring the scripts:
+
+```sh
+cd /home/obo/.t3/worktrees/netflow-analysis/t3code-5e5d2a1c
+cargo build --release -p atlantis-netflow-db
+/tmp/singularity-calibration/scripts/select_windows.py
+/tmp/singularity-calibration/scripts/run_conformance.py /tmp/singularity-calibration/inventory/conformance_candidates.txt target/release/netflow-db --target 24 --workers 2 --timeout-seconds 600
+/tmp/singularity-calibration/scripts/run_calibration.py /tmp/singularity-calibration/inventory/calibration_timestamps.txt target/release/netflow-db --workers 4
+/tmp/singularity-calibration/scripts/find_worst_deviations.py
+/tmp/singularity-calibration/scripts/augment_candidate_pairs.py /tmp/singularity-calibration/data/calibration_windows.csv /tmp/singularity-calibration/data/calibration_per_window
+/tmp/singularity-calibration/scripts/scan_thresholds.py /tmp/singularity-calibration/work/calibration/scores /tmp/singularity-calibration/data/threshold_grid.csv
+/tmp/singularity-calibration/scripts/analyze_calibration.py /tmp/singularity-calibration/data/calibration_windows.csv /tmp/singularity-calibration/data/threshold_grid.csv /tmp/singularity-calibration/data
+/tmp/singularity-calibration/scripts/analyze_repeat_offenders.py /tmp/singularity-calibration/work/calibration/scores /tmp/singularity-calibration/inventory/calibration_timestamps.txt /tmp/singularity-calibration/data --high 2.0 --low 0.3 --windows 8
+/tmp/singularity-calibration/scripts/render_report.py
+```
+
+## Final concise summary
+
+Conformance verdict: PASS across 24 real windows and 7,633,352 address comparisons. Maximum displayed deviations were 7.737e-15 alpha, 1.404e-08 intercept, and 1.168e-14 r2. Tolerance violations, level mismatches, and missing addresses were all zero.
+
+Recommended constants: `threshold_high = 2` and `threshold_low = 0.3`. High 2.0 brings the sparse-address tail back into the feed, while low 0.3 keeps the dense-cluster tail selective. At the 20-per-tail cap, expected alerts per window are mean 19.7, median 20.0, with p10 to p90 of 14.0 to 25.0.
+
+Sensitivity around the recommendation:
+
+| High | Low   | Raw mean | Raw median | Raw p10 to p90 | Capped mean | Capped median | Capped p10 to p90 | Zero windows |
+| ---- | ----- | -------- | ---------- | -------------- | ----------- | ------------- | ----------------- | ------------ |
+| 1.75 | 0.3   | 63.2     | 62.0       | 51.0 to 74.1   | 25.2        | 25.0          | 24.0 to 27.0      | 0.0%         |
+| 1.9  | 0.3   | 34.3     | 33.5       | 26.0 to 41.0   | 25.1        | 25.0          | 24.0 to 27.0      | 0.0%         |
+| 2    | 0.25  | 16.8     | 17.0       | 11.0 to 23.0   | 16.5        | 17.0          | 11.0 to 22.0      | 0.0%         |
+| 2    | 0.275 | 19.2     | 19.0       | 13.0 to 25.0   | 18.6        | 19.0          | 13.0 to 24.0      | 0.0%         |
+| 2    | 0.3   | 20.8     | 20.0       | 14.0 to 26.0   | 19.7        | 20.0          | 14.0 to 25.0      | 0.0%         |
+| 2    | 0.325 | 22.0     | 21.0       | 15.0 to 27.0   | 20.9        | 21.0          | 15.0 to 26.0      | 0.0%         |
+| 2    | 0.35  | 41.2     | 27.0       | 17.9 to 40.1   | 26.4        | 26.0          | 17.9 to 36.1      | 0.0%         |
+| 2.1  | 0.3   | 11.9     | 11.0       | 7.0 to 15.1    | 11.1        | 11.0          | 7.0 to 15.1       | 0.0%         |
+| 2.25 | 0.3   | 6.7      | 6.0        | 4.0 to 8.0     | 6.0         | 6.0           | 4.0 to 8.0        | 0.0%         |
diff --git a/tools/netflow-db/src/feed.rs b/tools/netflow-db/src/feed.rs
new file mode 100644
index 00000000..69baf8d6
--- /dev/null
+++ b/tools/netflow-db/src/feed.rs
@@ -0,0 +1,941 @@
+//! Continuous Singularity alert feed over an nfcapd capture tree.
+//!
+//! `netflow-db feed <dataset>` is a long-running process: it polls the
+//! dataset's capture tree for newly completed five-minute buckets, unions the
+//! distinct addresses across the dataset's members for each window, scores
+//! them with [`crate::singularity`], and appends threshold-crossing addresses
+//! to a rolling alert database (`alerts.sqlite` beside the dataset's product
+//! database). Rows older than the retention window are pruned on each pass.
+//!
+//! The alert database is an ephemeral rolling buffer owned by this module,
+//! not a pipeline product database: it carries no dataset identity or
+//! coverage semantics and is safe to delete at any time.
+
+use std::{
+    collections::BTreeSet,
+    fs,
+    net::{IpAddr, Ipv4Addr},
+    path::{Path, PathBuf},
+    time::{Duration, Instant},
+};
+
+use jiff::Timestamp;
+use rusqlite::{Connection, TransactionBehavior, params};
+
+use crate::{
+    domain::{AddressSide, CanonicalBucket, FlowSelection, IpVersion, Scope, Visibility},
+    ingest,
+    registry::{self, DatasetRegistry},
+    singularity,
+};
+
+const WINDOW_SECONDS: i64 = 300;
+const GRACE_SECONDS: i64 = 10 * 60;
+const SECONDS_PER_HOUR: i64 = 60 * 60;
+const SECONDS_PER_DAY: i64 = 24 * SECONDS_PER_HOUR;
+
+// This mirrors pipeline.rs's private DEFAULT_TIMEZONE. Registry datasets do
+// not carry a timezone, and registry-driven pipeline runs use this default.
+const TIMEZONE: &str = "America/Los_Angeles";
+
+// Calibrated on 200 uOregon five-minute windows spanning 2025-06 through
+// 2026-06 after conformance against the Haskell reference; a typical window
+// records ~20 alerts (p10-p90: 14-25) across both tails at these values.
+// Methodology and sensitivity: docs/agent/singularity-calibration.md.
+const DEFAULT_THRESHOLD_HIGH: f64 = 2.0;
+const DEFAULT_THRESHOLD_LOW: f64 = 0.3;
+
+const ALERT_SCHEMA: &str = r#"
+CREATE TABLE IF NOT EXISTS feed_meta (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS windows (
+    window_start INTEGER PRIMARY KEY,
+    window_end INTEGER NOT NULL,
+    member_files INTEGER NOT NULL,
+    address_count INTEGER NOT NULL,
+    alert_count INTEGER NOT NULL,
+    alpha_min REAL,
+    alpha_max REAL,
+    alpha_median REAL,
+    processed_at INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS alerts (
+    window_start INTEGER NOT NULL REFERENCES windows(window_start) ON DELETE CASCADE,
+    address TEXT NOT NULL,
+    alpha REAL NOT NULL,
+    tail TEXT NOT NULL CHECK (tail IN ('high','low')),
+    rank INTEGER NOT NULL,
+    r2 REAL NOT NULL,
+    prefix_levels INTEGER NOT NULL,
+    PRIMARY KEY (window_start, tail, rank)
+);
+CREATE INDEX IF NOT EXISTS alerts_address ON alerts(address, window_start);
+"#;
+
+/// Configuration for one `netflow-db feed` run.
+#[derive(Debug)]
+pub struct FeedOptions {
+    /// Dataset id resolved through the datasets registry.
+    pub dataset_id: String,
+    /// Registry path override; defaults to standard `datasets.json` discovery.
+    pub registry_path: Option<PathBuf>,
+    /// Alert database path; defaults to `alerts.sqlite` beside the dataset's
+    /// configured `db_path`.
+    pub database_path: Option<PathBuf>,
+    /// nfdump executable (the pinned fork supporting the atlantis contract).
+    pub nfdump: String,
+    /// Seconds between capture-tree scans.
+    pub poll_seconds: u64,
+    /// Days of alerts to retain.
+    pub retention_days: u32,
+    /// Cap on recorded alerts per tail per window.
+    pub max_per_tail: u32,
+    /// Alpha at or above which an address alerts; `None` uses the module default.
+    pub threshold_high: Option<f64>,
+    /// Alpha at or below which an address alerts; `None` uses the module default.
+    pub threshold_low: Option<f64>,
+    /// Also process historical windows this far back, e.g. `"36h"` or `"7d"`.
+    pub backfill: Option<String>,
+    /// Process available windows once and exit instead of polling.
+    pub once: bool,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum FeedError {
+    #[error(transparent)]
+    Registry(#[from] registry::RegistryError),
+    #[error("alert database error: {0}")]
+    Database(#[from] rusqlite::Error),
+    #[error("feed I/O error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("invalid feed configuration: {0}")]
+    InvalidConfig(String),
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum AlertTail {
+    High,
+    Low,
+}
+
+impl AlertTail {
+    const fn as_str(self) -> &'static str {
+        match self {
+            Self::High => "high",
+            Self::Low => "low",
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq)]
+struct SelectedAlert {
+    address: Ipv4Addr,
+    alpha: f64,
+    tail: AlertTail,
+    rank: u32,
+    r2: f64,
+    prefix_levels: u8,
+}
+
+#[derive(Debug)]
+struct MemberFile {
+    member_id: String,
+    path: PathBuf,
+}
+
+#[derive(Debug)]
+struct WindowReadiness {
+    existing_files: Vec<MemberFile>,
+    processable: bool,
+}
+
+struct FeedContext<'a> {
+    root_path: &'a Path,
+    members: &'a [String],
+    nfdump: &'a str,
+    selection: &'a FlowSelection,
+    threshold_high: f64,
+    threshold_low: f64,
+    max_per_tail: u32,
+    retention_days: u32,
+}
+
+/// Run the feed until interrupted (or once, with [`FeedOptions::once`]).
+pub fn run(options: FeedOptions) -> Result<(), FeedError> {
+    let repository_root = std::env::current_dir()?;
+    let registry = match &options.registry_path {
+        Some(path) => DatasetRegistry::load(path, &repository_root)?,
+        None => DatasetRegistry::load_default(&repository_root)?,
+    };
+    let dataset = registry.get(&options.dataset_id)?.clone();
+    let members = dataset
+        .logical_sources()?
+        .into_iter()
+        .flat_map(|source| source.members)
+        .collect::<BTreeSet<_>>()
+        .into_iter()
+        .collect::<Vec<_>>();
+    if members.is_empty() {
+        return Err(FeedError::InvalidConfig(format!(
+            "dataset {:?} has no capture members",
+            dataset.dataset_id
+        )));
+    }
+
+    let threshold_high = options.threshold_high.unwrap_or(DEFAULT_THRESHOLD_HIGH);
+    let threshold_low = options.threshold_low.unwrap_or(DEFAULT_THRESHOLD_LOW);
+    if !threshold_high.is_finite() || !threshold_low.is_finite() {
+        return Err(FeedError::InvalidConfig(
+            "alert thresholds must be finite numbers".into(),
+        ));
+    }
+    let backfill_seconds = options
+        .backfill
+        .as_deref()
+        .map(parse_backfill_duration)
+        .transpose()?;
+
+    let database_path = options.database_path.clone().unwrap_or_else(|| {
+        dataset
+            .db_path
+            .parent()
+            .unwrap_or_else(|| Path::new("."))
+            .join("alerts.sqlite")
+    });
+    if let Some(parent) = database_path
+        .parent()
+        .filter(|parent| !parent.as_os_str().is_empty())
+    {
+        fs::create_dir_all(parent)?;
+    }
+
+    let mut connection = open_alert_database(&database_path)?;
+    upsert_feed_meta(
+        &connection,
+        &dataset.dataset_id,
+        threshold_high,
+        threshold_low,
+        options.max_per_tail,
+    )?;
+
+    let startup_time = Timestamp::now().as_second();
+    // Windows older than the retention cutoff would be pruned in the same
+    // pass that processed them, so a backfill deeper than retention is
+    // clamped rather than wasted.
+    let retention_floor = retention_cutoff(startup_time, options.retention_days);
+    let requested_start = backfill_seconds
+        .map(|duration| startup_time.saturating_sub(duration))
+        .unwrap_or(startup_time);
+    if requested_start < retention_floor {
+        tracing::warn!(
+            retention_days = options.retention_days,
+            "backfill exceeds retention; clamping scan start to the retention cutoff"
+        );
+    }
+    let initial_scan_start = align_window_at_or_after(requested_start.max(retention_floor));
+    let selection = FlowSelection::default();
+    let context = FeedContext {
+        root_path: &dataset.root_path,
+        members: &members,
+        nfdump: &options.nfdump,
+        selection: &selection,
+        threshold_high,
+        threshold_low,
+        max_per_tail: options.max_per_tail,
+        retention_days: options.retention_days,
+    };
+
+    loop {
+        let now = Timestamp::now().as_second();
+        process_pass(&mut connection, &context, initial_scan_start, now)?;
+        prune_windows(&connection, retention_cutoff(now, options.retention_days))?;
+
+        if options.once {
+            return Ok(());
+        }
+        std::thread::sleep(Duration::from_secs(options.poll_seconds));
+    }
+}
+
+fn open_alert_database(path: &Path) -> Result<Connection, FeedError> {
+    let connection = Connection::open(path)?;
+    init_alert_schema(&connection)?;
+    Ok(connection)
+}
+
+fn init_alert_schema(connection: &Connection) -> Result<(), FeedError> {
+    connection.busy_timeout(Duration::from_millis(crate::storage::BUSY_TIMEOUT_MS))?;
+    connection.pragma_update(None, "foreign_keys", "ON")?;
+    connection.pragma_update(None, "journal_mode", "WAL")?;
+    connection.execute_batch(ALERT_SCHEMA)?;
+    Ok(())
+}
+
+fn upsert_feed_meta(
+    connection: &Connection,
+    dataset_id: &str,
+    threshold_high: f64,
+    threshold_low: f64,
+    max_per_tail: u32,
+) -> Result<(), FeedError> {
+    const UPSERT: &str = "
+        INSERT INTO feed_meta (key, value) VALUES (?1, ?2)
+        ON CONFLICT(key) DO UPDATE SET value = excluded.value
+    ";
+    for (key, value) in [
+        ("schema_version", "1".to_owned()),
+        ("dataset_id", dataset_id.to_owned()),
+        ("threshold_high", threshold_high.to_string()),
+        ("threshold_low", threshold_low.to_string()),
+        ("max_per_tail", max_per_tail.to_string()),
+    ] {
+        connection.execute(UPSERT, params![key, value])?;
+    }
+    Ok(())
+}
+
+fn process_pass(
+    connection: &mut Connection,
+    context: &FeedContext<'_>,
+    initial_scan_start: i64,
+    now: i64,
+) -> Result<(), FeedError> {
+    let mut window_start =
+        next_scan_start(connection, initial_scan_start, now, context.retention_days)?;
+
+    loop {
+        let window_end = window_start.checked_add(WINDOW_SECONDS).ok_or_else(|| {
+            FeedError::InvalidConfig("candidate window timestamp overflowed".into())
+        })?;
+        if window_end > now {
+            break;
+        }
+        if window_exists(connection, window_start)? {
+            window_start = window_end;
+            continue;
+        }
+
+        let readiness = window_readiness(context.root_path, context.members, window_start, now)?;
+        if !readiness.processable {
+            break;
+        }
+
+        let started_at = Instant::now();
+        let mut member_files = 0_i64;
+        let mut addresses = BTreeSet::new();
+        for member_file in readiness.existing_files {
+            match ingest::read_nfcapd_bucket(
+                &member_file.path,
+                &member_file.member_id,
+                context.selection,
+                context.nfdump,
+                TIMEZONE,
+            ) {
+                Ok(bucket) => {
+                    member_files += 1;
+                    collect_total_ipv4_addresses(bucket, &mut addresses);
+                }
+                Err(error) => tracing::warn!(
+                    path = %member_file.path.display(),
+                    member = %member_file.member_id,
+                    error = %error,
+                    "failed to read feed capture file"
+                ),
+            }
+        }
+
+        if member_files == 0 {
+            tracing::warn!(
+                window_start,
+                window_end,
+                "no member files could be read for feed window"
+            );
+            window_start = window_end;
+            continue;
+        }
+
+        let scores = singularity::score(addresses.into_iter().collect());
+        let alerts = select_alerts(
+            &scores,
+            context.threshold_high,
+            context.threshold_low,
+            context.max_per_tail,
+        );
+        let (alpha_min, alpha_max, alpha_median) = alpha_summary(&scores);
+        write_window(
+            connection,
+            &WindowRecord {
+                window_start,
+                window_end,
+                member_files,
+                scores: &scores,
+                alerts: &alerts,
+                alpha_min,
+                alpha_max,
+                alpha_median,
+                processed_at: now,
+            },
+        )?;
+        prune_windows(connection, retention_cutoff(now, context.retention_days))?;
+
+        tracing::info!(
+            window_start,
+            window_end,
+            address_count = scores.len(),
+            alert_count = alerts.len(),
+            duration_ms = started_at.elapsed().as_millis() as u64,
+            "processed feed window"
+        );
+        window_start = window_end;
+    }
+
+    Ok(())
+}
+
+fn collect_total_ipv4_addresses(bucket: CanonicalBucket, addresses: &mut BTreeSet<Ipv4Addr>) {
+    let total_ipv4_scope = Scope::new(IpVersion::V4, Visibility::All, Visibility::All);
+    for scoped in bucket.addresses {
+        if scoped.scope != total_ipv4_scope
+            || !matches!(
+                scoped.address_side,
+                AddressSide::Source | AddressSide::Destination
+            )
+        {
+            continue;
+        }
+        for address in scoped.addresses.iter() {
+            if let IpAddr::V4(address) = address {
+                addresses.insert(*address);
+            }
+        }
+    }
+}
+
+/// One processed window's row data, written transactionally by [`write_window`].
+struct WindowRecord<'a> {
+    window_start: i64,
+    window_end: i64,
+    member_files: i64,
+    scores: &'a [singularity::AddressScore],
+    alerts: &'a [SelectedAlert],
+    alpha_min: Option<f64>,
+    alpha_max: Option<f64>,
+    alpha_median: Option<f64>,
+    processed_at: i64,
+}
+
+fn write_window(connection: &mut Connection, record: &WindowRecord) -> Result<(), FeedError> {
+    let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
+    transaction.execute(
+        "INSERT INTO windows (
+            window_start, window_end, member_files, address_count, alert_count,
+            alpha_min, alpha_max, alpha_median, processed_at
+         ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
+        params![
+            record.window_start,
+            record.window_end,
+            record.member_files,
+            record.scores.len() as i64,
+            record.alerts.len() as i64,
+            record.alpha_min,
+            record.alpha_max,
+            record.alpha_median,
+            record.processed_at,
+        ],
+    )?;
+    for alert in record.alerts {
+        transaction.execute(
+            "INSERT INTO alerts (
+                window_start, address, alpha, tail, rank, r2, prefix_levels
+             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
+            params![
+                record.window_start,
+                alert.address.to_string(),
+                alert.alpha,
+                alert.tail.as_str(),
+                i64::from(alert.rank),
+                alert.r2,
+                i64::from(alert.prefix_levels),
+            ],
+        )?;
+    }
+    transaction.commit()?;
+    Ok(())
+}
+
+fn prune_windows(connection: &Connection, cutoff: i64) -> Result<(), FeedError> {
+    connection.execute(
+        "DELETE FROM windows WHERE window_start < ?1",
+        params![cutoff],
+    )?;
+    Ok(())
+}
+
+fn window_readiness(
+    root: &Path,
+    members: &[String],
+    window_start: i64,
+    now: i64,
+) -> Result<WindowReadiness, FeedError> {
+    let mut existing_files = Vec::with_capacity(members.len());
+    for member in members {
+        let path = expected_nfcapd_path(root, member, window_start)?;
+        if path.exists() {
+            existing_files.push(MemberFile {
+                member_id: member.clone(),
+                path,
+            });
+        }
+    }
+    let all_present = existing_files.len() == members.len();
+    let window_end = window_start
+        .checked_add(WINDOW_SECONDS)
+        .ok_or_else(|| FeedError::InvalidConfig("candidate window timestamp overflowed".into()))?;
+    let past_grace = now.saturating_sub(window_end) > GRACE_SECONDS;
+    Ok(WindowReadiness {
+        existing_files,
+        processable: all_present || past_grace,
+    })
+}
+
+fn expected_nfcapd_path(
+    root: &Path,
+    member: &str,
+    window_start: i64,
+) -> Result<PathBuf, FeedError> {
+    let timestamp = Timestamp::from_second(window_start)
+        .and_then(|timestamp| timestamp.in_tz(TIMEZONE))
+        .map_err(|error| {
+            FeedError::InvalidConfig(format!(
+                "invalid feed window timestamp {window_start}: {error}"
+            ))
+        })?;
+    Ok(root
+        .join(member)
+        .join(timestamp.strftime("%Y").to_string())
+        .join(timestamp.strftime("%m").to_string())
+        .join(timestamp.strftime("%d").to_string())
+        .join(format!("nfcapd.{}", timestamp.strftime("%Y%m%d%H%M"))))
+}
+
+fn next_scan_start(
+    connection: &Connection,
+    initial_scan_start: i64,
+    now: i64,
+    retention_days: u32,
+) -> Result<i64, FeedError> {
+    let newest = connection.query_row("SELECT MAX(window_start) FROM windows", [], |row| {
+        row.get::<_, Option<i64>>(0)
+    })?;
+    match newest {
+        Some(window_start) => {
+            let after_newest = window_start.checked_add(WINDOW_SECONDS).ok_or_else(|| {
+                FeedError::InvalidConfig("processed window timestamp overflowed".into())
+            })?;
+            Ok(after_newest.max(align_window_at_or_after(retention_cutoff(
+                now,
+                retention_days,
+            ))))
+        }
+        None => Ok(initial_scan_start),
+    }
+}
+
+fn window_exists(connection: &Connection, window_start: i64) -> Result<bool, FeedError> {
+    Ok(connection.query_row(
+        "SELECT EXISTS(SELECT 1 FROM windows WHERE window_start = ?1)",
+        params![window_start],
+        |row| row.get(0),
+    )?)
+}
+
+fn retention_cutoff(now: i64, retention_days: u32) -> i64 {
+    now.saturating_sub(i64::from(retention_days) * SECONDS_PER_DAY)
+}
+
+fn align_window_at_or_after(timestamp: i64) -> i64 {
+    let remainder = timestamp.rem_euclid(WINDOW_SECONDS);
+    if remainder == 0 {
+        timestamp
+    } else {
+        timestamp.saturating_add(WINDOW_SECONDS - remainder)
+    }
+}
+
+fn parse_backfill_duration(value: &str) -> Result<i64, FeedError> {
+    let Some((unit_index, unit)) = value.char_indices().last() else {
+        return Err(invalid_backfill(value));
+    };
+    let amount = &value[..unit_index];
+    if amount.is_empty() || !amount.bytes().all(|byte| byte.is_ascii_digit()) {
+        return Err(invalid_backfill(value));
+    }
+    let amount = amount.parse::<i64>().map_err(|_| invalid_backfill(value))?;
+    let unit_seconds = match unit {
+        'm' => 60,
+        'h' => SECONDS_PER_HOUR,
+        'd' => SECONDS_PER_DAY,
+        _ => return Err(invalid_backfill(value)),
+    };
+    amount.checked_mul(unit_seconds).ok_or_else(|| {
+        FeedError::InvalidConfig(format!("backfill duration {value:?} is too large"))
+    })
+}
+
+fn invalid_backfill(value: &str) -> FeedError {
+    FeedError::InvalidConfig(format!(
+        "invalid backfill duration {value:?}; expected an integer followed by 'm', 'h', or 'd'"
+    ))
+}
+
+fn alpha_summary(scores: &[singularity::AddressScore]) -> (Option<f64>, Option<f64>, Option<f64>) {
+    if scores.is_empty() {
+        return (None, None, None);
+    }
+    let mut alphas = scores.iter().map(|score| score.alpha).collect::<Vec<_>>();
+    alphas.sort_by(f64::total_cmp);
+    let median_index = alphas.len() / 2;
+    let median = if alphas.len() % 2 == 0 {
+        alphas[median_index - 1] / 2.0 + alphas[median_index] / 2.0
+    } else {
+        alphas[median_index]
+    };
+    (
+        alphas.first().copied(),
+        alphas.last().copied(),
+        Some(median),
+    )
+}
+
+fn select_alerts(
+    scores: &[singularity::AddressScore],
+    threshold_high: f64,
+    threshold_low: f64,
+    max_per_tail: u32,
+) -> Vec<SelectedAlert> {
+    let mut high = scores
+        .iter()
+        .filter(|score| score.alpha >= threshold_high)
+        .collect::<Vec<_>>();
+    high.sort_by(|left, right| {
+        right
+            .alpha
+            .total_cmp(&left.alpha)
+            .then_with(|| left.address.cmp(&right.address))
+    });
+
+    let mut low = scores
+        .iter()
+        .filter(|score| score.alpha <= threshold_low)
+        .collect::<Vec<_>>();
+    low.sort_by(|left, right| {
+        left.alpha
+            .total_cmp(&right.alpha)
+            .then_with(|| left.address.cmp(&right.address))
+    });
+
+    let limit = max_per_tail as usize;
+    high.into_iter()
+        .take(limit)
+        .enumerate()
+        .map(|(index, score)| selected_alert(score, AlertTail::High, index))
+        .chain(
+            low.into_iter()
+                .take(limit)
+                .enumerate()
+                .map(|(index, score)| selected_alert(score, AlertTail::Low, index)),
+        )
+        .collect()
+}
+
+fn selected_alert(
+    score: &singularity::AddressScore,
+    tail: AlertTail,
+    index: usize,
+) -> SelectedAlert {
+    SelectedAlert {
+        address: score.address,
+        alpha: score.alpha,
+        tail,
+        rank: index as u32 + 1,
+        r2: score.r_squared,
+        prefix_levels: score.prefix_levels,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::fs;
+
+    use tempfile::tempdir;
+
+    use super::*;
+
+    #[test]
+    fn schema_initialization_is_idempotent() {
+        let temporary = tempdir().unwrap();
+        let database_path = temporary.path().join("alerts.sqlite");
+
+        {
+            let connection = Connection::open(&database_path).unwrap();
+            init_alert_schema(&connection).unwrap();
+            init_alert_schema(&connection).unwrap();
+        }
+        let connection = Connection::open(&database_path).unwrap();
+        init_alert_schema(&connection).unwrap();
+
+        let tables = connection
+            .prepare(
+                "SELECT name FROM sqlite_master
+                 WHERE type = 'table' AND name IN ('feed_meta', 'windows', 'alerts')
+                 ORDER BY name",
+            )
+            .unwrap()
+            .query_map([], |row| row.get::<_, String>(0))
+            .unwrap()
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        assert_eq!(tables, vec!["alerts", "feed_meta", "windows"]);
+
+        let alert_columns = connection
+            .prepare("PRAGMA table_info(alerts)")
+            .unwrap()
+            .query_map([], |row| row.get::<_, String>(1))
+            .unwrap()
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        assert_eq!(
+            alert_columns,
+            vec![
+                "window_start",
+                "address",
+                "alpha",
+                "tail",
+                "rank",
+                "r2",
+                "prefix_levels"
+            ]
+        );
+    }
+
+    #[test]
+    fn pruning_removes_old_windows_and_their_alerts() {
+        let temporary = tempdir().unwrap();
+        let connection = open_alert_database(&temporary.path().join("alerts.sqlite")).unwrap();
+        insert_window(&connection, 100);
+        insert_window(&connection, 1_000);
+        for window_start in [100, 1_000] {
+            connection
+                .execute(
+                    "INSERT INTO alerts
+                     (window_start, address, alpha, tail, rank, r2, prefix_levels)
+                     VALUES (?1, '192.0.2.1', 4.0, 'high', 1, 0.9, 12)",
+                    params![window_start],
+                )
+                .unwrap();
+        }
+
+        prune_windows(&connection, 1_000).unwrap();
+
+        let windows = query_i64_column(&connection, "SELECT window_start FROM windows");
+        let alerts = query_i64_column(&connection, "SELECT window_start FROM alerts");
+        assert_eq!(windows, vec![1_000]);
+        assert_eq!(alerts, vec![1_000]);
+    }
+
+    #[test]
+    fn parses_minute_hour_and_day_backfills() {
+        assert_eq!(parse_backfill_duration("15m").unwrap(), 15 * 60);
+        assert_eq!(
+            parse_backfill_duration("36h").unwrap(),
+            36 * SECONDS_PER_HOUR
+        );
+        assert_eq!(parse_backfill_duration("7d").unwrap(), 7 * SECONDS_PER_DAY);
+    }
+
+    #[test]
+    fn rejects_malformed_backfills() {
+        for value in ["", "36", "h", "1.5h", "-2d", "7w", " 7d"] {
+            assert!(
+                parse_backfill_duration(value).is_err(),
+                "accepted {value:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn window_readiness_obeys_file_completeness_and_grace_period() {
+        let temporary = tempdir().unwrap();
+        let members = vec!["alpha".to_owned(), "beta".to_owned()];
+        let complete_start = 1_700_000_100;
+        for member in &members {
+            create_empty_capture(temporary.path(), member, complete_start);
+        }
+
+        let complete = window_readiness(
+            temporary.path(),
+            &members,
+            complete_start,
+            complete_start + WINDOW_SECONDS,
+        )
+        .unwrap();
+        assert!(complete.processable);
+        assert_eq!(complete.existing_files.len(), 2);
+
+        let partial_start = complete_start + WINDOW_SECONDS;
+        create_empty_capture(temporary.path(), &members[0], partial_start);
+        let before_grace = window_readiness(
+            temporary.path(),
+            &members,
+            partial_start,
+            partial_start + WINDOW_SECONDS + GRACE_SECONDS,
+        )
+        .unwrap();
+        assert!(!before_grace.processable);
+        assert_eq!(before_grace.existing_files.len(), 1);
+
+        let after_grace = window_readiness(
+            temporary.path(),
+            &members,
+            partial_start,
+            partial_start + WINDOW_SECONDS + GRACE_SECONDS + 1,
+        )
+        .unwrap();
+        assert!(after_grace.processable);
+        assert_eq!(after_grace.existing_files.len(), 1);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn process_pass_skips_missing_window_and_processes_later_window() {
+        let temporary = tempdir().unwrap();
+        let root = temporary.path().join("captures");
+        let members = vec!["alpha".to_owned(), "beta".to_owned()];
+        let missing_start = 1_700_000_100;
+        let populated_start = missing_start + WINDOW_SECONDS;
+        for member in &members {
+            create_empty_capture(&root, member, populated_start);
+        }
+
+        let decoder = temporary.path().join("fake-nfdump");
+        write_fake_nfdump(&decoder, "");
+        let mut connection = open_alert_database(&temporary.path().join("alerts.sqlite")).unwrap();
+        let selection = FlowSelection::default();
+        let context = FeedContext {
+            root_path: &root,
+            members: &members,
+            nfdump: decoder.to_str().unwrap(),
+            selection: &selection,
+            threshold_high: DEFAULT_THRESHOLD_HIGH,
+            threshold_low: DEFAULT_THRESHOLD_LOW,
+            max_per_tail: 0,
+            retention_days: 1,
+        };
+        let now = missing_start + WINDOW_SECONDS + GRACE_SECONDS + 1;
+
+        process_pass(&mut connection, &context, missing_start, now).unwrap();
+
+        assert!(!window_exists(&connection, missing_start).unwrap());
+        assert!(window_exists(&connection, populated_start).unwrap());
+    }
+
+    #[test]
+    fn resume_starts_after_the_newest_window_and_respects_retention_floor() {
+        let temporary = tempdir().unwrap();
+        let connection = open_alert_database(&temporary.path().join("alerts.sqlite")).unwrap();
+        insert_window(&connection, 900);
+
+        assert_eq!(next_scan_start(&connection, 0, 2_000, 7).unwrap(), 1_200);
+
+        connection.execute("DELETE FROM windows", []).unwrap();
+        insert_window(&connection, 0);
+        let now = 10 * SECONDS_PER_DAY;
+        let retention_floor = align_window_at_or_after(now - SECONDS_PER_DAY);
+        assert_eq!(
+            next_scan_start(&connection, 0, now, 1).unwrap(),
+            retention_floor
+        );
+    }
+
+    #[test]
+    fn alert_selection_caps_and_ranks_both_tails_deterministically() {
+        let scores = vec![
+            score_fixture([192, 0, 2, 4], 5.0),
+            score_fixture([192, 0, 2, 2], 5.0),
+            score_fixture([192, 0, 2, 3], 4.0),
+            score_fixture([192, 0, 2, 9], -0.2),
+            score_fixture([192, 0, 2, 7], 0.1),
+            score_fixture([192, 0, 2, 8], 0.3),
+            score_fixture([192, 0, 2, 6], 1.0),
+        ];
+
+        let alerts = select_alerts(&scores, 3.5, 0.4, 2);
+        assert_eq!(alerts.len(), 4);
+        assert_eq!(
+            alerts
+                .iter()
+                .map(|alert| (alert.tail, alert.rank, alert.address, alert.alpha))
+                .collect::<Vec<_>>(),
+            vec![
+                (AlertTail::High, 1, Ipv4Addr::new(192, 0, 2, 2), 5.0),
+                (AlertTail::High, 2, Ipv4Addr::new(192, 0, 2, 4), 5.0),
+                (AlertTail::Low, 1, Ipv4Addr::new(192, 0, 2, 9), -0.2),
+                (AlertTail::Low, 2, Ipv4Addr::new(192, 0, 2, 7), 0.1),
+            ]
+        );
+    }
+
+    fn insert_window(connection: &Connection, window_start: i64) {
+        connection
+            .execute(
+                "INSERT INTO windows (
+                    window_start, window_end, member_files, address_count, alert_count,
+                    alpha_min, alpha_max, alpha_median, processed_at
+                 ) VALUES (?1, ?2, 1, 1, 1, 1.0, 1.0, 1.0, ?1)",
+                params![window_start, window_start + WINDOW_SECONDS],
+            )
+            .unwrap();
+    }
+
+    fn query_i64_column(connection: &Connection, query: &str) -> Vec<i64> {
+        connection
+            .prepare(query)
+            .unwrap()
+            .query_map([], |row| row.get(0))
+            .unwrap()
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap()
+    }
+
+    fn create_empty_capture(root: &Path, member: &str, window_start: i64) {
+        let path = expected_nfcapd_path(root, member, window_start).unwrap();
+        fs::create_dir_all(path.parent().unwrap()).unwrap();
+        fs::write(path, []).unwrap();
+    }
+
+    #[cfg(unix)]
+    fn write_fake_nfdump(executable: &std::path::Path, setup: &str) {
+        use std::os::unix::fs::PermissionsExt;
+
+        let stream = executable.with_extension("stream");
+        fs::write(&stream, crate::nfdump::ONE_V4_TEST_STREAM).unwrap();
+        fs::write(
+            executable,
+            format!("#!/bin/sh\n{setup}\ncat '{}'\n", stream.display()),
+        )
+        .unwrap();
+        fs::set_permissions(executable, fs::Permissions::from_mode(0o755)).unwrap();
+    }
+
+    fn score_fixture(address: [u8; 4], alpha: f64) -> singularity::AddressScore {
+        singularity::AddressScore {
+            address: Ipv4Addr::from(address),
+            alpha,
+            intercept: 0.0,
+            r_squared: 0.95,
+            prefix_levels: 12,
+        }
+    }
+}
diff --git a/tools/netflow-db/src/lib.rs b/tools/netflow-db/src/lib.rs
index 45e0db6a..abfbbb42 100644
--- a/tools/netflow-db/src/lib.rs
+++ b/tools/netflow-db/src/lib.rs
@@ -10,6 +10,7 @@ pub mod config;
 pub mod coverage;
 pub mod domain;
 pub mod export;
+pub mod feed;
 pub mod ingest;
 pub mod maad;
 pub(crate) mod nfdump;
@@ -20,5 +21,6 @@ pub mod prepare;
 pub mod provenance;
 pub mod publish;
 pub mod registry;
+pub mod singularity;
 pub mod storage;
 pub mod verify;
diff --git a/tools/netflow-db/src/main.rs b/tools/netflow-db/src/main.rs
index 41c1bc68..e047d5b7 100644
--- a/tools/netflow-db/src/main.rs
+++ b/tools/netflow-db/src/main.rs
@@ -11,12 +11,13 @@ use clap::{Args, Parser, Subcommand, ValueEnum};
 use netflow_db::{
     compare::{CompareOptions, compare_databases},
     export::{ExtractRequest, extract_window, validate_extract_plan},
-    maad,
+    feed, maad,
     operations::{
         UgrAssetKind, scrape_ugr16_urls, select_web_verification_window, verify_web_routes,
     },
     prepare::{PrepareOptions, prepare_archive},
     registry::DatasetRegistry,
+    singularity,
     storage::{backup_database, promote_database},
     verify::{VerifyOptions, verify_database},
 };
@@ -48,6 +49,10 @@ enum Command {
     VerifyWebRoutes(WebVerifyArgs),
     /// Compute MAAD JSON from IPv4 addresses, one per line.
     Maad(MaadArgs),
+    /// Score IPv4 addresses (one per line) by Singularity alpha, as CSV.
+    Singularity(SingularityArgs),
+    /// Maintain a rolling Singularity alert feed over live five-minute captures.
+    Feed(FeedArgs),
     /// Print the persisted pipeline contract version.
     ContractVersion,
 }
@@ -235,6 +240,47 @@ struct MaadArgs {
     input: Option<PathBuf>,
 }
 
+#[derive(Debug, Args)]
+struct SingularityArgs {
+    /// Read addresses from this file instead of standard input.
+    input: Option<PathBuf>,
+}
+
+#[derive(Debug, Args)]
+struct FeedArgs {
+    /// Dataset id from the datasets registry.
+    dataset: String,
+    /// Registry path override (defaults to datasets.json discovery).
+    #[arg(long)]
+    datasets: Option<PathBuf>,
+    /// Alert database path (defaults to alerts.sqlite beside the dataset's database).
+    #[arg(long)]
+    database_path: Option<PathBuf>,
+    #[arg(long, default_value = "nfdump")]
+    nfdump: String,
+    /// Seconds between capture-tree scans.
+    #[arg(long, default_value_t = 30)]
+    poll_seconds: u64,
+    /// Days of alerts to retain.
+    #[arg(long, default_value_t = 7)]
+    retention_days: u32,
+    /// Maximum alerts recorded per tail per window.
+    #[arg(long, default_value_t = 20)]
+    max_per_tail: u32,
+    /// Alpha at or above which an address alerts (defaults to the calibrated value).
+    #[arg(long)]
+    threshold_high: Option<f64>,
+    /// Alpha at or below which an address alerts (defaults to the calibrated value).
+    #[arg(long)]
+    threshold_low: Option<f64>,
+    /// Also process historical windows this far back (e.g. "36h", "7d").
+    #[arg(long)]
+    backfill: Option<String>,
+    /// Process available windows once and exit instead of polling.
+    #[arg(long)]
+    once: bool,
+}
+
 fn main() -> Result<()> {
     tracing_subscriber::fmt()
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
@@ -262,6 +308,20 @@ fn main() -> Result<()> {
         Command::ScrapeUgr16(args) => run_scrape(args)?,
         Command::VerifyWebRoutes(args) => run_web_verify(args)?,
         Command::Maad(args) => run_maad(args)?,
+        Command::Singularity(args) => run_singularity(args)?,
+        Command::Feed(args) => feed::run(feed::FeedOptions {
+            dataset_id: args.dataset,
+            registry_path: args.datasets,
+            database_path: args.database_path,
+            nfdump: args.nfdump,
+            poll_seconds: args.poll_seconds,
+            retention_days: args.retention_days,
+            max_per_tail: args.max_per_tail,
+            threshold_high: args.threshold_high,
+            threshold_low: args.threshold_low,
+            backfill: args.backfill,
+            once: args.once,
+        })?,
         Command::ContractVersion => println!("{}", netflow_db::PIPELINE_CONTRACT_VERSION),
     }
     Ok(())
@@ -511,7 +571,22 @@ fn run_web_verify(args: WebVerifyArgs) -> Result<()> {
 }
 
 fn run_maad(args: MaadArgs) -> Result<()> {
-    let input: Box<dyn BufRead> = match args.input {
+    let addresses = read_ipv4_lines(args.input)?;
+    maad::write_json(&maad::compute(addresses), io::stdout().lock())?;
+    io::stdout().flush()?;
+    Ok(())
+}
+
+fn run_singularity(args: SingularityArgs) -> Result<()> {
+    let addresses = read_ipv4_lines(args.input)?;
+    singularity::write_csv(&singularity::score(addresses), io::stdout().lock())?;
+    io::stdout().flush()?;
+    Ok(())
+}
+
+/// Read IPv4 addresses, one per line, from a file or standard input.
+fn read_ipv4_lines(input: Option<PathBuf>) -> Result<Vec<Ipv4Addr>> {
+    let input: Box<dyn BufRead> = match input {
         Some(path) => Box::new(BufReader::new(
             File::open(&path).with_context(|| format!("unable to open {}", path.display()))?,
         )),
@@ -530,9 +605,7 @@ fn run_maad(args: MaadArgs) -> Result<()> {
                 .with_context(|| format!("invalid IPv4 address {value:?}"))?,
         );
     }
-    maad::write_json(&maad::compute(addresses), io::stdout().lock())?;
-    io::stdout().flush()?;
-    Ok(())
+    Ok(addresses)
 }
 
 fn parse_boundary(raw: &str, timezone: &str) -> Result<i64> {
diff --git a/tools/netflow-db/src/singularity.rs b/tools/netflow-db/src/singularity.rs
new file mode 100644
index 00000000..5732a43c
--- /dev/null
+++ b/tools/netflow-db/src/singularity.rs
@@ -0,0 +1,293 @@
+//! Per-address Singularity scoring, a port of MAAD's `Singularities.hs`.
+//!
+//! For each distinct IPv4 address `x`, `alpha(x)` is the OLS slope of
+//! `-log2(mu_l(x) / n)` against prefix length `l`, where `mu_l(x)` counts the
+//! distinct addresses sharing `x`'s `/l` prefix and `n` is the total distinct
+//! address count. Prefix levels stop at the first isolated prefix
+//! (`mu == 1`), matching the reference in `vendor/maad/Singularities.hs`.
+//!
+//! High alpha marks an address in a sparse, isolated region of address
+//! space; low alpha marks one that stays inside a dense cluster across many
+//! prefix levels. Both tails are anomalous.
+
+use std::io;
+use std::net::Ipv4Addr;
+
+/// Fitted singularity exponent for one distinct address.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub struct AddressScore {
+    pub address: Ipv4Addr,
+    pub alpha: f64,
+    pub intercept: f64,
+    pub r_squared: f64,
+    /// Number of prefix levels used in the regression.
+    pub prefix_levels: u8,
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+struct RunningOls {
+    count: f64,
+    sum_x: f64,
+    sum_y: f64,
+    sum_xx: f64,
+    sum_xy: f64,
+    sum_yy: f64,
+}
+
+impl RunningOls {
+    fn add(self, x: f64, y: f64) -> Self {
+        Self {
+            count: self.count + 1.0,
+            sum_x: self.sum_x + x,
+            sum_y: self.sum_y + y,
+            sum_xx: self.sum_xx + x * x,
+            sum_xy: self.sum_xy + x * y,
+            sum_yy: self.sum_yy + y * y,
+        }
+    }
+
+    /// Fits the accumulated points without branches for degenerate inputs.
+    ///
+    /// Zero or one point, and a constant ordinate across two or more points,
+    /// leave at least one zero denominator. IEEE-754 division therefore
+    /// produces `NaN`, matching MAAD's rank-deficient OLS behavior. These
+    /// values are intentional and must not be replaced with fallback scores.
+    fn fit(self) -> Regression {
+        let mean_x = self.sum_x / self.count;
+        let mean_y = self.sum_y / self.count;
+        let sxx = self.sum_xx - self.count * mean_x * mean_x;
+        let sxy = self.sum_xy - self.count * mean_x * mean_y;
+        let syy = self.sum_yy - self.count * mean_y * mean_y;
+        let alpha = sxy / sxx;
+
+        Regression {
+            alpha,
+            intercept: mean_y - alpha * mean_x,
+            r_squared: (sxy * sxy) / (sxx * syy),
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+struct Regression {
+    alpha: f64,
+    intercept: f64,
+    r_squared: f64,
+}
+
+/// Score every distinct address in `addresses` (duplicates are ignored).
+/// Returns scores sorted by ascending alpha, ties broken by address, matching
+/// the reference ordering.
+pub fn score(addresses: Vec<Ipv4Addr>) -> Vec<AddressScore> {
+    let mut addresses: Vec<u32> = addresses.into_iter().map(u32::from).collect();
+    addresses.sort_unstable();
+    addresses.dedup();
+
+    if addresses.is_empty() {
+        return Vec::new();
+    }
+
+    let mut scores = Vec::with_capacity(addresses.len());
+    visit_prefix(
+        &addresses,
+        0,
+        RunningOls::default(),
+        &mut scores,
+        addresses.len() as f64,
+    );
+    scores.sort_by(|left, right| {
+        left.alpha
+            .total_cmp(&right.alpha)
+            .then_with(|| left.address.cmp(&right.address))
+    });
+    scores
+}
+
+fn visit_prefix(
+    addresses: &[u32],
+    level: u8,
+    running: RunningOls,
+    scores: &mut Vec<AddressScore>,
+    total: f64,
+) {
+    if addresses.len() == 1 {
+        let regression = running.fit();
+        scores.push(AddressScore {
+            address: Ipv4Addr::from(addresses[0]),
+            alpha: regression.alpha,
+            intercept: regression.intercept,
+            r_squared: regression.r_squared,
+            prefix_levels: level,
+        });
+        return;
+    }
+
+    let ordinate = -((addresses.len() as f64) / total).log2();
+    let running = running.add(f64::from(level), ordinate);
+
+    if level < 32 {
+        let bit = 31 - u32::from(level);
+        let split = addresses.partition_point(|address| (address >> bit) & 1 == 0);
+
+        if split > 0 {
+            visit_prefix(&addresses[..split], level + 1, running, scores, total);
+        }
+        if split < addresses.len() {
+            visit_prefix(&addresses[split..], level + 1, running, scores, total);
+        }
+    }
+}
+
+/// Write scores as CSV with an `addr,alpha,intercept,r2,n_levels` header.
+pub fn write_csv(scores: &[AddressScore], mut output: impl io::Write) -> io::Result<()> {
+    writeln!(output, "addr,alpha,intercept,r2,n_levels")?;
+    for score in scores {
+        writeln!(
+            output,
+            "{},{},{},{},{}",
+            score.address, score.alpha, score.intercept, score.r_squared, score.prefix_levels
+        )?;
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const EPSILON: f64 = 1e-9;
+
+    #[test]
+    fn scores_hand_verified_prefix_fixture() {
+        let scores = score(vec![
+            Ipv4Addr::new(0, 0, 0, 0),
+            Ipv4Addr::new(32, 0, 0, 0),
+            Ipv4Addr::new(128, 0, 0, 0),
+            Ipv4Addr::new(192, 0, 0, 0),
+        ]);
+
+        let expected = [
+            (Ipv4Addr::new(0, 0, 0, 0), 0.5, 1.0 / 6.0, 0.75, 3),
+            (Ipv4Addr::new(32, 0, 0, 0), 0.5, 1.0 / 6.0, 0.75, 3),
+            (Ipv4Addr::new(128, 0, 0, 0), 1.0, 0.0, 1.0, 2),
+            (Ipv4Addr::new(192, 0, 0, 0), 1.0, 0.0, 1.0, 2),
+        ];
+
+        assert_eq!(scores.len(), expected.len());
+        for (actual, (address, alpha, intercept, r_squared, prefix_levels)) in
+            scores.iter().zip(expected)
+        {
+            assert_eq!(actual.address, address);
+            assert_close(actual.alpha, alpha);
+            assert_close(actual.intercept, intercept);
+            assert_close(actual.r_squared, r_squared);
+            assert_eq!(actual.prefix_levels, prefix_levels);
+        }
+    }
+
+    #[test]
+    fn ignores_duplicate_addresses() {
+        let address = Ipv4Addr::new(10, 0, 0, 1);
+        let duplicate_only = score(vec![address, address]);
+        assert_eq!(duplicate_only.len(), 1);
+        assert_eq!(duplicate_only[0].address, address);
+
+        let deduped = vec![
+            Ipv4Addr::new(0, 0, 0, 0),
+            Ipv4Addr::new(32, 0, 0, 0),
+            Ipv4Addr::new(128, 0, 0, 0),
+            Ipv4Addr::new(192, 0, 0, 0),
+        ];
+        let mut with_duplicate = deduped.clone();
+        with_duplicate.insert(2, Ipv4Addr::new(32, 0, 0, 0));
+
+        assert_eq!(score(with_duplicate), score(deduped));
+    }
+
+    #[test]
+    fn empty_input_has_no_scores() {
+        assert!(score(Vec::new()).is_empty());
+    }
+
+    #[test]
+    fn single_address_has_degenerate_regression() {
+        let address = Ipv4Addr::new(203, 0, 113, 7);
+        let scores = score(vec![address]);
+
+        assert_eq!(scores.len(), 1);
+        assert_eq!(scores[0].address, address);
+        assert_eq!(scores[0].prefix_levels, 0);
+        assert!(scores[0].alpha.is_nan());
+        assert!(scores[0].intercept.is_nan());
+        assert!(scores[0].r_squared.is_nan());
+    }
+
+    #[test]
+    fn sorts_three_alpha_groups_before_breaking_ties_by_address() {
+        let scores = score(vec![
+            Ipv4Addr::new(192, 0, 0, 0),
+            Ipv4Addr::new(16, 0, 0, 0),
+            Ipv4Addr::new(128, 0, 0, 0),
+            Ipv4Addr::new(0, 0, 0, 0),
+            Ipv4Addr::new(32, 0, 0, 0),
+        ]);
+
+        assert_eq!(
+            scores.iter().map(|entry| entry.address).collect::<Vec<_>>(),
+            vec![
+                Ipv4Addr::new(32, 0, 0, 0),
+                Ipv4Addr::new(0, 0, 0, 0),
+                Ipv4Addr::new(16, 0, 0, 0),
+                Ipv4Addr::new(128, 0, 0, 0),
+                Ipv4Addr::new(192, 0, 0, 0),
+            ]
+        );
+        assert!(scores[0].alpha < scores[1].alpha);
+        assert_close(scores[1].alpha, scores[2].alpha);
+        assert!(scores[2].alpha < scores[3].alpha);
+        assert_close(scores[3].alpha, scores[4].alpha);
+    }
+
+    #[test]
+    fn writes_scores_as_csv() {
+        let scores = score(vec![
+            Ipv4Addr::new(0, 0, 0, 0),
+            Ipv4Addr::new(32, 0, 0, 0),
+            Ipv4Addr::new(128, 0, 0, 0),
+            Ipv4Addr::new(192, 0, 0, 0),
+        ]);
+        let mut csv = Vec::new();
+
+        write_csv(&scores, &mut csv).unwrap();
+
+        let csv = String::from_utf8(csv).unwrap();
+        let mut lines = csv.lines();
+        assert_eq!(lines.next(), Some("addr,alpha,intercept,r2,n_levels"));
+
+        let expected = [
+            ("0.0.0.0", 0.5, 1.0 / 6.0, 0.75, 3),
+            ("32.0.0.0", 0.5, 1.0 / 6.0, 0.75, 3),
+            ("128.0.0.0", 1.0, 0.0, 1.0, 2),
+            ("192.0.0.0", 1.0, 0.0, 1.0, 2),
+        ];
+        for (line, (address, alpha, intercept, r_squared, prefix_levels)) in
+            lines.by_ref().zip(expected)
+        {
+            let fields: Vec<_> = line.split(',').collect();
+            assert_eq!(fields.len(), 5);
+            assert_eq!(fields[0], address);
+            assert_close(fields[1].parse().unwrap(), alpha);
+            assert_close(fields[2].parse().unwrap(), intercept);
+            assert_close(fields[3].parse().unwrap(), r_squared);
+            assert_eq!(fields[4].parse::<u8>().unwrap(), prefix_levels);
+        }
+        assert_eq!(lines.next(), None);
+    }
+
+    fn assert_close(actual: f64, expected: f64) {
+        assert!(
+            (actual - expected).abs() <= EPSILON,
+            "actual={actual:?}, expected={expected:?}"
+        );
+    }
+}
diff --git a/tools/netflow-db/tests/singularity_cli.rs b/tools/netflow-db/tests/singularity_cli.rs
new file mode 100644
index 00000000..966624cd
--- /dev/null
+++ b/tools/netflow-db/tests/singularity_cli.rs
@@ -0,0 +1,44 @@
+use std::{fs, process::Command};
+
+const EPSILON: f64 = 1e-9;
+const FIXTURE: &str = "0.0.0.0\n32.0.0.0\n128.0.0.0\n192.0.0.0\n";
+
+#[test]
+fn singularity_file_input_emits_scores_as_csv() {
+    let directory = tempfile::tempdir().unwrap();
+    let input = directory.path().join("addresses.txt");
+    fs::write(&input, FIXTURE).unwrap();
+
+    let output = Command::new(env!("CARGO_BIN_EXE_netflow-db"))
+        .args(["singularity"])
+        .arg(&input)
+        .output()
+        .unwrap();
+    assert!(
+        output.status.success(),
+        "stdout={}\nstderr={}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+
+    let stdout = String::from_utf8(output.stdout).unwrap();
+    let mut lines = stdout.lines();
+    assert_eq!(lines.next(), Some("addr,alpha,intercept,r2,n_levels"));
+
+    let rows: Vec<_> = lines.collect();
+    assert_eq!(rows.len(), 4);
+    let first: Vec<_> = rows[0].split(',').collect();
+    assert_eq!(first.len(), 5);
+    assert_eq!(first[0], "0.0.0.0");
+    assert_close(first[1].parse().unwrap(), 0.5);
+    assert_close(first[2].parse().unwrap(), 1.0 / 6.0);
+    assert_close(first[3].parse().unwrap(), 0.75);
+    assert_eq!(first[4], "3");
+}
+
+fn assert_close(actual: f64, expected: f64) {
+    assert!(
+        (actual - expected).abs() <= EPSILON,
+        "actual={actual:?}, expected={expected:?}"
+    );
+}