diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f8a07ad43f..04975d2c09 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -469,7 +469,7 @@ jobs:
               run: bun run --cwd lib/effect-clickhouse-http test
 
             # Migration replay and search indexes, raw-vs-rollup parity for web
-            # analytics and the service map (a divergence renders as a plausible wrong number), the
+            # analytics, the service map and SLOs (a divergence renders as a plausible wrong number), the
             # cascaded trace facets rollup (one that never fires reads as "no data"), and the users
             # directory's group view.
             - name: Replay migrations and verify rollups
@@ -482,6 +482,7 @@ jobs:
                   src/services/warehouse/ci-runs.clickhouse.e2e.test.ts
                   src/services/warehouse/service-api-materialization.clickhouse.e2e.test.ts
                   src/services/warehouse/service-map-parity.clickhouse.e2e.test.ts
+                  src/services/warehouse/slo-parity.clickhouse.e2e.test.ts
                   src/services/users/users-groups.clickhouse.e2e.test.ts
 
             # The fingerprint the error views compute, against the TS mirror every
diff --git a/apps/api/scripts/bench-suites/slos.ts b/apps/api/scripts/bench-suites/slos.ts
new file mode 100644
index 0000000000..1f8505e068
--- /dev/null
+++ b/apps/api/scripts/bench-suites/slos.ts
@@ -0,0 +1,87 @@
+/**
+ * SLO reads (SlosService) against a populated snapshot, plus the flat scan the rollup splice
+ * replaces. Set BENCH_ORG and BENCH_SERVICE to a busy service; windows end at BENCH_END
+ * (`YYYY-MM-DD HH:MM:SS`, default now) so the splice edges are partial minutes and hours.
+ */
+import { Effect, Array as Arr } from "effect"
+import * as CH from "@maple/query-engine/ch"
+import * as E from "@maple-dev/effect-orm/expr"
+import * as Bench from "@maple-dev/effect-orm/benchmark"
+import { OrgId } from "@maple/domain"
+import type { SloSli } from "@maple/domain/slo"
+import { formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/query-engine/datetime"
+
+const ORG = OrgId.make(process.env.BENCH_ORG ?? "org_sql_catalog")
+const SERVICE = process.env.BENCH_SERVICE ?? "api"
+const DAY_MS = 86_400_000
+const HOUR_MS = 3_600_000
+
+const END_MS = process.env.BENCH_END ? parseWarehouseDateTime(process.env.BENCH_END) : Date.now()
+const window = (spanMs: number) => ({
+	orgId: ORG,
+	startTime: formatWarehouseDateTime(END_MS - spanMs),
+	endTime: formatWarehouseDateTime(END_MS),
+})
+
+const AVAILABILITY: SloSli = { _tag: "availability" }
+const LATENCY_500: SloSli = { _tag: "latency", thresholdMs: 500 }
+const opts = (sli: SloSli) => ({ serviceName: SERVICE, environments: [], sli })
+
+const totals = (id: string, sli: SloSli) =>
+	Bench.query({
+		id,
+		inputs: {
+			...window(30 * DAY_MS),
+			sli: sli._tag === "latency" ? `latency-${sli.thresholdMs}` : sli._tag,
+		},
+		compile: ({ orgId, startTime, endTime }) =>
+			CH.compile(CH.sloTotalsQuery(opts(sli)), { orgId, startTime, endTime }),
+		results: "unordered",
+	})
+
+const buckets = (id: string, grain: "hour" | "minute", bucketSeconds: number, spanMs: number) =>
+	Bench.query({
+		id,
+		inputs: { ...window(spanMs), bucketSeconds },
+		compile: (inputs) => CH.compile(CH.sloBucketsQuery(opts(AVAILABILITY), grain), inputs),
+		results: "ordered",
+	})
+
+/** What the splice replaces: one scan of every entry-point span in the window. */
+const flatScan = Bench.query({
+	id: "baseline/30d/flat-scan-availability",
+	inputs: window(30 * DAY_MS),
+	compile: (inputs) =>
+		CH.compile(
+			CH.from(CH.tables.ServiceOverviewSpans)
+				.select(($) => ({
+					total: E.sum($.SampleRate),
+					good: E.sumIf($.SampleRate, $.StatusCode.neq("Error")),
+				}))
+				.where(($) => [
+					$.OrgId.eq(CH.tables.orgIdParam),
+					$.ServiceName.eq(SERVICE),
+					$.Timestamp.gte(CH.tables.utcSecondsParam("startTime")),
+					$.Timestamp.lte(CH.tables.utcSecondsParam("endTime")),
+				])
+				.format("JSON"),
+			inputs,
+		),
+	results: "unordered",
+})
+
+export default Bench.defineSuite({
+	name: "slos",
+	dataset: process.env.BENCH_DATASET ?? "replace-with-your-snapshot-revision",
+	cases: Arr.map(
+		[
+			totals("totals/30d/availability", AVAILABILITY),
+			totals("totals/30d/latency-500ms", LATENCY_500),
+			buckets("burn/6h/minute-60s", "minute", 60, 6 * HOUR_MS),
+			buckets("history/60d/hour-3600s", "hour", 3600, 60 * DAY_MS),
+			flatScan,
+		],
+		// `defineSuite` takes one error type; a compile failure is a broken suite.
+		(sample) => Effect.orDie(sample),
+	),
+})
diff --git a/apps/api/src/routes/v2/slos.http.test.ts b/apps/api/src/routes/v2/slos.http.test.ts
index db999144db..afaf014ce7 100644
--- a/apps/api/src/routes/v2/slos.http.test.ts
+++ b/apps/api/src/routes/v2/slos.http.test.ts
@@ -14,6 +14,13 @@ import { DashboardPersistenceService } from "@maple/backend/services/dashboards/
 import { SharedDashboardService } from "@maple/backend/services/dashboards/SharedDashboardService"
 import { LiveActivitiesService } from "@maple/backend/services/push/LiveActivitiesService"
 import { MobileDevicesService } from "@maple/backend/services/push/MobileDevicesService"
+import {
+	WarehouseQueryService,
+	type WarehouseQueryServiceApi,
+} from "@maple/backend/services/warehouse/WarehouseQueryService"
+import { makeWarehouseServiceStub } from "@maple/backend/testing/warehouse-test-support"
+import { compiledQueryOf } from "@maple/query-engine/execution"
+import { formatWarehouseDateTime } from "@maple/query-engine"
 import { V2TransportErrorBoundaryLive } from "./error-envelope"
 import {
 	AlertsServiceStubLayer,
@@ -50,7 +57,18 @@ const testConfig = () =>
 		}),
 	)
 
-const makeHarness = () => {
+/** A warehouse whose SLO reads answer from fixed rows, decoded like real ones. */
+const sloWarehouse = (
+	rowsByContext: Record<string, ReadonlyArray<Record<string, unknown>>>,
+): WarehouseQueryServiceApi =>
+	makeWarehouseServiceStub({
+		compiledQuery: (_tenant, compiled, options) =>
+			compiledQueryOf(compiled)
+				.decodeRows(rowsByContext[options?.context ?? ""] ?? [])
+				.pipe(Effect.orDie),
+	})
+
+const makeHarness = (warehouse: WarehouseQueryServiceApi = makeWarehouseServiceStub()) => {
 	const testDb = createTestDb(createdDbs)
 	const envLive = Env.layer.pipe(Layer.provide(testConfig()))
 	const servicesLive = Layer.mergeAll(
@@ -64,6 +82,8 @@ const makeHarness = () => {
 
 	const routes = HttpApiBuilder.layer(MapleApiV2).pipe(
 		Layer.provide(AllV2GroupLayersLive),
+		// Ahead of the stub bundles, which carry an inert warehouse of their own.
+		Layer.provide(Layer.succeed(WarehouseQueryService, warehouse)),
 		Layer.provide(V2TransportErrorBoundaryLive),
 		Layer.provide(AlertsServiceStubLayer),
 		Layer.provide(ConfigResourceServiceStubsLayer),
@@ -116,10 +136,13 @@ const makeHarness = () => {
 	}
 }
 
-const withHarness = (body: (harness: ReturnType<typeof makeHarness>) => Promise<void>) =>
+const withHarness = (
+	body: (harness: ReturnType<typeof makeHarness>) => Promise<void>,
+	warehouse?: WarehouseQueryServiceApi,
+) =>
 	Effect.runPromise(
 		Effect.acquireUseRelease(
-			Effect.sync(makeHarness),
+			Effect.sync(() => makeHarness(warehouse)),
 			(harness) => Effect.promise(() => body(harness)),
 			(harness) => Effect.promise(() => harness.dispose()),
 		),
@@ -250,4 +273,105 @@ describe("v2 slos", () => {
 			const { secret: writer } = await harness.bootstrapKey(undefined, ["slos:write"])
 			expect((await harness.request("PATCH", path, writer, { target_percent: 99 })).status).toBe(200)
 		}))
+
+	it("reports the SLI, budget left and burn windows from the warehouse", () => {
+		const nowMinute = Math.floor(Date.now() / 60_000) * 60_000
+		// 2% of every minute is bad: 20x the 0.1% budget in every window.
+		const minutes = Arr.makeBy(400, (index) => ({
+			bucket: formatWarehouseDateTime(nowMinute - (index - 1) * 60_000),
+			total: 100,
+			good: 98,
+		}))
+		const warehouse = sloWarehouse({
+			sloWindowTotals: [{ total: 10_000, good: 9_995 }],
+			sloBurnWindows: minutes,
+		})
+		return withHarness(async (harness) => {
+			const { secret: token } = await harness.bootstrapKey()
+			const created = await harness.request("POST", "/v2/slos", token, checkoutSlo)
+			const status = await harness.request("GET", `/v2/slos/${created.body.id}/status`, token)
+			expect(status.status).toBe(200)
+			expect(status.body).toMatchObject({ object: "slo.status", slo_id: created.body.id })
+			expect(status.body.total_events).toBe(10_000)
+			expect(status.body.sli).toBeCloseTo(0.9995)
+			expect(status.body.budget_remaining).toBeCloseTo(0.5)
+			expect(
+				Arr.map(status.body.burn_windows, (window: { window_minutes: number; burn_rate: number }) => [
+					window.window_minutes,
+					Math.round(window.burn_rate),
+				]),
+			).toEqual([
+				[5, 20],
+				[30, 20],
+				[60, 20],
+				[360, 20],
+			])
+
+			// An SLO that is gone, or another org's, is a 404 on both reads.
+			const gone = "slo_YofPTrK9782DWwcnXhpcCw"
+			const missing = await Promise.all([
+				harness.request("GET", `/v2/slos/${gone}/status`, token),
+				harness.request("GET", `/v2/slos/${gone}/history`, token),
+			])
+			expect(Arr.map(missing, (response) => [response.status, response.body.error.code])).toEqual([
+				[404, "slo_not_found"],
+				[404, "slo_not_found"],
+			])
+
+			const history = await harness.request("GET", `/v2/slos/${created.body.id}/history`, token)
+			expect(history.status).toBe(200)
+			// 30 days of hourly points, untouched since the stub has no hourly rows.
+			expect(history.body.points).toHaveLength(720)
+			expect(history.body.points[0].budget_remaining).toBe(1)
+		}, warehouse)
+	})
+
+	it("reads every SLO's status in one batch, counting only each SLO's service and environments", () => {
+		const burn = (total: number, good: number) => ({
+			total5: total,
+			good5: good,
+			total30: total * 6,
+			good30: good * 6,
+			total60: total * 12,
+			good60: good * 12,
+			total360: total * 72,
+			good360: good * 72,
+		})
+		const warehouse = sloWarehouse({
+			sloStatusTotals: [
+				{ serviceName: "checkout", environment: "production", total: 10_000, good: 9_995 },
+				// Outside the checkout SLO's environments: counting it would spend the whole budget.
+				{ serviceName: "checkout", environment: "staging", total: 500, good: 0 },
+				{ serviceName: "search", environment: "production", total: 2_000, good: 2_000 },
+			],
+			sloStatusBurn: [
+				{ serviceName: "checkout", environment: "production", ...burn(100, 98) },
+				{ serviceName: "checkout", environment: "staging", ...burn(100, 0) },
+				{ serviceName: "search", environment: "production", ...burn(100, 100) },
+			],
+		})
+		return withHarness(async (harness) => {
+			const { secret: token } = await harness.bootstrapKey()
+			const checkout = await harness.request("POST", "/v2/slos", token, checkoutSlo)
+			const search = await harness.request("POST", "/v2/slos", token, {
+				name: "Search latency",
+				service_name: "search",
+				sli: { type: "latency", threshold_ms: 500 },
+				target_percent: 99,
+			})
+			const listed = await harness.request("GET", "/v2/slos/statuses", token)
+			expect(listed.status).toBe(200)
+			expect(listed.body.object).toBe("slo.status_list")
+			const byId = new Map(
+				Arr.map(listed.body.data, (status: { slo_id: string }) => [status.slo_id, status] as const),
+			)
+			const checkoutStatus = byId.get(checkout.body.id)
+			expect(checkoutStatus.total_events).toBe(10_000)
+			expect(checkoutStatus.budget_remaining).toBeCloseTo(0.5)
+			expect(Math.round(checkoutStatus.burn_windows[0].burn_rate)).toBe(20)
+			const searchStatus = byId.get(search.body.id)
+			expect(searchStatus.sli).toBe(1)
+			expect(searchStatus.burn_windows[3].burn_rate).toBe(0)
+		}, warehouse)
+	})
 })
diff --git a/apps/api/src/routes/v2/slos.http.ts b/apps/api/src/routes/v2/slos.http.ts
index 8a99a051dc..42846ffec9 100644
--- a/apps/api/src/routes/v2/slos.http.ts
+++ b/apps/api/src/routes/v2/slos.http.ts
@@ -8,12 +8,12 @@ import {
 	sloSliToWire,
 	V2InsufficientPermissions,
 } from "@maple/domain/http/v2"
-import type { V2Slo } from "@maple/domain/http/v2"
+import type { V2Slo, V2SloStatus } from "@maple/domain/http/v2"
 import { Effect, Array as Arr } from "effect"
 import { auditDiff } from "@/routes/v2/audit-changes"
 import { recordHttpAudit } from "@maple/backend/services/audit/AuditLogService"
 import * as Policy from "@maple/domain/policy"
-import { type SloRecord, SlosService } from "@maple/backend/services/slos/SlosService"
+import { type SloRecord, type SloStatus, SlosService } from "@maple/backend/services/slos/SlosService"
 
 // An SLO owns a burn-rate alert rule, so writes need the same permission as alert rules.
 
@@ -32,6 +32,22 @@ export const toV2Slo = (slo: SloRecord): V2Slo => ({
 	updated_at: isoTimestamp(slo.updatedAt),
 })
 
+const toV2Status = (sloId: SloRecord["id"], status: SloStatus): V2SloStatus => ({
+	object: "slo.status",
+	slo_id: sloId,
+	as_of: isoTimestamp(status.asOfMs),
+	total_events: status.total,
+	good_events: status.good,
+	sli: status.sli,
+	budget_remaining: status.budgetRemaining,
+	burn_windows: Arr.map(status.burnWindows, (window) => ({
+		window_minutes: window.windowMinutes,
+		burn_rate: window.burnRate,
+		total_events: window.total,
+		good_events: window.good,
+	})),
+})
+
 const sloAuditDiff = auditDiff({
 	fields: ["name", "description", "service_name", "environments", "sli", "target_percent", "window_days"],
 })
@@ -122,6 +138,41 @@ export const HttpV2SlosLive = HttpApiBuilder.group(MapleApiV2, "slos", (handlers
 					return toV2Slo(updated)
 				}),
 			)
+			.handle("status", ({ params }) =>
+				Effect.gen(function* () {
+					const tenant = yield* CurrentTenant.Context
+					const status = yield* service.status(tenant, params.id)
+					return toV2Status(params.id, status)
+				}),
+			)
+			.handle("statuses", () =>
+				Effect.gen(function* () {
+					const tenant = yield* CurrentTenant.Context
+					const rows = yield* service.statuses(tenant)
+					return {
+						object: "slo.status_list" as const,
+						data: Arr.map(rows, (row) => toV2Status(row.slo.id, row.status)),
+					}
+				}),
+			)
+			.handle("history", ({ params }) =>
+				Effect.gen(function* () {
+					const tenant = yield* CurrentTenant.Context
+					const history = yield* service.history(tenant, params.id)
+					return {
+						object: "slo.history" as const,
+						slo_id: params.id,
+						as_of: isoTimestamp(history.asOfMs),
+						bucket_seconds: history.bucketSeconds,
+						points: Arr.map(history.points, (point) => ({
+							at: isoTimestamp(point.atMs),
+							sli: point.sli,
+							budget_remaining: point.budgetRemaining,
+							burn_rate: point.burnRate,
+						})),
+					}
+				}),
+			)
 			.handle("delete", ({ params }) =>
 				Effect.gen(function* () {
 					const tenant = yield* requireManage
diff --git a/apps/api/src/routes/v2/v2-test-support.ts b/apps/api/src/routes/v2/v2-test-support.ts
index 07bb2cc910..06a6ba6b14 100644
--- a/apps/api/src/routes/v2/v2-test-support.ts
+++ b/apps/api/src/routes/v2/v2-test-support.ts
@@ -194,8 +194,8 @@ const v2GroupLayersExceptOnboardingChecklist = (
 		HttpV2AuditLogLive.pipe(Layer.provide(Layer.merge(auditLog, OrgMembersServiceStubLayer))),
 		HttpV2ScrapeTargetsLive,
 		HttpV2UptimeMonitorsLive.pipe(Layer.provide(uptimeMonitors)),
-		// Real service: it needs only the Database every harness already provides.
-		HttpV2SlosLive.pipe(Layer.provide(SlosService.layer)),
+		// Real service over whichever `WarehouseQueryService` stub the test provides.
+		HttpV2SlosLive.pipe(Layer.provide(Layer.effect(SlosService, SlosService.make))),
 		// Real service: it needs only the Database every harness already provides.
 		Layer.mergeAll(HttpV2StatusPagesLive, HttpV2StatusPageIncidentsLive).pipe(
 			Layer.provide(
diff --git a/packages/backend/src/services/slos/SlosService.ts b/packages/backend/src/services/slos/SlosService.ts
index d28cc0a7b3..0afa69cf0e 100644
--- a/packages/backend/src/services/slos/SlosService.ts
+++ b/packages/backend/src/services/slos/SlosService.ts
@@ -1,13 +1,28 @@
 import { randomUUID } from "node:crypto"
 import * as PG from "@maple-dev/effect-orm/postgres"
 import { Slos, type SloRow } from "@maple/db/tables"
-import { SloNotFoundError, SloPersistenceError, SloValidationError } from "@maple/domain/http"
+import {
+	SloNotFoundError,
+	SloPersistenceError,
+	SloValidationError,
+	type WarehouseReadError,
+} from "@maple/domain/http"
 import { type AlertRuleId, type OrgId, SloId, SloWindowDays, type UserId } from "@maple/domain/primitives"
-import { SloSli, SloTargetPercent } from "@maple/domain/slo"
+import {
+	budgetRemaining,
+	sliRatio,
+	SloTargetPercent,
+	type SloBurnWindow,
+	type SloHistoryPoint,
+	type SloSli,
+} from "@maple/domain/slo"
 import { Array as Arr, Clock, Context, Effect, Layer, Schema, String as Str } from "effect"
 import type { MapleDb } from "@maple/db/client"
 import { Database } from "@maple/backend/platform/DatabaseLive"
 import { makeDbExecute } from "@maple/backend/platform/db-execute"
+import type { TenantContext } from "@maple/backend/services/auth/AuthService"
+import { WarehouseQueryService } from "@maple/backend/services/warehouse/WarehouseQueryService"
+import { readBudgetHistory, readBurnWindows, readStatuses, readWindowTotals } from "./slo-reads"
 
 export interface SloRecord {
 	readonly id: SloId
@@ -48,6 +63,26 @@ export interface UpdateSloInput {
 
 type SloError = SloNotFoundError | SloValidationError | SloPersistenceError
 
+export interface SloStatus {
+	/** The read's end, snapped to the minute. */
+	readonly asOfMs: number
+	readonly total: number
+	readonly good: number
+	/** Good share over the SLO window, `null` without events. */
+	readonly sli: number | null
+	/** Share of the window's budget left: 1 untouched, negative when overspent. */
+	readonly budgetRemaining: number
+	readonly burnWindows: ReadonlyArray<SloBurnWindow>
+}
+
+export interface SloHistory {
+	readonly asOfMs: number
+	readonly bucketSeconds: number
+	readonly points: ReadonlyArray<SloHistoryPoint>
+}
+
+type SloReadError = SloNotFoundError | SloPersistenceError | WarehouseReadError
+
 export interface SlosServiceApi {
 	readonly list: (orgId: OrgId) => Effect.Effect<ReadonlyArray<SloRecord>, SloPersistenceError>
 	readonly get: (
@@ -63,6 +98,17 @@ export interface SlosServiceApi {
 		orgId: OrgId,
 		sloId: SloId,
 	) => Effect.Effect<SloId, SloNotFoundError | SloPersistenceError>
+	/** Where the SLO stands now: its SLI, budget left and each burn window. */
+	readonly status: (tenant: TenantContext, sloId: SloId) => Effect.Effect<SloStatus, SloReadError>
+	/** Every SLO's status at once, for the SLOs list: a handful of reads, not two per SLO. */
+	readonly statuses: (
+		tenant: TenantContext,
+	) => Effect.Effect<
+		ReadonlyArray<{ readonly slo: SloRecord; readonly status: SloStatus }>,
+		SloPersistenceError | WarehouseReadError
+	>
+	/** The budget chart: hourly points over the SLO window. */
+	readonly history: (tenant: TenantContext, sloId: SloId) => Effect.Effect<SloHistory, SloReadError>
 }
 
 export const DEFAULT_SLO_WINDOW_DAYS: SloWindowDays = 30
@@ -142,6 +188,9 @@ export class SlosService extends Context.Service<SlosService, SlosServiceApi>()(
 		make: Effect.gen(function* () {
 			const database = yield* Database
 			const dbExecute = makeDbExecute(database, "SlosService", toPersistenceError)
+			const warehouse = yield* WarehouseQueryService
+			const withWarehouse = Effect.provideService(WarehouseQueryService, warehouse)
+			const nowMinute = Effect.map(Clock.currentTimeMillis, (now) => Math.floor(now / 60_000) * 60_000)
 
 			const selectOne = (orgId: OrgId, sloId: SloId) =>
 				dbExecute((db) =>
@@ -282,9 +331,66 @@ export class SlosService extends Context.Service<SlosService, SlosServiceApi>()(
 				return sloId
 			})
 
-			return { list, get, create, update, delete: remove } satisfies SlosServiceApi
+			const status = Effect.fn("SlosService.status")(function* (tenant: TenantContext, sloId: SloId) {
+				yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId, "maple.slo.id": sloId })
+				const slo = yield* selectOne(tenant.orgId, sloId)
+				const asOfMs = yield* nowMinute
+				const [totals, windows] = yield* Effect.all(
+					[readWindowTotals(tenant, slo, asOfMs), readBurnWindows(tenant, slo, asOfMs)],
+					{ concurrency: 2 },
+				).pipe(withWarehouse)
+				return {
+					asOfMs,
+					total: totals.total,
+					good: totals.good,
+					sli: sliRatio(totals),
+					budgetRemaining: budgetRemaining(totals, slo.targetPercent),
+					burnWindows: windows,
+				} satisfies SloStatus
+			})
+
+			const statuses = Effect.fn("SlosService.statuses")(function* (tenant: TenantContext) {
+				yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId })
+				const slos = yield* list(tenant.orgId)
+				const asOfMs = yield* nowMinute
+				const read = yield* readStatuses(tenant, slos, asOfMs).pipe(withWarehouse)
+				return Arr.map(slos, (slo) => {
+					const found = read.get(slo.id)
+					const totals = found?.totals ?? { total: 0, good: 0 }
+					return {
+						slo,
+						status: {
+							asOfMs,
+							total: totals.total,
+							good: totals.good,
+							sli: sliRatio(totals),
+							budgetRemaining: budgetRemaining(totals, slo.targetPercent),
+							burnWindows: found?.windows ?? [],
+						} satisfies SloStatus,
+					}
+				})
+			})
+
+			const history = Effect.fn("SlosService.history")(function* (tenant: TenantContext, sloId: SloId) {
+				yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId, "maple.slo.id": sloId })
+				const slo = yield* selectOne(tenant.orgId, sloId)
+				const asOfMs = yield* nowMinute
+				const points = yield* readBudgetHistory(tenant, slo, asOfMs).pipe(withWarehouse)
+				return { asOfMs, bucketSeconds: 3600, points } satisfies SloHistory
+			})
+
+			return {
+				list,
+				get,
+				create,
+				update,
+				delete: remove,
+				status,
+				statuses,
+				history,
+			} satisfies SlosServiceApi
 		}),
 	},
 ) {
-	static readonly layer = Layer.effect(this, this.make)
+	static readonly layer = Layer.effect(this, this.make).pipe(Layer.provide(WarehouseQueryService.layer))
 }
diff --git a/packages/backend/src/services/slos/slo-reads.ts b/packages/backend/src/services/slos/slo-reads.ts
new file mode 100644
index 0000000000..4e0dfda470
--- /dev/null
+++ b/packages/backend/src/services/slos/slo-reads.ts
@@ -0,0 +1,242 @@
+// Warehouse reads behind an SLO: the window's totals, the burn windows and the budget chart.
+// Shared by the SLO API and the burn-rate alert evaluator.
+import { CH, formatWarehouseDateTime } from "@maple/query-engine"
+import {
+	budgetHistory,
+	burnRate,
+	burnWindows,
+	SLO_BURN_WINDOWS_MINUTES,
+	SLO_BURN_LOOKBACK_MINUTES,
+	type SloBucket,
+	type SloBurnWindow,
+	type SloCounts,
+	type SloHistoryPoint,
+	type SloSli,
+} from "@maple/domain/slo"
+import type { OrgId } from "@maple/domain/primitives"
+import { Array as Arr, DateTime, Effect, Record as Rec } from "effect"
+import type { TenantContext } from "@maple/backend/services/auth/AuthService"
+import { WarehouseQueryService } from "@maple/backend/services/warehouse/WarehouseQueryService"
+
+const MINUTE_MS = 60_000
+const HOUR_MS = 60 * MINUTE_MS
+const DAY_MS = 24 * HOUR_MS
+
+/** What an SLO read needs from the stored SLO. */
+export interface SloTarget {
+	readonly orgId: OrgId
+	readonly serviceName: string
+	readonly environments: ReadonlyArray<string>
+	readonly sli: SloSli
+	readonly targetPercent: number
+	readonly windowDays: number
+}
+
+const queryOpts = (slo: SloTarget) => ({
+	serviceName: slo.serviceName,
+	environments: slo.environments,
+	sli: slo.sli,
+})
+
+const toBuckets = (rows: ReadonlyArray<CH.SloBucketsOutput>): ReadonlyArray<SloBucket> =>
+	Arr.map(rows, (row) => ({
+		startMs: DateTime.toEpochMillis(row.bucket),
+		total: row.total,
+		good: row.good,
+	}))
+
+/** Good and total events over the SLO window ending at `endMs`. */
+export const readWindowTotals = Effect.fn("SloReads.windowTotals")(function* (
+	tenant: TenantContext,
+	slo: SloTarget,
+	endMs: number,
+) {
+	const warehouse = yield* WarehouseQueryService
+	const rows = yield* warehouse.compiledQuery(
+		tenant,
+		CH.compile(CH.sloTotalsQuery(queryOpts(slo)), {
+			orgId: slo.orgId,
+			startTime: formatWarehouseDateTime(endMs - slo.windowDays * DAY_MS),
+			endTime: formatWarehouseDateTime(endMs),
+		}),
+		{ profile: "aggregation", context: "sloWindowTotals" },
+	)
+	const row = rows[0]
+	return { total: row?.total ?? 0, good: row?.good ?? 0 } satisfies SloCounts
+})
+
+/** The burn of each alert window (5m to 6h) ending at `endMs`, from one minute-grain read. */
+export const readBurnWindows = Effect.fn("SloReads.burnWindows")(function* (
+	tenant: TenantContext,
+	slo: SloTarget,
+	endMs: number,
+) {
+	const warehouse = yield* WarehouseQueryService
+	const rows = yield* warehouse.compiledQuery(
+		tenant,
+		CH.compile(CH.sloBucketsQuery(queryOpts(slo), "minute"), {
+			orgId: slo.orgId,
+			startTime: formatWarehouseDateTime(endMs - SLO_BURN_LOOKBACK_MINUTES * MINUTE_MS),
+			endTime: formatWarehouseDateTime(endMs),
+			bucketSeconds: 60,
+		}),
+		{ profile: "aggregation", context: "sloBurnWindows" },
+	)
+	return burnWindows(toBuckets(rows), endMs, slo.targetPercent) satisfies ReadonlyArray<SloBurnWindow>
+})
+
+/** Hourly points over the SLO window ending at `endMs`; reads two windows of hourly buckets. */
+export const readBudgetHistory = Effect.fn("SloReads.budgetHistory")(function* (
+	tenant: TenantContext,
+	slo: SloTarget,
+	endMs: number,
+) {
+	const warehouse = yield* WarehouseQueryService
+	const windowMs = slo.windowDays * DAY_MS
+	const rows = yield* warehouse.compiledQuery(
+		tenant,
+		CH.compile(CH.sloBucketsQuery(queryOpts(slo), "hour"), {
+			orgId: slo.orgId,
+			startTime: formatWarehouseDateTime(Math.ceil(endMs / HOUR_MS) * HOUR_MS - 2 * windowMs),
+			endTime: formatWarehouseDateTime(endMs),
+			bucketSeconds: 3600,
+		}),
+		{ profile: "aggregation", context: "sloBudgetHistory" },
+	)
+	return budgetHistory({
+		buckets: toBuckets(rows),
+		bucketMs: HOUR_MS,
+		windowMs,
+		endMs,
+		targetPercent: slo.targetPercent,
+	}) satisfies ReadonlyArray<SloHistoryPoint>
+})
+
+/** An SLO the batch read covers, keyed by its id. */
+export interface SloBatchTarget extends SloTarget {
+	readonly id: string
+}
+
+export interface SloBatchStatus {
+	readonly totals: SloCounts
+	readonly windows: ReadonlyArray<SloBurnWindow>
+}
+
+const sliKey = (sli: SloSli) => (sli._tag === "availability" ? "availability" : `latency:${sli.thresholdMs}`)
+
+/** The rows of one (service, environment) breakdown that an SLO counts. */
+const counted = <Row extends { readonly serviceName: string; readonly environment: string }>(
+	rows: ReadonlyArray<Row>,
+	slo: SloTarget,
+) =>
+	Arr.filter(
+		rows,
+		(row) =>
+			row.serviceName === slo.serviceName &&
+			(slo.environments.length === 0 || Arr.contains(slo.environments, row.environment)),
+	)
+
+/**
+ * Every SLO's window totals and burn windows ending at `endMs`, in one read per (SLI, window)
+ * and one per SLI, each broken down by service and environment and summed per SLO here.
+ */
+export const readStatuses = Effect.fn("SloReads.statuses")(function* (
+	tenant: TenantContext,
+	slos: ReadonlyArray<SloBatchTarget>,
+	endMs: number,
+) {
+	const warehouse = yield* WarehouseQueryService
+	const orgId = tenant.orgId
+	const services = (group: ReadonlyArray<SloBatchTarget>) =>
+		Arr.dedupe(Arr.map(group, (slo) => slo.serviceName))
+	const bySli = Arr.groupBy(slos, (slo) => sliKey(slo.sli))
+	const byWindow = Arr.groupBy(slos, (slo) => `${sliKey(slo.sli)}|${slo.windowDays}`)
+	yield* Effect.annotateCurrentSpan({
+		"maple.slo.count": slos.length,
+		"maple.slo.reads": Rec.keys(bySli).length + Rec.keys(byWindow).length,
+	})
+
+	const totalsReads = Effect.forEach(
+		Rec.values(byWindow),
+		(group) => {
+			const first = Arr.headNonEmpty(group)
+			return warehouse
+				.compiledQuery(
+					tenant,
+					CH.compile(
+						CH.sloTotalsByServiceQuery({ serviceNames: services(group), sli: first.sli }),
+						{
+							orgId,
+							startTime: formatWarehouseDateTime(endMs - first.windowDays * DAY_MS),
+							endTime: formatWarehouseDateTime(endMs),
+						},
+					),
+					{ profile: "aggregation", context: "sloStatusTotals" },
+				)
+				.pipe(
+					Effect.map((rows) =>
+						Arr.map(group, (slo) => [slo.id, sumCounts(counted(rows, slo))] as const),
+					),
+				)
+		},
+		{ concurrency: 4 },
+	)
+	const burnReads = Effect.forEach(
+		Rec.values(bySli),
+		(group) => {
+			const first = Arr.headNonEmpty(group)
+			return warehouse
+				.compiledQuery(
+					tenant,
+					CH.compile(CH.sloBurnByServiceQuery({ serviceNames: services(group), sli: first.sli }), {
+						orgId,
+						startTime: formatWarehouseDateTime(endMs - SLO_BURN_LOOKBACK_MINUTES * MINUTE_MS),
+						endTime: formatWarehouseDateTime(endMs),
+					}),
+					{ profile: "aggregation", context: "sloStatusBurn" },
+				)
+				.pipe(
+					Effect.map((rows) =>
+						Arr.map(
+							group,
+							(slo) => [slo.id, burnWindowsOf(counted(rows, slo), slo.targetPercent)] as const,
+						),
+					),
+				)
+		},
+		{ concurrency: 4 },
+	)
+	const [totals, burns] = yield* Effect.all([totalsReads, burnReads], { concurrency: 2 })
+	const totalsById = new Map<string, SloCounts>(Arr.flatten(totals))
+	const burnsById = new Map<string, ReadonlyArray<SloBurnWindow>>(Arr.flatten(burns))
+	return new Map(
+		Arr.map(slos, (slo): readonly [string, SloBatchStatus] => [
+			slo.id,
+			{
+				totals: totalsById.get(slo.id) ?? { total: 0, good: 0 },
+				windows: burnsById.get(slo.id) ?? burnWindowsOf([], slo.targetPercent),
+			},
+		]),
+	)
+})
+
+const sumCounts = (rows: ReadonlyArray<SloCounts>): SloCounts =>
+	Arr.reduce(rows, { total: 0, good: 0 }, (acc, row) => ({
+		total: acc.total + row.total,
+		good: acc.good + row.good,
+	}))
+
+/** One SLO's four burn windows from its breakdown rows. */
+const burnWindowsOf = (
+	rows: ReadonlyArray<CH.SloBurnByServiceOutput>,
+	targetPercent: number,
+): ReadonlyArray<SloBurnWindow> =>
+	Arr.map(SLO_BURN_WINDOWS_MINUTES, (windowMinutes) => {
+		const counts = sumCounts(
+			Arr.map(rows, (row) => ({
+				total: row[`total${windowMinutes}`],
+				good: row[`good${windowMinutes}`],
+			})),
+		)
+		return { windowMinutes, ...counts, burnRate: burnRate(counts, targetPercent) }
+	})
diff --git a/packages/backend/src/services/warehouse/slo-parity.clickhouse.e2e.test.ts b/packages/backend/src/services/warehouse/slo-parity.clickhouse.e2e.test.ts
new file mode 100644
index 0000000000..9338eea43c
--- /dev/null
+++ b/packages/backend/src/services/warehouse/slo-parity.clickhouse.e2e.test.ts
@@ -0,0 +1,448 @@
+// SAFETY-FILE: JSON in this test is emitted by the fixture or unit under test before its fields are asserted.
+// SLO splice parity: `sloTotalsQuery` and `sloBucketsQuery` against a flat scan.
+//
+// The SLO reads tile a window from raw `service_overview_spans` (partial minutes),
+// `service_overview_minutely` (partial hours) and `service_overview_hourly` (the rest).
+// Spans sit exactly on every seam of several windows, aligned and not, so a boundary
+// that drifts by one bucket double-counts or drops a whole cluster of spans.
+
+import { afterAll, assert, beforeAll, describe, it } from "@effect/vitest"
+import type { SloSli } from "@maple/domain/slo"
+import * as CH from "@maple/query-engine/ch"
+import { normalizeSqlForClickHouseClient } from "@maple/query-engine/execution"
+import { Array as Arr, Record as Rec, Schema, String as Str } from "effect"
+import {
+	applyRealMigrations,
+	clickhouseE2eEnabled,
+	clickhouseExec,
+	uniqueDatabase,
+} from "./clickhouse-e2e-support"
+
+const database = uniqueDatabase("maple_slo_parity_e2e")
+const ORG_ID = "org_slo_parity"
+const LEGACY_ORG_ID = "org_slo_parity_pre_ladder"
+
+const DAY_MS = 86_400_000
+const HOUR_MS = 3_600_000
+const MINUTE_MS = 60_000
+const SECOND_MS = 1000
+
+/** Midnight three days back: inside the 30-day TTL of `traces` and `service_overview_spans`. */
+const BASE_MS = Math.floor((Date.now() - 3 * DAY_MS) / DAY_MS) * DAY_MS
+const at = (h: number, m: number, s = 0): number => BASE_MS + h * HOUR_MS + m * MINUTE_MS + s * SECOND_MS
+
+const chDateTime = (epochMs: number): string =>
+	Str.slice(0, 19)(Str.replace("T", " ")(new Date(epochMs).toISOString()))
+
+interface Window {
+	readonly label: string
+	readonly startMs: number
+	readonly endMs: number
+}
+
+const WINDOWS: ReadonlyArray<Window> = [
+	{ label: "unaligned 10:30:30-14:15:30", startMs: at(10, 30, 30), endMs: at(14, 15, 30) },
+	{ label: "hour-aligned 10:00-14:00", startMs: at(10, 0), endMs: at(14, 0) },
+	{ label: "minute-aligned 10:30-14:15", startMs: at(10, 30), endMs: at(14, 15) },
+	{ label: "hour-aligned start, unaligned end 10:00-14:15:30", startMs: at(10, 0), endMs: at(14, 15, 30) },
+	{ label: "unaligned start, hour-aligned end 10:30:30-14:00", startMs: at(10, 30, 30), endMs: at(14, 0) },
+	{ label: "sub-hour 10:30:30-10:45:30", startMs: at(10, 30, 30), endMs: at(10, 45, 30) },
+]
+
+/** Every seam of every window above, plus the second on each side of each window bound. */
+const SEAMS_MS: ReadonlyArray<number> = [
+	at(9, 59, 59),
+	at(10, 0),
+	at(10, 0, 1),
+	at(10, 0, 30),
+	at(10, 29, 59),
+	at(10, 30),
+	at(10, 30, 1),
+	at(10, 30, 29),
+	at(10, 30, 30),
+	at(10, 30, 59),
+	at(10, 31),
+	at(10, 45),
+	at(10, 45, 29),
+	at(10, 45, 30),
+	at(10, 45, 31),
+	at(10, 59, 59),
+	at(11, 0),
+	at(11, 59, 59),
+	at(12, 0),
+	at(13, 59, 59),
+	at(14, 0),
+	at(14, 0, 1),
+	at(14, 14, 59),
+	at(14, 15),
+	at(14, 15, 1),
+	at(14, 15, 29),
+	at(14, 15, 30),
+	at(14, 15, 31),
+]
+
+interface SeedSpan {
+	readonly org: string
+	readonly ms: number
+	readonly service: string
+	readonly environment: string
+	readonly kind: "Server" | "Internal"
+	readonly parent: string
+	readonly durationNs: number
+	readonly status: "Ok" | "Error"
+	readonly sampleRate: number
+}
+
+const span = (ms: number, overrides: Partial<Omit<SeedSpan, "ms">> = {}): SeedSpan => ({
+	org: ORG_ID,
+	ms,
+	service: "api",
+	environment: "production",
+	kind: "Server",
+	parent: "",
+	durationNs: 120_000_000,
+	status: "Ok",
+	sampleRate: 1,
+	...overrides,
+})
+
+/** One cluster per seam: both sides of the 500ms rung, errors, sampled rows, and decoys. */
+const seamCluster = (ms: number): ReadonlyArray<SeedSpan> => [
+	span(ms, { durationNs: 499_999_999 }),
+	span(ms, { durationNs: 500_000_000, sampleRate: 2 }),
+	span(ms, { durationNs: 100_000_000, status: "Error", sampleRate: 3 }),
+	span(ms, { durationNs: 1_200_000_000, status: "Error" }),
+	span(ms, { environment: "staging", durationNs: 50_000_000, sampleRate: 5 }),
+	span(ms, { environment: "staging", durationNs: 900_000_000, status: "Error", sampleRate: 7 }),
+	span(ms, { service: "worker", durationNs: 2_000_000_000, status: "Error" }),
+	// Not an entry point: the views never see it.
+	span(ms, { kind: "Internal", parent: "parent-span", status: "Error", sampleRate: 11 }),
+]
+
+const SEED_SPANS: ReadonlyArray<SeedSpan> = [
+	...Arr.flatMap(SEAMS_MS, seamCluster),
+	// Interior volume every 7 minutes from 09:00 to 15:30, so every tier aggregates real rows.
+	...Arr.map(Arr.range(0, 55), (i) =>
+		span(at(9, 0) + i * 7 * MINUTE_MS + 13 * SECOND_MS, {
+			durationNs: (40 + i * 23) * 1_000_000,
+			status: i % 5 === 0 ? "Error" : "Ok",
+			sampleRate: i % 3 === 0 ? 10 : 1,
+			environment: i % 4 === 0 ? "staging" : "production",
+		}),
+	),
+	// The pre-ladder org's real spans, read beside its directly inserted legacy rollup rows.
+	...Arr.map([at(10, 30, 45), at(10, 50), at(12, 20), at(13, 40), at(14, 15, 10)], (ms, i) =>
+		span(ms, {
+			org: LEGACY_ORG_ID,
+			durationNs: i % 2 === 0 ? 200_000_000 : 800_000_000,
+			status: i === 3 ? "Error" : "Ok",
+			sampleRate: 2,
+		}),
+	),
+]
+
+/** A ClickHouse string literal: backslashes and quotes escaped in one pass. */
+const quote = (value: string): string => `'${Str.replace(/[\\']/g, (char: string) => `\\${char}`)(value)}'`
+
+const seed = async (): Promise<void> => {
+	const rows = Arr.join(
+		Arr.map(SEED_SPANS, (row, index) => {
+			const resourceAttributes = `map('deployment.environment', ${quote(row.environment)}, 'service.namespace', 'ns', 'vcs.ref.head.revision', 'abc123')`
+			return `(${quote(row.org)}, ${quote(chDateTime(row.ms))}, ${quote(`trace-${index}`)}, ${quote(`span-${index}`)}, ${quote(row.parent)}, 'GET /x', ${quote(row.kind)}, ${quote(row.service)}, ${row.durationNs}, ${quote(row.status)}, ${row.sampleRate}, ${resourceAttributes})`
+		}),
+		",\n",
+	)
+	await clickhouseExec(
+		`INSERT INTO traces
+		 (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, SampleRate, ResourceAttributes)
+		 VALUES\n${rows}`,
+		database,
+	)
+}
+
+/** Rows as written before migration 0058: no ladder columns, so they default to 0. */
+const LEGACY_ROWS = [
+	{ table: "service_overview_hourly", time: "Hour", ms: at(12, 0), estimated: 50, errors: 5 },
+	// The leading partial hour of the unaligned window: the minutely tier owns it.
+	{ table: "service_overview_minutely", time: "Minute", ms: at(10, 40), estimated: 30, errors: 3 },
+	// Inside the hourly interior: a whole-window read must take the hour row, not this one.
+	{ table: "service_overview_minutely", time: "Minute", ms: at(12, 10), estimated: 1000, errors: 0 },
+] as const
+
+const seedLegacy = async (): Promise<void> => {
+	for (const row of LEGACY_ROWS) {
+		await clickhouseExec(
+			`INSERT INTO ${row.table} (OrgId, ${row.time}, ServiceName, DeploymentEnv, ServiceNamespace, CommitSha, SpanCount, EstimatedSpanCount, ErrorCount, EstimatedErrorCount, DurationSum, FirstSeen, ApdexSatisfiedCount, ApdexToleratingCount)
+			 VALUES (${quote(LEGACY_ORG_ID)}, ${quote(chDateTime(row.ms))}, 'api', 'production', 'ns', 'abc123', ${row.estimated}, ${row.estimated}, ${row.errors}, ${row.errors}, 1000000, ${quote(chDateTime(row.ms))}, 0, 0)`,
+			database,
+		)
+	}
+}
+
+const runJson = async (sql: string): Promise<ReadonlyArray<Record<string, unknown>>> => {
+	const body = await clickhouseExec(normalizeSqlForClickHouseClient(sql), database, {
+		default_format: "JSON",
+		output_format_json_quote_64bit_integers: "0",
+	})
+	const parsed = JSON.parse(body) as { readonly data?: ReadonlyArray<Record<string, unknown>> }
+	return parsed.data ?? []
+}
+
+const num = (value: unknown): number => Number(value ?? 0)
+
+interface Case {
+	readonly label: string
+	readonly sli: SloSli
+	readonly environments: readonly string[]
+}
+
+const SLIS: ReadonlyArray<{ readonly label: string; readonly sli: SloSli }> = [
+	{ label: "availability", sli: { _tag: "availability" } },
+	{ label: "latency 500ms", sli: { _tag: "latency", thresholdMs: 500 } },
+	{ label: "latency 250ms", sli: { _tag: "latency", thresholdMs: 250 } },
+]
+
+const ENVIRONMENTS: ReadonlyArray<readonly string[]> = [[], ["production"], ["production", "staging"]]
+
+const goodPredicate = (sli: SloSli): string =>
+	sli._tag === "availability" ? "StatusCode != 'Error'" : `Duration < ${sli.thresholdMs * 1_000_000}`
+
+/** Ground truth: one scan of the per-span projection the rollups derive from, no splice. */
+const flatSql = (orgId: string, window: Window, c: Case, bucketSeconds: number | undefined): string => `
+	SELECT
+		${bucketSeconds === undefined ? "'all'" : `toString(toStartOfInterval(toDateTime(Timestamp), INTERVAL ${bucketSeconds} SECOND))`} AS bucket,
+		toFloat64(sum(SampleRate)) AS total,
+		toFloat64(sumIf(SampleRate, ${goodPredicate(c.sli)})) AS good
+	FROM service_overview_spans
+	WHERE OrgId = ${quote(orgId)}
+		AND ServiceName = 'api'
+		AND Timestamp >= ${quote(chDateTime(window.startMs))}
+		AND Timestamp <= ${quote(chDateTime(window.endMs))}
+		${c.environments.length > 0 ? `AND DeploymentEnv IN (${Arr.join(Arr.map(c.environments, quote), ", ")})` : ""}
+	GROUP BY bucket
+	ORDER BY bucket
+`
+
+const params = (orgId: string, window: Window) => ({
+	orgId,
+	startTime: chDateTime(window.startMs),
+	endTime: chDateTime(window.endMs),
+})
+
+const decodeTotals = Schema.decodeUnknownSync(CH.sloTotalsRowSchema)
+
+const splicedTotals = async (orgId: string, window: Window, c: Case) => {
+	const rows = await runJson(
+		CH.compileUnsafe(
+			CH.sloTotalsQuery({ serviceName: "api", environments: c.environments, sli: c.sli }),
+			params(orgId, window),
+		).sql,
+	)
+	assert.lengthOf(rows, 1, "sloTotalsQuery returns exactly one row")
+	return decodeTotals(rows[0])
+}
+
+const flatTotals = async (orgId: string, window: Window, c: Case) => {
+	const rows = await runJson(flatSql(orgId, window, c, undefined))
+	return { total: num(rows[0]?.total), good: num(rows[0]?.good) }
+}
+
+/** Buckets keyed by their start; empty buckets carry no information on either side. */
+const byBucket = (rows: ReadonlyArray<Record<string, unknown>>) =>
+	Rec.fromEntries(
+		Arr.map(
+			Arr.filter(rows, (row) => num(row.total) !== 0 || num(row.good) !== 0),
+			(row): readonly [string, { readonly total: number; readonly good: number }] => [
+				Str.slice(0, 19)(String(row.bucket)),
+				{ total: num(row.total), good: num(row.good) },
+			],
+		),
+	)
+
+const BUCKET_GRAINS = [
+	{ grain: "hour", bucketSeconds: 3600 },
+	{ grain: "hour", bucketSeconds: 7200 },
+	{ grain: "minute", bucketSeconds: 60 },
+	{ grain: "minute", bucketSeconds: 300 },
+] as const
+
+describe.skipIf(!clickhouseE2eEnabled)("SLO raw-vs-rollup parity", () => {
+	beforeAll(async () => {
+		await clickhouseExec(`CREATE DATABASE ${database}`)
+		await applyRealMigrations(database)
+		await seed()
+		await seedLegacy()
+	}, 180_000)
+
+	afterAll(async () => {
+		await clickhouseExec(`DROP DATABASE IF EXISTS ${database}`)
+	}, 30_000)
+
+	it("lands the seed in the projection and both rollups, with a cluster on every seam", async () => {
+		const entryPoints = Arr.filter(SEED_SPANS, (row) => row.org === ORG_ID && row.kind === "Server")
+		const weight = Arr.reduce(entryPoints, 0, (sum, row) => sum + row.sampleRate)
+		const where = `WHERE OrgId = ${quote(ORG_ID)}`
+		const [spans, minutely, hourly] = await Promise.all([
+			runJson(
+				`SELECT count() AS n, toFloat64(sum(SampleRate)) AS w FROM service_overview_spans ${where}`,
+			),
+			runJson(
+				`SELECT toFloat64(sum(EstimatedSpanCount)) AS w, toFloat64(sum(EstimatedLadderSpanCount)) AS c FROM service_overview_minutely ${where}`,
+			),
+			runJson(
+				`SELECT toFloat64(sum(EstimatedSpanCount)) AS w, toFloat64(sum(EstimatedLadderSpanCount)) AS c FROM service_overview_hourly ${where}`,
+			),
+		])
+		assert.strictEqual(num(spans[0]?.n), entryPoints.length, "check the seed against the 30-day TTL")
+		assert.strictEqual(num(spans[0]?.w), weight)
+		for (const rollup of [minutely[0], hourly[0]]) {
+			assert.strictEqual(num(rollup?.w), weight)
+			assert.strictEqual(num(rollup?.c), weight)
+		}
+		for (const seam of SEAMS_MS) {
+			const rows = await runJson(
+				`SELECT count() AS n FROM service_overview_spans ${where} AND Timestamp = ${quote(chDateTime(seam))}`,
+			)
+			assert.isAbove(num(rows[0]?.n), 0, `no span sits on ${chDateTime(seam)}`)
+		}
+	})
+
+	// `for` loops, not `describe.each`: the latter OOMs tsc in this repo.
+	for (const window of WINDOWS) {
+		for (const { label, sli } of SLIS) {
+			for (const environments of ENVIRONMENTS) {
+				const c: Case = { label, sli, environments }
+				const name = `${window.label}, ${label}, env=[${Arr.join(environments, ",")}]`
+
+				it(`totals match a flat scan: ${name}`, async () => {
+					const [spliced, truth] = await Promise.all([
+						splicedTotals(ORG_ID, window, c),
+						flatTotals(ORG_ID, window, c),
+					])
+					assert.isAbove(truth.total, 0, "the flat scan found nothing: the seed missed the window")
+					assert.isAbove(truth.total, truth.good, "every span is good: the SLI cannot fail")
+					assert.isAbove(truth.good, 0, "no span is good")
+					assert.deepStrictEqual(spliced, truth)
+				})
+			}
+		}
+
+		// The SLOs tab's batch reads: one row per (service, environment), summed per SLO.
+		for (const { label, sli } of SLIS) {
+			it(`grouped totals sum to each SLO's totals: ${window.label}, ${label}`, async () => {
+				const rows = await runJson(
+					CH.compileUnsafe(
+						CH.sloTotalsByServiceQuery({ serviceNames: ["api", "worker"], sli }),
+						params(ORG_ID, window),
+					).sql,
+				)
+				for (const environments of ENVIRONMENTS) {
+					const counted = Arr.filter(
+						rows,
+						(row) =>
+							row.serviceName === "api" &&
+							(environments.length === 0 ||
+								Arr.contains(environments, String(row.environment))),
+					)
+					const summed = Arr.reduce(counted, { total: 0, good: 0 }, (sum, row) => ({
+						total: sum.total + num(row.total),
+						good: sum.good + num(row.good),
+					}))
+					assert.deepStrictEqual(
+						summed,
+						await flatTotals(ORG_ID, window, { label, sli, environments }),
+					)
+				}
+			})
+
+			it(`grouped burn windows match flat scans: ${window.label}, ${label}`, async () => {
+				const rows = await runJson(
+					CH.compileUnsafe(
+						CH.sloBurnByServiceQuery({ serviceNames: ["api", "worker"], sli }),
+						params(ORG_ID, window),
+					).sql,
+				)
+				const api = Arr.filter(rows, (row) => row.serviceName === "api")
+				for (const minutes of [5, 30, 60, 360]) {
+					const truth = await runJson(`
+						SELECT toFloat64(sum(SampleRate)) AS total, toFloat64(sumIf(SampleRate, ${goodPredicate(sli)})) AS good
+						FROM service_overview_spans
+						WHERE OrgId = ${quote(ORG_ID)} AND ServiceName = 'api'
+							AND Timestamp >= ${quote(chDateTime(window.startMs))}
+							AND Timestamp <= ${quote(chDateTime(window.endMs))}
+							AND toStartOfMinute(Timestamp) >= toDateTime(${quote(chDateTime(window.endMs))}) - INTERVAL ${minutes} MINUTE
+							AND toStartOfMinute(Timestamp) < toDateTime(${quote(chDateTime(window.endMs))})
+					`)
+					const summed = Arr.reduce(api, { total: 0, good: 0 }, (sum, row) => ({
+						total: sum.total + num(row[`total${minutes}`]),
+						good: sum.good + num(row[`good${minutes}`]),
+					}))
+					assert.deepStrictEqual(
+						summed,
+						{ total: num(truth[0]?.total), good: num(truth[0]?.good) },
+						`${minutes}m window`,
+					)
+				}
+			})
+		}
+
+		for (const { grain, bucketSeconds } of BUCKET_GRAINS) {
+			for (const { label, sli } of Arr.take(SLIS, 2)) {
+				for (const environments of Arr.take(ENVIRONMENTS, 2)) {
+					const c: Case = { label, sli, environments }
+					const name = `${window.label}, ${grain} ${bucketSeconds}s, ${label}, env=[${Arr.join(environments, ",")}]`
+
+					it(`buckets match per-bucket flat scans and sum to the totals: ${name}`, async () => {
+						const [spliced, truth, totals] = await Promise.all([
+							runJson(
+								CH.compileUnsafe(
+									CH.sloBucketsQuery(
+										{ serviceName: "api", environments: c.environments, sli: c.sli },
+										grain,
+									),
+									{ ...params(ORG_ID, window), bucketSeconds },
+								).sql,
+							),
+							runJson(flatSql(ORG_ID, window, c, bucketSeconds)),
+							splicedTotals(ORG_ID, window, c),
+						])
+						assert.isAbove(truth.length, 0)
+						assert.deepStrictEqual(byBucket(spliced), byBucket(truth))
+						assert.deepStrictEqual(
+							Arr.reduce(spliced, { total: 0, good: 0 }, (sum, row) => ({
+								total: sum.total + num(row.total),
+								good: sum.good + num(row.good),
+							})),
+							totals,
+						)
+					})
+				}
+			}
+		}
+	}
+
+	it("drops pre-ladder rollup rows from a latency SLI and still counts them for availability", async () => {
+		const window = Arr.getUnsafe(WINDOWS, 0)
+		const availability: Case = { label: "availability", sli: { _tag: "availability" }, environments: [] }
+		const latency: Case = {
+			label: "latency",
+			sli: { _tag: "latency", thresholdMs: 500 },
+			environments: [],
+		}
+		const [availabilitySpliced, availabilityFlat, latencySpliced, latencyFlat] = await Promise.all([
+			splicedTotals(LEGACY_ORG_ID, window, availability),
+			flatTotals(LEGACY_ORG_ID, window, availability),
+			splicedTotals(LEGACY_ORG_ID, window, latency),
+			flatTotals(LEGACY_ORG_ID, window, latency),
+		])
+		assert.isAbove(latencyFlat.total, 0)
+		// Legacy rows have zero coverage: the latency read is exactly the real spans.
+		assert.deepStrictEqual(latencySpliced, latencyFlat)
+		// The whole-window read takes the 12:00 hour row and the 10:40 minute row; the 12:10
+		// minute row sits inside the hourly interior, where the minutely tier does not read.
+		assert.deepStrictEqual(availabilitySpliced, {
+			total: availabilityFlat.total + 50 + 30,
+			good: availabilityFlat.good + (50 - 5) + (30 - 3),
+		})
+	})
+})
diff --git a/packages/domain/src/http/v2/openapi.test.ts b/packages/domain/src/http/v2/openapi.test.ts
index fae301198d..b000ac60c9 100644
--- a/packages/domain/src/http/v2/openapi.test.ts
+++ b/packages/domain/src/http/v2/openapi.test.ts
@@ -176,7 +176,10 @@ describe("MapleApiV2 OpenAPI", () => {
 			"GET /v2/session_replays/{id}/manifest",
 			"GET /v2/session_replays/{id}/transcript",
 			"GET /v2/slos",
+			"GET /v2/slos/statuses",
 			"GET /v2/slos/{id}",
+			"GET /v2/slos/{id}/history",
+			"GET /v2/slos/{id}/status",
 			"GET /v2/sources",
 			"GET /v2/sources/usage",
 			"GET /v2/sources/{id}",
diff --git a/packages/domain/src/http/v2/slos.ts b/packages/domain/src/http/v2/slos.ts
index 3b9895c4af..b9fc38e38a 100644
--- a/packages/domain/src/http/v2/slos.ts
+++ b/packages/domain/src/http/v2/slos.ts
@@ -1,11 +1,12 @@
 import { HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/http-api"
 import { Schema } from "effect"
 import { SloId, SloLatencyThresholdMs, SloWindowDays } from "../../primitives"
-import { SloTargetPercent, type SloSli } from "../../slo"
+import { SLO_BURN_WINDOWS_MINUTES, SloTargetPercent, type SloSli } from "../../slo"
 import { SloNotFoundError, SloPersistenceError, SloValidationError } from "../slos"
 import { AuthorizationV2 } from "./auth"
 import { wireExample, ListOf, ListQuery, Timestamp } from "./envelopes"
 import { V2InsufficientPermissions, V2ParameterInvalid } from "./errors"
+import { V2WarehouseReadErrors } from "./query-errors"
 import { publicErrors } from "./public-error"
 import { PublicId, PublicIdPrefixes } from "./public-id"
 import { AlertRulePublicId } from "./resource-ids"
@@ -162,6 +163,73 @@ export const V2SloDeleteResponse = Schema.Struct({
 })
 export type V2SloDeleteResponse = Schema.Schema.Type<typeof V2SloDeleteResponse>
 
+const ratioField = (description: string) => Schema.Number.annotate({ description })
+
+export const V2SloBurnWindow = Schema.Struct({
+	window_minutes: Schema.Literals(SLO_BURN_WINDOWS_MINUTES).annotate({
+		description: "The window's length in minutes, ending at `as_of`.",
+	}),
+	burn_rate: ratioField(
+		"How fast the window spends the error budget: 1 spends exactly the budget over the SLO window.",
+	),
+	total_events: ratioField("Sampling-weighted entry-point spans in the window."),
+	good_events: ratioField("The good ones among them."),
+}).annotate({ identifier: "SloBurnWindow", title: "SLO burn window" })
+
+export const V2SloStatus = Schema.Struct({
+	object: Schema.Literal("slo.status"),
+	slo_id: SloPublicId,
+	as_of: Timestamp.annotate({ description: "When the status was read, to the minute." }),
+	total_events: ratioField("Sampling-weighted entry-point spans over the SLO window."),
+	good_events: ratioField("The good ones among them."),
+	sli: Schema.NullOr(Schema.Number).annotate({
+		description: "Share of good events over the SLO window (0 to 1), `null` when there were none.",
+	}),
+	budget_remaining: ratioField(
+		"Share of the window's error budget left: 1 untouched, 0 spent, negative when overspent.",
+	),
+	burn_windows: Schema.Array(V2SloBurnWindow).annotate({
+		description: "The windows the burn-rate alert reads: 5 minutes, 30 minutes, 1 hour and 6 hours.",
+	}),
+}).annotate({
+	identifier: "SloStatus",
+	title: "SLO status",
+	description: "Where an SLO stands now: its SLI, the budget left and how fast it burns.",
+})
+export type V2SloStatus = Schema.Schema.Type<typeof V2SloStatus>
+
+export const V2SloStatusList = Schema.Struct({
+	object: Schema.Literal("slo.status_list"),
+	data: Schema.Array(V2SloStatus).annotate({ description: "One status per SLO, in list order." }),
+}).annotate({
+	identifier: "SloStatusList",
+	title: "SLO status list",
+	description: "Every SLO's status at once, read together rather than one request per SLO.",
+})
+export type V2SloStatusList = Schema.Schema.Type<typeof V2SloStatusList>
+
+export const V2SloHistoryPoint = Schema.Struct({
+	at: Timestamp,
+	sli: Schema.NullOr(Schema.Number).annotate({
+		description: "Share of good events in the bucket ending here, `null` when there were none.",
+	}),
+	budget_remaining: ratioField("Budget left over the SLO window ending here."),
+	burn_rate: ratioField("Burn rate of the bucket ending here."),
+}).annotate({ identifier: "SloHistoryPoint", title: "SLO history point" })
+
+export const V2SloHistory = Schema.Struct({
+	object: Schema.Literal("slo.history"),
+	slo_id: SloPublicId,
+	as_of: Timestamp,
+	bucket_seconds: Schema.Number.annotate({ description: "Width of each point's bucket." }),
+	points: Schema.Array(V2SloHistoryPoint).annotate({ description: "Oldest first, over the SLO window." }),
+}).annotate({
+	identifier: "SloHistory",
+	title: "SLO history",
+	description: "The error budget over the SLO window, one point per hour.",
+})
+export type V2SloHistory = Schema.Schema.Type<typeof V2SloHistory>
+
 const [sloNotFound, sloValidation, sloPersistence] = publicErrors(
 	SloNotFoundError,
 	SloValidationError,
@@ -189,6 +257,19 @@ export class V2SlosApiGroup extends HttpApiGroup.make("slos")
 			}),
 		),
 	)
+	.add(
+		HttpApiEndpoint.get("statuses", "/statuses", {
+			success: V2SloStatusList,
+			error: [sloPersistence, ...V2WarehouseReadErrors],
+		}).annotateMerge(
+			OpenApi.annotations({
+				identifier: "listSloStatuses",
+				summary: "List every SLO's status",
+				description:
+					"Returns the SLI, budget left and burn windows of every SLO in the organization, read together.",
+			}),
+		),
+	)
 	.add(
 		HttpApiEndpoint.post("create", "/", {
 			payload: V2SloCreateParams,
@@ -244,6 +325,33 @@ export class V2SlosApiGroup extends HttpApiGroup.make("slos")
 			}),
 		),
 	)
+	.add(
+		HttpApiEndpoint.get("status", "/:id/status", {
+			params: { id: SloPublicId },
+			success: V2SloStatus,
+			error: [sloNotFound, sloPersistence, ...V2WarehouseReadErrors],
+		}).annotateMerge(
+			OpenApi.annotations({
+				identifier: "getSloStatus",
+				summary: "Retrieve an SLO's status",
+				description:
+					"Returns the SLO's SLI and error budget over its window, and its current burn rates.",
+			}),
+		),
+	)
+	.add(
+		HttpApiEndpoint.get("history", "/:id/history", {
+			params: { id: SloPublicId },
+			success: V2SloHistory,
+			error: [sloNotFound, sloPersistence, ...V2WarehouseReadErrors],
+		}).annotateMerge(
+			OpenApi.annotations({
+				identifier: "getSloHistory",
+				summary: "Retrieve an SLO's budget history",
+				description: "Returns the error budget left over the SLO window, one point per hour.",
+			}),
+		),
+	)
 	.prefix("/v2/slos")
 	.middleware(AuthorizationV2)
 	.annotateMerge(
diff --git a/packages/domain/src/slo.test.ts b/packages/domain/src/slo.test.ts
new file mode 100644
index 0000000000..42bb0b3400
--- /dev/null
+++ b/packages/domain/src/slo.test.ts
@@ -0,0 +1,120 @@
+import { describe, expect, it } from "vitest"
+import { Array as Arr, Option } from "effect"
+import {
+	budgetHistory,
+	budgetRemaining,
+	burnRate,
+	burnWindows,
+	SLO_BURN_TIERS,
+	sliRatio,
+	tierBreached,
+	type SloBucket,
+} from "./slo"
+
+const MINUTE = 60_000
+const HOUR = 60 * MINUTE
+const [critical, warning] = SLO_BURN_TIERS
+
+describe("SLO math", () => {
+	it("reads a 99.9% target's budget from the bad share", () => {
+		// 2 bad in 1000 at 99.9%: twice the allowed rate, the whole budget and as much again.
+		expect(burnRate({ total: 1000, good: 998 }, 99.9)).toBeCloseTo(2)
+		expect(budgetRemaining({ total: 1000, good: 998 }, 99.9)).toBeCloseTo(-1)
+		expect(budgetRemaining({ total: 10_000, good: 9_995 }, 99.9)).toBeCloseTo(0.5)
+		expect(sliRatio({ total: 1000, good: 998 })).toBeCloseTo(0.998)
+	})
+
+	it("treats a window without events as untouched", () => {
+		expect(sliRatio({ total: 0, good: 0 })).toBeNull()
+		expect(burnRate({ total: 0, good: 0 }, 99.9)).toBe(0)
+		expect(budgetRemaining({ total: 0, good: 0 }, 99.9)).toBe(1)
+	})
+})
+
+/** One bucket per minute over the last `minutes` before `endMs`, `bad` of 1000 bad in each. */
+const minuteBuckets = (endMs: number, minutes: number, bad: (minutesAgo: number) => number) =>
+	Arr.makeBy(minutes, (index): SloBucket => {
+		const minutesAgo = index + 1
+		return { startMs: endMs - minutesAgo * MINUTE, total: 1000, good: 1000 - bad(minutesAgo) }
+	})
+
+describe("burn windows and tiers", () => {
+	const endMs = Date.UTC(2026, 10, 6, 12, 0)
+
+	it("pages on a fast burn across both the 1h and 5m windows", () => {
+		// 2% errors for the last hour at 99.9% burns 20x.
+		const windows = burnWindows(
+			minuteBuckets(endMs, 360, (ago) => (ago <= 60 ? 20 : 0)),
+			endMs,
+			99.9,
+		)
+		expect(
+			Arr.map(windows, (window) => [window.windowMinutes, Math.round(window.burnRate * 10) / 10]),
+		).toEqual([
+			[5, 20],
+			[30, 20],
+			[60, 20],
+			[360, 3.3],
+		])
+		expect(tierBreached(critical, windows)).toBe(true)
+		expect(tierBreached(warning, windows)).toBe(false)
+	})
+
+	it("stops paging once the short window recovers, though the hour still burns", () => {
+		const windows = burnWindows(
+			minuteBuckets(endMs, 360, (ago) => (ago > 5 && ago <= 60 ? 20 : 0)),
+			endMs,
+			99.9,
+		)
+		expect(tierBreached(critical, windows)).toBe(false)
+	})
+
+	it("tickets a slow burn over 6h and 30m", () => {
+		// 0.8% errors for 6h: 8x, under the page factor, over the ticket factor.
+		const windows = burnWindows(
+			minuteBuckets(endMs, 360, () => 8),
+			endMs,
+			99.9,
+		)
+		expect(tierBreached(critical, windows)).toBe(false)
+		expect(tierBreached(warning, windows)).toBe(true)
+	})
+
+	it("ignores buckets outside the windows", () => {
+		const windows = burnWindows(
+			[...minuteBuckets(endMs, 360, () => 0), { startMs: endMs, total: 1000, good: 0 }],
+			endMs,
+			99.9,
+		)
+		expect(Arr.every(windows, (window) => window.burnRate === 0)).toBe(true)
+	})
+})
+
+describe("budgetHistory", () => {
+	it("reads the trailing window at each hour, ending on the partial current hour", () => {
+		const windowMs = 24 * HOUR
+		const endMs = Date.UTC(2026, 10, 6, 12, 30)
+		// Two days of hourly buckets; one bad hour 30h ago (in the first window only) and one 2h ago.
+		const firstBucket = Date.UTC(2026, 10, 4, 13)
+		const buckets = Arr.makeBy(48, (index): SloBucket => {
+			const startMs = firstBucket + index * HOUR
+			const bad = startMs === Date.UTC(2026, 10, 5, 6) || startMs === Date.UTC(2026, 10, 6, 10) ? 50 : 0
+			return { startMs, total: 1000, good: 1000 - bad }
+		})
+		const points = budgetHistory({ buckets, bucketMs: HOUR, windowMs, endMs, targetPercent: 99 })
+
+		expect(points).toHaveLength(24)
+		expect(points[points.length - 1]?.atMs).toBe(endMs)
+		expect(points[0]?.atMs).toBe(Date.UTC(2026, 10, 5, 14))
+		// 24 000 events allow 240 bad at 99%. The window ending at 14:00 on day 5 still holds
+		// the 06:00 bad hour: 50/240 spent.
+		expect(points[0]?.budgetRemaining).toBeCloseTo(1 - 50 / 240)
+		// The latest window holds only the 10:00 bad hour.
+		expect(points[points.length - 1]?.budgetRemaining).toBeCloseTo(1 - 50 / 240)
+		// The bad hour's own point burns 5% / 1% = 5x.
+		const badPoint = Arr.findFirst(points, (point) => point.atMs === Date.UTC(2026, 10, 6, 11))
+		expect(Option.map(badPoint, (point) => [Math.round(point.burnRate * 100) / 100, point.sli])).toEqual(
+			Option.some([5, 0.95]),
+		)
+	})
+})
diff --git a/packages/domain/src/slo.ts b/packages/domain/src/slo.ts
index 9c73471185..b88344fcda 100644
--- a/packages/domain/src/slo.ts
+++ b/packages/domain/src/slo.ts
@@ -1,4 +1,4 @@
-import { Schema } from "effect"
+import { Array as Arr, Option, Schema } from "effect"
 import { SloLatencyThresholdMs } from "./primitives"
 
 // Service level objectives: what counts as a good event (the SLI), the target share of good
@@ -20,3 +20,143 @@ export type SloTargetPercent = Schema.Schema.Type<typeof SloTargetPercent>
 
 /** Share of events the target allows to be bad, as a ratio: 99.9% gives 0.001. */
 export const errorBudgetRatio = (targetPercent: number): number => 1 - targetPercent / 100
+
+/** Sampling-weighted event counts over some window. */
+export interface SloCounts {
+	readonly total: number
+	readonly good: number
+}
+
+const badRatio = (counts: SloCounts): number =>
+	counts.total > 0 ? Math.max(0, counts.total - counts.good) / counts.total : 0
+
+/** Share of good events, or `null` when the window saw none. */
+export const sliRatio = (counts: SloCounts): number | null =>
+	counts.total > 0 ? counts.good / counts.total : null
+
+/**
+ * How fast the window spends the budget: 1 spends exactly the budget over the SLO window,
+ * 14.4 spends 2% of a 30-day budget in an hour. 0 when the window saw no events.
+ */
+export const burnRate = (counts: SloCounts, targetPercent: number): number =>
+	badRatio(counts) / errorBudgetRatio(targetPercent)
+
+/** Share of the window's error budget still unspent: 1 untouched, 0 spent, negative overspent. */
+export const budgetRemaining = (counts: SloCounts, targetPercent: number): number =>
+	1 - burnRate(counts, targetPercent)
+
+/**
+ * The multiwindow, multi-burn-rate alert tiers (Google SRE workbook, chapter 5): a tier fires
+ * when both its long and short windows burn at least `factor` times the sustainable rate. The
+ * short window makes the alert resolve soon after the burn stops.
+ */
+export const SLO_BURN_TIERS = [
+	{ severity: "critical", factor: 14.4, longWindowMinutes: 60, shortWindowMinutes: 5 },
+	{ severity: "warning", factor: 6, longWindowMinutes: 360, shortWindowMinutes: 30 },
+] as const
+export type SloBurnTier = (typeof SLO_BURN_TIERS)[number]
+
+/** Every window a tier reads, ascending. */
+export const SLO_BURN_WINDOWS_MINUTES = [5, 30, 60, 360] as const
+export type SloBurnWindowMinutes = (typeof SLO_BURN_WINDOWS_MINUTES)[number]
+
+/** The longest burn window: the span one burn-rate read covers. */
+export const SLO_BURN_LOOKBACK_MINUTES = Math.max(...SLO_BURN_WINDOWS_MINUTES)
+
+export interface SloBurnWindow extends SloCounts {
+	readonly windowMinutes: SloBurnWindowMinutes
+	readonly burnRate: number
+}
+
+/** Whether a tier fires on the given window burns. */
+export const tierBreached = (tier: SloBurnTier, windows: ReadonlyArray<SloBurnWindow>): boolean => {
+	const rateOf = (minutes: number) =>
+		Arr.findFirst(windows, (window) => window.windowMinutes === minutes).pipe(
+			Option.match({ onNone: () => 0, onSome: (window) => window.burnRate }),
+		)
+	return rateOf(tier.longWindowMinutes) >= tier.factor && rateOf(tier.shortWindowMinutes) >= tier.factor
+}
+
+/** A bucket of events starting at `startMs`. */
+export interface SloBucket extends SloCounts {
+	readonly startMs: number
+}
+
+const sumCounts = (buckets: ReadonlyArray<SloCounts>): SloCounts =>
+	Arr.reduce(buckets, { total: 0, good: 0 }, (acc, bucket) => ({
+		total: acc.total + bucket.total,
+		good: acc.good + bucket.good,
+	}))
+
+/** Each burn window ending at `endMs`, summed from minute buckets. */
+export const burnWindows = (
+	buckets: ReadonlyArray<SloBucket>,
+	endMs: number,
+	targetPercent: number,
+): ReadonlyArray<SloBurnWindow> =>
+	Arr.map(SLO_BURN_WINDOWS_MINUTES, (windowMinutes) => {
+		const startMs = endMs - windowMinutes * 60_000
+		const counts = sumCounts(
+			Arr.filter(buckets, (bucket) => bucket.startMs >= startMs && bucket.startMs < endMs),
+		)
+		return { windowMinutes, ...counts, burnRate: burnRate(counts, targetPercent) }
+	})
+
+/** One point of the budget chart: the trailing SLO window as of `atMs`. */
+export interface SloHistoryPoint {
+	readonly atMs: number
+	/** Good share of the bucket ending at `atMs`, `null` when it saw no events. */
+	readonly sli: number | null
+	/** Budget left over the SLO window ending at `atMs`. */
+	readonly budgetRemaining: number
+	/** Burn rate of the bucket ending at `atMs`. */
+	readonly burnRate: number
+}
+
+/**
+ * The budget chart from buckets covering two SLO windows: one point per bucket of the last
+ * window, each reading the SLO window that ends with it. Buckets are `bucketMs` wide and
+ * aligned to it; the last point is `endMs`, inside the newest (partial) bucket.
+ */
+export const budgetHistory = (input: {
+	readonly buckets: ReadonlyArray<SloBucket>
+	readonly bucketMs: number
+	readonly windowMs: number
+	readonly endMs: number
+	readonly targetPercent: number
+}): ReadonlyArray<SloHistoryPoint> => {
+	const slotsPerWindow = Math.round(input.windowMs / input.bucketMs)
+	const slotCount = 2 * slotsPerWindow
+	const alignedEnd = Math.ceil(input.endMs / input.bucketMs) * input.bucketMs
+	const origin = alignedEnd - slotCount * input.bucketMs
+	const slots = Arr.makeBy(slotCount, () => ({ total: 0, good: 0 }))
+	for (const bucket of input.buckets) {
+		const slot = slots[Math.floor((bucket.startMs - origin) / input.bucketMs)]
+		if (slot === undefined) continue
+		slot.total += bucket.total
+		slot.good += bucket.good
+	}
+	// prefix[j] sums slots [0, j).
+	const prefix = Arr.scan(slots, { total: 0, good: 0 }, (acc, slot) => ({
+		total: acc.total + slot.total,
+		good: acc.good + slot.good,
+	}))
+	const between = (from: number, to: number): SloCounts => {
+		const end = prefix[to] ?? { total: 0, good: 0 }
+		const start = prefix[from] ?? { total: 0, good: 0 }
+		return { total: end.total - start.total, good: end.good - start.good }
+	}
+	return Arr.makeBy(slotsPerWindow, (index) => {
+		const slot = slotsPerWindow + index
+		const last = between(slot, slot + 1)
+		return {
+			atMs: Math.min(origin + (slot + 1) * input.bucketMs, input.endMs),
+			sli: sliRatio(last),
+			budgetRemaining: budgetRemaining(
+				between(slot + 1 - slotsPerWindow, slot + 1),
+				input.targetPercent,
+			),
+			burnRate: burnRate(last, input.targetPercent),
+		}
+	})
+}
diff --git a/packages/query-engine/src/__sql_baseline__/catalog.sql b/packages/query-engine/src/__sql_baseline__/catalog.sql
index 4be9657ef0..37bc2ae94b 100644
--- a/packages/query-engine/src/__sql_baseline__/catalog.sql
+++ b/packages/query-engine/src/__sql_baseline__/catalog.sql
@@ -9550,6 +9550,273 @@ SELECT
           AND product_events.Timestamp <= '2026-01-03 14:15:00'
 FORMAT JSON
 
+-- builder:slos:sloBucketsQuery:hour  [7c98183a]
+SELECT
+          toStartOfInterval(slo_buckets.bBucket, INTERVAL 3600 SECOND) AS bucket,
+          sum(slo_buckets.bTotal) AS total,
+          sum(slo_buckets.bGood) AS good
+        FROM (
+SELECT
+          toStartOfHour(service_overview_spans.Timestamp) AS bBucket,
+          service_overview_spans.ServiceName AS bService,
+          service_overview_spans.DeploymentEnv AS bEnv,
+          sum(service_overview_spans.SampleRate) AS bTotal,
+          sumIf(service_overview_spans.SampleRate, service_overview_spans.StatusCode != 'Error') AS bGood
+        FROM service_overview_spans
+        WHERE service_overview_spans.OrgId = 'org_sql_catalog'
+          AND service_overview_spans.Timestamp >= '2026-01-01 10:30:00'
+          AND service_overview_spans.Timestamp <= '2026-01-03 14:15:00'
+          AND service_overview_spans.ServiceName = 'api'
+          AND (Timestamp < if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR) OR Timestamp >= toStartOfHour(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_hourly.Hour AS bBucket,
+          service_overview_hourly.ServiceName AS bService,
+          service_overview_hourly.DeploymentEnv AS bEnv,
+          sum(service_overview_hourly.EstimatedSpanCount) AS bTotal,
+          sum(service_overview_hourly.EstimatedSpanCount) - sum(service_overview_hourly.EstimatedErrorCount) AS bGood
+        FROM service_overview_hourly
+        WHERE service_overview_hourly.OrgId = 'org_sql_catalog'
+          AND service_overview_hourly.ServiceName = 'api'
+          AND service_overview_hourly.Hour >= if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR)
+          AND service_overview_hourly.Hour < toStartOfHour(toDateTime('2026-01-03 14:15:00'))
+        GROUP BY bBucket, bService, bEnv
+) AS slo_buckets
+        GROUP BY bucket
+        ORDER BY bucket ASC
+        FORMAT JSON
+
+-- builder:slos:sloBucketsQuery:minute  [25d44e08]
+SELECT
+          toStartOfInterval(slo_buckets.bBucket, INTERVAL 60 SECOND) AS bucket,
+          sum(slo_buckets.bTotal) AS total,
+          sum(slo_buckets.bGood) AS good
+        FROM (
+SELECT
+          toStartOfMinute(service_overview_spans.Timestamp) AS bBucket,
+          service_overview_spans.ServiceName AS bService,
+          service_overview_spans.DeploymentEnv AS bEnv,
+          sum(service_overview_spans.SampleRate) AS bTotal,
+          sumIf(service_overview_spans.SampleRate, service_overview_spans.StatusCode != 'Error') AS bGood
+        FROM service_overview_spans
+        WHERE service_overview_spans.OrgId = 'org_sql_catalog'
+          AND service_overview_spans.Timestamp >= '2026-01-01 10:30:00'
+          AND service_overview_spans.Timestamp <= '2026-01-03 14:15:00'
+          AND service_overview_spans.ServiceName = 'api'
+          AND (Timestamp < if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE) OR Timestamp >= toStartOfMinute(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_minutely.Minute AS bBucket,
+          service_overview_minutely.ServiceName AS bService,
+          service_overview_minutely.DeploymentEnv AS bEnv,
+          sum(service_overview_minutely.EstimatedSpanCount) AS bTotal,
+          sum(service_overview_minutely.EstimatedSpanCount) - sum(service_overview_minutely.EstimatedErrorCount) AS bGood
+        FROM service_overview_minutely
+        WHERE service_overview_minutely.OrgId = 'org_sql_catalog'
+          AND service_overview_minutely.ServiceName = 'api'
+          AND service_overview_minutely.Minute >= if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE)
+          AND service_overview_minutely.Minute < toStartOfMinute(toDateTime('2026-01-03 14:15:00'))
+        GROUP BY bBucket, bService, bEnv
+) AS slo_buckets
+        GROUP BY bucket
+        ORDER BY bucket ASC
+        FORMAT JSON
+
+-- builder:slos:sloBurnByServiceQuery:default  [a665111c]
+SELECT
+          slo_buckets.bService AS serviceName,
+          slo_buckets.bEnv AS environment,
+          sumIf(slo_buckets.bTotal, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 5 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS total5,
+          sumIf(slo_buckets.bGood, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 5 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS good5,
+          sumIf(slo_buckets.bTotal, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 30 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS total30,
+          sumIf(slo_buckets.bGood, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 30 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS good30,
+          sumIf(slo_buckets.bTotal, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 60 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS total60,
+          sumIf(slo_buckets.bGood, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 60 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS good60,
+          sumIf(slo_buckets.bTotal, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 360 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS total360,
+          sumIf(slo_buckets.bGood, (bBucket >= toDateTime('2026-01-03 14:15:00') - INTERVAL 360 MINUTE AND bBucket < toDateTime('2026-01-03 14:15:00'))) AS good360
+        FROM (
+SELECT
+          toStartOfMinute(service_overview_spans.Timestamp) AS bBucket,
+          service_overview_spans.ServiceName AS bService,
+          service_overview_spans.DeploymentEnv AS bEnv,
+          sum(service_overview_spans.SampleRate) AS bTotal,
+          sumIf(service_overview_spans.SampleRate, service_overview_spans.Duration < 500000000) AS bGood
+        FROM service_overview_spans
+        WHERE service_overview_spans.OrgId = 'org_sql_catalog'
+          AND service_overview_spans.Timestamp >= '2026-01-01 10:30:00'
+          AND service_overview_spans.Timestamp <= '2026-01-03 14:15:00'
+          AND service_overview_spans.ServiceName IN ('api', 'checkout')
+          AND (Timestamp < if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE) OR Timestamp >= toStartOfMinute(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_minutely.Minute AS bBucket,
+          service_overview_minutely.ServiceName AS bService,
+          service_overview_minutely.DeploymentEnv AS bEnv,
+          sum(service_overview_minutely.EstimatedLadderSpanCount) AS bTotal,
+          sum(service_overview_minutely.EstimatedUnder500Ms) AS bGood
+        FROM service_overview_minutely
+        WHERE service_overview_minutely.OrgId = 'org_sql_catalog'
+          AND service_overview_minutely.ServiceName IN ('api', 'checkout')
+          AND service_overview_minutely.Minute >= if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE)
+          AND service_overview_minutely.Minute < toStartOfMinute(toDateTime('2026-01-03 14:15:00'))
+        GROUP BY bBucket, bService, bEnv
+) AS slo_buckets
+        GROUP BY serviceName, environment
+        FORMAT JSON
+
+-- builder:slos:sloTotalsByServiceQuery:default  [602abdae]
+SELECT
+          slo_buckets.bService AS serviceName,
+          slo_buckets.bEnv AS environment,
+          sum(slo_buckets.bTotal) AS total,
+          sum(slo_buckets.bGood) AS good
+        FROM (
+SELECT
+          toStartOfMinute(service_overview_spans.Timestamp) AS bBucket,
+          service_overview_spans.ServiceName AS bService,
+          service_overview_spans.DeploymentEnv AS bEnv,
+          sum(service_overview_spans.SampleRate) AS bTotal,
+          sumIf(service_overview_spans.SampleRate, service_overview_spans.StatusCode != 'Error') AS bGood
+        FROM service_overview_spans
+        WHERE service_overview_spans.OrgId = 'org_sql_catalog'
+          AND service_overview_spans.Timestamp >= '2026-01-01 10:30:00'
+          AND service_overview_spans.Timestamp <= '2026-01-03 14:15:00'
+          AND service_overview_spans.ServiceName IN ('api', 'checkout')
+          AND (Timestamp < if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE) OR Timestamp >= toStartOfMinute(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_minutely.Minute AS bBucket,
+          service_overview_minutely.ServiceName AS bService,
+          service_overview_minutely.DeploymentEnv AS bEnv,
+          sum(service_overview_minutely.EstimatedSpanCount) AS bTotal,
+          sum(service_overview_minutely.EstimatedSpanCount) - sum(service_overview_minutely.EstimatedErrorCount) AS bGood
+        FROM service_overview_minutely
+        WHERE service_overview_minutely.OrgId = 'org_sql_catalog'
+          AND service_overview_minutely.ServiceName IN ('api', 'checkout')
+          AND service_overview_minutely.Minute >= if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE)
+          AND service_overview_minutely.Minute < toStartOfMinute(toDateTime('2026-01-03 14:15:00'))
+          AND (Minute < if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR) OR Minute >= toStartOfHour(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_hourly.Hour AS bBucket,
+          service_overview_hourly.ServiceName AS bService,
+          service_overview_hourly.DeploymentEnv AS bEnv,
+          sum(service_overview_hourly.EstimatedSpanCount) AS bTotal,
+          sum(service_overview_hourly.EstimatedSpanCount) - sum(service_overview_hourly.EstimatedErrorCount) AS bGood
+        FROM service_overview_hourly
+        WHERE service_overview_hourly.OrgId = 'org_sql_catalog'
+          AND service_overview_hourly.ServiceName IN ('api', 'checkout')
+          AND service_overview_hourly.Hour >= if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR)
+          AND service_overview_hourly.Hour < toStartOfHour(toDateTime('2026-01-03 14:15:00'))
+        GROUP BY bBucket, bService, bEnv
+) AS slo_buckets
+        GROUP BY serviceName, environment
+        FORMAT JSON
+
+-- builder:slos:sloTotalsQuery:availability  [66cc1667]
+SELECT
+          sum(slo_buckets.bTotal) AS total,
+          sum(slo_buckets.bGood) AS good
+        FROM (
+SELECT
+          toStartOfMinute(service_overview_spans.Timestamp) AS bBucket,
+          service_overview_spans.ServiceName AS bService,
+          service_overview_spans.DeploymentEnv AS bEnv,
+          sum(service_overview_spans.SampleRate) AS bTotal,
+          sumIf(service_overview_spans.SampleRate, service_overview_spans.StatusCode != 'Error') AS bGood
+        FROM service_overview_spans
+        WHERE service_overview_spans.OrgId = 'org_sql_catalog'
+          AND service_overview_spans.Timestamp >= '2026-01-01 10:30:00'
+          AND service_overview_spans.Timestamp <= '2026-01-03 14:15:00'
+          AND service_overview_spans.ServiceName = 'api'
+          AND service_overview_spans.DeploymentEnv IN ('production')
+          AND (Timestamp < if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE) OR Timestamp >= toStartOfMinute(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_minutely.Minute AS bBucket,
+          service_overview_minutely.ServiceName AS bService,
+          service_overview_minutely.DeploymentEnv AS bEnv,
+          sum(service_overview_minutely.EstimatedSpanCount) AS bTotal,
+          sum(service_overview_minutely.EstimatedSpanCount) - sum(service_overview_minutely.EstimatedErrorCount) AS bGood
+        FROM service_overview_minutely
+        WHERE service_overview_minutely.OrgId = 'org_sql_catalog'
+          AND service_overview_minutely.ServiceName = 'api'
+          AND service_overview_minutely.DeploymentEnv IN ('production')
+          AND service_overview_minutely.Minute >= if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE)
+          AND service_overview_minutely.Minute < toStartOfMinute(toDateTime('2026-01-03 14:15:00'))
+          AND (Minute < if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR) OR Minute >= toStartOfHour(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_hourly.Hour AS bBucket,
+          service_overview_hourly.ServiceName AS bService,
+          service_overview_hourly.DeploymentEnv AS bEnv,
+          sum(service_overview_hourly.EstimatedSpanCount) AS bTotal,
+          sum(service_overview_hourly.EstimatedSpanCount) - sum(service_overview_hourly.EstimatedErrorCount) AS bGood
+        FROM service_overview_hourly
+        WHERE service_overview_hourly.OrgId = 'org_sql_catalog'
+          AND service_overview_hourly.ServiceName = 'api'
+          AND service_overview_hourly.DeploymentEnv IN ('production')
+          AND service_overview_hourly.Hour >= if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR)
+          AND service_overview_hourly.Hour < toStartOfHour(toDateTime('2026-01-03 14:15:00'))
+        GROUP BY bBucket, bService, bEnv
+) AS slo_buckets
+        FORMAT JSON
+
+-- builder:slos:sloTotalsQuery:latency  [6487f581]
+SELECT
+          sum(slo_buckets.bTotal) AS total,
+          sum(slo_buckets.bGood) AS good
+        FROM (
+SELECT
+          toStartOfMinute(service_overview_spans.Timestamp) AS bBucket,
+          service_overview_spans.ServiceName AS bService,
+          service_overview_spans.DeploymentEnv AS bEnv,
+          sum(service_overview_spans.SampleRate) AS bTotal,
+          sumIf(service_overview_spans.SampleRate, service_overview_spans.Duration < 500000000) AS bGood
+        FROM service_overview_spans
+        WHERE service_overview_spans.OrgId = 'org_sql_catalog'
+          AND service_overview_spans.Timestamp >= '2026-01-01 10:30:00'
+          AND service_overview_spans.Timestamp <= '2026-01-03 14:15:00'
+          AND service_overview_spans.ServiceName = 'api'
+          AND (Timestamp < if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE) OR Timestamp >= toStartOfMinute(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_minutely.Minute AS bBucket,
+          service_overview_minutely.ServiceName AS bService,
+          service_overview_minutely.DeploymentEnv AS bEnv,
+          sum(service_overview_minutely.EstimatedLadderSpanCount) AS bTotal,
+          sum(service_overview_minutely.EstimatedUnder500Ms) AS bGood
+        FROM service_overview_minutely
+        WHERE service_overview_minutely.OrgId = 'org_sql_catalog'
+          AND service_overview_minutely.ServiceName = 'api'
+          AND service_overview_minutely.Minute >= if(toDateTime('2026-01-01 10:30:00') = toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')), toStartOfMinute(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 MINUTE)
+          AND service_overview_minutely.Minute < toStartOfMinute(toDateTime('2026-01-03 14:15:00'))
+          AND (Minute < if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR) OR Minute >= toStartOfHour(toDateTime('2026-01-03 14:15:00')))
+        GROUP BY bBucket, bService, bEnv
+UNION ALL
+SELECT
+          service_overview_hourly.Hour AS bBucket,
+          service_overview_hourly.ServiceName AS bService,
+          service_overview_hourly.DeploymentEnv AS bEnv,
+          sum(service_overview_hourly.EstimatedLadderSpanCount) AS bTotal,
+          sum(service_overview_hourly.EstimatedUnder500Ms) AS bGood
+        FROM service_overview_hourly
+        WHERE service_overview_hourly.OrgId = 'org_sql_catalog'
+          AND service_overview_hourly.ServiceName = 'api'
+          AND service_overview_hourly.Hour >= if(toDateTime('2026-01-01 10:30:00') = toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')), toStartOfHour(toDateTime('2026-01-01 10:30:00')) + INTERVAL 1 HOUR)
+          AND service_overview_hourly.Hour < toStartOfHour(toDateTime('2026-01-03 14:15:00'))
+        GROUP BY bBucket, bService, bEnv
+) AS slo_buckets
+        FORMAT JSON
+
 -- builder:source-usage:sourceUsageHourlyQuery:default  [7dcda428]
 SELECT
           toString(source_usage_hourly.Hour) AS hour,
diff --git a/packages/query-engine/src/benchmark/builders.ts b/packages/query-engine/src/benchmark/builders.ts
index 8af8a48ddd..5ce4676e09 100644
--- a/packages/query-engine/src/benchmark/builders.ts
+++ b/packages/query-engine/src/benchmark/builders.ts
@@ -1184,6 +1184,89 @@ export const builderFixtures: ReadonlyArray<BuilderFixture> = [
 		label: "default",
 		compile: () => CH.compileUnionUnsafe(CH.signalPresenceQuery(), window),
 	},
+	// SLOs (SlosService): the budget window, the hourly budget chart, the minute burn windows.
+	{
+		module: "slos",
+		name: "sloTotalsQuery",
+		label: "availability",
+		compile: () =>
+			CH.compileUnsafe(
+				CH.sloTotalsQuery({
+					serviceName: "api",
+					environments: ["production"],
+					sli: { _tag: "availability" },
+				}),
+				window,
+			),
+	},
+	{
+		module: "slos",
+		name: "sloTotalsQuery",
+		label: "latency",
+		compile: () =>
+			CH.compileUnsafe(
+				CH.sloTotalsQuery({
+					serviceName: "api",
+					environments: [],
+					sli: { _tag: "latency", thresholdMs: 500 },
+				}),
+				window,
+			),
+	},
+	{
+		module: "slos",
+		name: "sloBucketsQuery",
+		label: "hour",
+		compile: () =>
+			CH.compileUnsafe(
+				CH.sloBucketsQuery(
+					{ serviceName: "api", environments: [], sli: { _tag: "availability" } },
+					"hour",
+				),
+				{ ...window, bucketSeconds: 3600 },
+			),
+	},
+	{
+		module: "slos",
+		name: "sloBucketsQuery",
+		label: "minute",
+		compile: () =>
+			CH.compileUnsafe(
+				CH.sloBucketsQuery(
+					{ serviceName: "api", environments: [], sli: { _tag: "availability" } },
+					"minute",
+				),
+				{ ...window, bucketSeconds: 60 },
+			),
+	},
+	{
+		// The SLOs tab: every SLO's window totals in one read per (SLI, window).
+		module: "slos",
+		name: "sloTotalsByServiceQuery",
+		label: "default",
+		compile: () =>
+			CH.compileUnsafe(
+				CH.sloTotalsByServiceQuery({
+					serviceNames: ["api", "checkout"],
+					sli: { _tag: "availability" },
+				}),
+				window,
+			),
+	},
+	{
+		// The SLOs tab: every SLO's four burn windows in one read per SLI.
+		module: "slos",
+		name: "sloBurnByServiceQuery",
+		label: "default",
+		compile: () =>
+			CH.compileUnsafe(
+				CH.sloBurnByServiceQuery({
+					serviceNames: ["api", "checkout"],
+					sli: { _tag: "latency", thresholdMs: 500 },
+				}),
+				window,
+			),
+	},
 	// Audit log listing (apps/api/src/services/audit/AuditLogService.ts `list`).
 	{
 		module: "audit-log",
diff --git a/packages/query-engine/src/benchmark/catalog.test.ts b/packages/query-engine/src/benchmark/catalog.test.ts
index 9ab44385fc..22bcefc82e 100644
--- a/packages/query-engine/src/benchmark/catalog.test.ts
+++ b/packages/query-engine/src/benchmark/catalog.test.ts
@@ -48,6 +48,7 @@ import * as serviceEndpointDetailQueries from "../ch/queries/service-endpoint-de
 import * as serviceOperationQueries from "../ch/queries/service-operations"
 import * as prReviewQueries from "../ch/queries/pr-review"
 import * as serviceQueries from "../ch/queries/services"
+import * as sloQueries from "../ch/queries/slos"
 import * as releaseQueries from "../ch/queries/releases"
 import * as sessionEventQueries from "../ch/queries/session-events"
 import * as sessionReplayQueries from "../ch/queries/session-replays"
@@ -296,6 +297,7 @@ const QUERY_MODULES: Record<string, Record<string, unknown>> = {
 	"pr-review": prReviewQueries,
 	"service-operations": serviceOperationQueries,
 	services: serviceQueries,
+	slos: sloQueries,
 	"signal-presence": signalPresenceQueries,
 	"source-usage": sourceUsageQueries,
 	"aws-ingest": awsIngestQueries,
diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts
index dece64e4f6..37f0f9d38e 100644
--- a/packages/query-engine/src/ch/index.ts
+++ b/packages/query-engine/src/ch/index.ts
@@ -880,3 +880,20 @@ export {
 	type UptimeChecksOutput,
 	type UptimePhaseTimingsOutput,
 } from "./queries/uptime"
+export {
+	sloBucketsQuery,
+	sloBucketsRowSchema,
+	sloBurnByServiceQuery,
+	sloBurnByServiceRowSchema,
+	sloTotalsByServiceQuery,
+	sloTotalsByServiceRowSchema,
+	type SloBurnByServiceOutput,
+	type SloGroupQueryOpts,
+	type SloTotalsByServiceOutput,
+	sloTotalsQuery,
+	sloTotalsRowSchema,
+	type SloBucketsOutput,
+	type SloReadGrain,
+	type SloSliQueryOpts,
+	type SloTotalsOutput,
+} from "./queries/slos"
diff --git a/packages/query-engine/src/ch/queries/slos.test.ts b/packages/query-engine/src/ch/queries/slos.test.ts
new file mode 100644
index 0000000000..a4fb74567a
--- /dev/null
+++ b/packages/query-engine/src/ch/queries/slos.test.ts
@@ -0,0 +1,70 @@
+import { describe, expect, it } from "vitest"
+import { compileUnsafe } from "@maple-dev/effect-orm/clickhouse"
+import { OrgId } from "@maple/domain"
+import { sloBucketsQuery, sloTotalsQuery, type SloSliQueryOpts } from "./slos"
+
+const params = {
+	orgId: OrgId.make("org_1"),
+	startTime: "2024-01-01 00:00:30",
+	endTime: "2024-01-31 00:00:30",
+}
+
+const availability: SloSliQueryOpts = {
+	serviceName: "checkout",
+	environments: ["production"],
+	sli: { _tag: "availability" },
+}
+
+const latency: SloSliQueryOpts = {
+	serviceName: "checkout",
+	environments: [],
+	sli: { _tag: "latency", thresholdMs: 250 },
+}
+
+describe("sloTotalsQuery", () => {
+	it("reads the whole window from raw minutes, minutely partial hours and the hourly interior", () => {
+		const { sql } = compileUnsafe(sloTotalsQuery(availability), params)
+		expect(sql).toContain("FROM service_overview_spans")
+		expect(sql).toContain("FROM service_overview_minutely")
+		expect(sql).toContain("FROM service_overview_hourly")
+		// Raw spans only for the partial minutes at each end.
+		expect(sql).toContain("toStartOfMinute(toDateTime(")
+		expect(sql).toContain("ServiceName = 'checkout'")
+		expect(sql).toContain("DeploymentEnv IN ('production')")
+		expect(sql).toContain("StatusCode != 'Error'")
+	})
+
+	it("counts every environment when none is named", () => {
+		const { sql } = compileUnsafe(sloTotalsQuery(latency), params)
+		expect(sql).not.toContain("DeploymentEnv IN")
+	})
+
+	it("counts spans under the threshold as good for a latency SLI", () => {
+		const { sql } = compileUnsafe(sloTotalsQuery(latency), params)
+		expect(sql).toContain("Duration < 250000000")
+		expect(sql).toContain("sum(service_overview_hourly.EstimatedUnder250Ms)")
+		// Rollup buckets from before the ladder cover nothing and drop out of the total.
+		expect(sql).toContain("sum(service_overview_hourly.EstimatedLadderSpanCount)")
+	})
+})
+
+describe("sloBucketsQuery", () => {
+	it("buckets by hour from raw edges and the hourly rollup only", () => {
+		const { sql } = compileUnsafe(sloBucketsQuery(availability, "hour"), {
+			...params,
+			bucketSeconds: 3600,
+		})
+		expect(sql).toContain("FROM service_overview_hourly")
+		expect(sql).not.toContain("FROM service_overview_minutely")
+		expect(sql).toContain("GROUP BY bucket")
+	})
+
+	it("buckets sub-hour windows without the hourly rollup", () => {
+		const { sql } = compileUnsafe(sloBucketsQuery(availability, "minute"), {
+			...params,
+			bucketSeconds: 60,
+		})
+		expect(sql).toContain("FROM service_overview_minutely")
+		expect(sql).not.toContain("FROM service_overview_hourly")
+	})
+})
diff --git a/packages/query-engine/src/ch/queries/slos.ts b/packages/query-engine/src/ch/queries/slos.ts
new file mode 100644
index 0000000000..f72b47fece
--- /dev/null
+++ b/packages/query-engine/src/ch/queries/slos.ts
@@ -0,0 +1,296 @@
+// Typed SLO Queries
+//
+// An SLO counts a service's entry-point spans (the spans the service overview counts) as good
+// or bad. Counts are sampling-weighted (`SampleRate`), like the service overview's estimates.
+
+import { type DateTime, Schema } from "effect"
+import * as CH from "@maple-dev/effect-orm/expr"
+import { fromUnion, param, unionAll, type CompiledQueryRowSchema } from "@maple-dev/effect-orm/clickhouse"
+import { from } from "@maple-dev/effect-orm/clickhouse"
+import type { SloLatencyThresholdMs } from "@maple/domain/primitives"
+import type { SloSli } from "@maple/domain/slo"
+import { SLO_LATENCY_COVERAGE_COLUMN, sloLatencyColumnName } from "@maple/domain/tinybird"
+import { DateTimeUtcFromWarehouse } from "../../datetime"
+import { CHNumber } from "../schema"
+import { ServiceOverviewHourly, ServiceOverviewMinutely, ServiceOverviewSpans, orgIdParam } from "../tables"
+import { serviceOverviewWhereConditions } from "./query-helpers"
+import { edgeCondition, hourGrain, minuteGrain, utcInteriorConditions } from "./rollup-splice"
+
+export interface SloSliQueryOpts {
+	readonly serviceName: string
+	/** Empty counts every environment. */
+	readonly environments: readonly string[]
+	readonly sli: SloSli
+}
+
+/**
+ * How a read tiles its window:
+ * - `window`: one row for the whole range. Raw spans cover only the partial minutes at each
+ *   end, the minutely rollup the partial hours, the hourly rollup the rest.
+ * - `hour`: rows bucketed by `bucketSeconds`, a whole number of hours.
+ * - `minute`: rows bucketed by `bucketSeconds`, a whole number of minutes under an hour.
+ */
+export type SloReadGrain = "window" | "hour" | "minute"
+
+/** Which spans a read counts: one or more services, optionally some environments. */
+interface SloScope {
+	readonly serviceNames: readonly string[]
+	/** Empty counts every environment. */
+	readonly environments: readonly string[]
+	readonly sli: SloSli
+}
+
+const scopeOf = (opts: SloSliQueryOpts): SloScope => ({
+	serviceNames: [opts.serviceName],
+	environments: opts.environments,
+	sli: opts.sli,
+})
+
+const serviceCondition = (column: CH.Expr<string>, serviceNames: readonly string[]) =>
+	serviceNames.length === 1 && serviceNames[0] !== undefined
+		? column.eq(serviceNames[0])
+		: CH.inList(column, serviceNames)
+
+/** Every tier groups by these, so a read can sum them away or keep a row per service. */
+const BUCKET_KEYS = ["bBucket", "bService", "bEnv"] as const
+
+const rollupFilters = <
+	A extends {
+		readonly OrgId: CH.Expr<string>
+		readonly ServiceName: CH.Expr<string>
+		readonly DeploymentEnv: CH.Expr<string>
+	},
+>(
+	$: A,
+	opts: SloScope,
+) => [
+	$.OrgId.eq(orgIdParam),
+	serviceCondition($.ServiceName, opts.serviceNames),
+	opts.environments.length > 0 ? CH.inList($.DeploymentEnv, opts.environments) : undefined,
+]
+
+const nanos = (ms: number) => ms * 1_000_000
+
+/** Exact counts from the per-span projection, for the partial buckets at the window's ends. */
+const rawEdges = (opts: SloScope, grain: "hour" | "minute") =>
+	from(ServiceOverviewSpans)
+		.select(($) => ({
+			bBucket: grain === "minute" ? CH.toStartOfMinute($.Timestamp) : CH.toStartOfHour($.Timestamp),
+			bService: $.ServiceName,
+			bEnv: $.DeploymentEnv,
+			bTotal: CH.sum($.SampleRate),
+			bGood: CH.sumIf(
+				$.SampleRate,
+				opts.sli._tag === "availability"
+					? $.StatusCode.neq("Error")
+					: $.Duration.lt(nanos(opts.sli.thresholdMs)),
+			),
+		}))
+		.where(($) => [
+			...serviceOverviewWhereConditions($, {
+				serviceNames: opts.serviceNames,
+				environments: opts.environments,
+			}),
+			edgeCondition("Timestamp", grain === "minute" ? minuteGrain : hourGrain),
+		])
+		.groupBy(...BUCKET_KEYS)
+
+type LadderColumns = {
+	readonly [K in `EstimatedUnder${SloLatencyThresholdMs}Ms`]: CH.Expr<number>
+} & { readonly EstimatedLadderSpanCount: CH.Expr<number> }
+
+/**
+ * A rollup row's counts: every span and the non-errors, or for latency the spans the ladder
+ * covers and those under the threshold. Buckets written before the ladder cover nothing, so
+ * they drop out of a latency SLI instead of reading as all slow.
+ */
+const rollupCounts = <
+	A extends {
+		readonly EstimatedSpanCount: CH.Expr<number>
+		readonly EstimatedErrorCount: CH.Expr<number>
+	} & LadderColumns,
+>(
+	$: A,
+	sli: SloSli,
+) =>
+	sli._tag === "availability"
+		? {
+				bTotal: CH.sum($.EstimatedSpanCount),
+				bGood: CH.sum($.EstimatedSpanCount).sub(CH.sum($.EstimatedErrorCount)),
+			}
+		: {
+				bTotal: CH.sum($[SLO_LATENCY_COVERAGE_COLUMN]),
+				bGood: CH.sum($[sloLatencyColumnName(sli.thresholdMs)]),
+			}
+
+const hourlyInterior = (opts: SloScope) =>
+	from(ServiceOverviewHourly)
+		.select(($) => ({
+			bBucket: $.Hour,
+			bService: $.ServiceName,
+			bEnv: $.DeploymentEnv,
+			...rollupCounts($, opts.sli),
+		}))
+		.where(($) => [...rollupFilters($, opts), ...utcInteriorConditions($.Hour)])
+		.groupBy(...BUCKET_KEYS)
+
+const minutelyInterior = (opts: SloScope, withinHours: boolean) =>
+	from(ServiceOverviewMinutely)
+		.select(($) => ({
+			bBucket: $.Minute,
+			bService: $.ServiceName,
+			bEnv: $.DeploymentEnv,
+			...rollupCounts($, opts.sli),
+		}))
+		.where(($) => [
+			...rollupFilters($, opts),
+			...utcInteriorConditions($.Minute, minuteGrain),
+			// With the hourly tier present, only the partial hours at each end.
+			withinHours ? edgeCondition("Minute", hourGrain) : undefined,
+		])
+		.groupBy(...BUCKET_KEYS)
+
+const sloBuckets = (opts: SloScope, grain: SloReadGrain) => {
+	switch (grain) {
+		case "window":
+			return fromUnion(
+				unionAll(rawEdges(opts, "minute"), minutelyInterior(opts, true), hourlyInterior(opts)),
+				"slo_buckets",
+			)
+		case "hour":
+			return fromUnion(unionAll(rawEdges(opts, "hour"), hourlyInterior(opts)), "slo_buckets")
+		case "minute":
+			return fromUnion(unionAll(rawEdges(opts, "minute"), minutelyInterior(opts, false)), "slo_buckets")
+	}
+}
+
+export interface SloTotalsOutput {
+	/** Sampling-weighted entry-point spans in the window. */
+	readonly total: number
+	/** The good ones among them. */
+	readonly good: number
+}
+
+export const sloTotalsRowSchema = Schema.Struct({
+	total: CHNumber,
+	good: CHNumber,
+}) satisfies CompiledQueryRowSchema<SloTotalsOutput>
+
+/** Good and total events over `[startTime, endTime]`, one row. */
+export function sloTotalsQuery(opts: SloSliQueryOpts) {
+	return sloBuckets(scopeOf(opts), "window")
+		.select(($) => ({ total: CH.sum($.bTotal), good: CH.sum($.bGood) }))
+		.format("JSON")
+}
+
+export interface SloBucketsOutput extends SloTotalsOutput {
+	readonly bucket: DateTime.Utc
+}
+
+export const sloBucketsRowSchema = Schema.Struct({
+	bucket: DateTimeUtcFromWarehouse,
+	total: CHNumber,
+	good: CHNumber,
+}) satisfies CompiledQueryRowSchema<SloBucketsOutput>
+
+/**
+ * Good and total events per `bucketSeconds` over `[startTime, endTime]`. `hour` needs a bucket
+ * that is a whole number of hours; `minute` a whole number of minutes under an hour, since an
+ * hourly row has no position inside a sub-hour bucket.
+ */
+export function sloBucketsQuery(opts: SloSliQueryOpts, grain: "hour" | "minute") {
+	return sloBuckets(scopeOf(opts), grain)
+		.select(($) => ({
+			bucket: CH.toStartOfInterval($.bBucket, param.int("bucketSeconds")),
+			total: CH.sum($.bTotal),
+			good: CH.sum($.bGood),
+		}))
+		.groupBy("bucket")
+		.orderBy(["bucket", "asc"])
+		.format("JSON")
+}
+
+/** A batch read: every service at once, every environment, one row per (service, environment). */
+export interface SloGroupQueryOpts {
+	readonly serviceNames: readonly string[]
+	readonly sli: SloSli
+}
+
+export interface SloTotalsByServiceOutput extends SloTotalsOutput {
+	readonly serviceName: string
+	readonly environment: string
+}
+
+export const sloTotalsByServiceRowSchema = Schema.Struct({
+	serviceName: Schema.String,
+	environment: Schema.String,
+	total: CHNumber,
+	good: CHNumber,
+}) satisfies CompiledQueryRowSchema<SloTotalsByServiceOutput>
+
+/** `sloTotalsQuery` for many services in one read: the SLOs tab sums the rows each SLO counts. */
+export function sloTotalsByServiceQuery(opts: SloGroupQueryOpts) {
+	return sloBuckets({ serviceNames: opts.serviceNames, environments: [], sli: opts.sli }, "window")
+		.select(($) => ({
+			serviceName: $.bService,
+			environment: $.bEnv,
+			total: CH.sum($.bTotal),
+			good: CH.sum($.bGood),
+		}))
+		.groupBy("serviceName", "environment")
+		.format("JSON")
+}
+
+export interface SloBurnByServiceOutput {
+	readonly serviceName: string
+	readonly environment: string
+	readonly total5: number
+	readonly good5: number
+	readonly total30: number
+	readonly good30: number
+	readonly total60: number
+	readonly good60: number
+	readonly total360: number
+	readonly good360: number
+}
+
+export const sloBurnByServiceRowSchema = Schema.Struct({
+	serviceName: Schema.String,
+	environment: Schema.String,
+	total5: CHNumber,
+	good5: CHNumber,
+	total30: CHNumber,
+	good30: CHNumber,
+	total60: CHNumber,
+	good60: CHNumber,
+	total360: CHNumber,
+	good360: CHNumber,
+}) satisfies CompiledQueryRowSchema<SloBurnByServiceOutput>
+
+/** Minute buckets in the `minutes` before `endTime`, as the burn windows count them. */
+const inLastMinutes = (minutes: number) =>
+	CH.rawCond(
+		`(bBucket >= ${minuteGrain.endDt} - INTERVAL ${minutes} MINUTE AND bBucket < ${minuteGrain.endDt})`,
+	)
+
+/**
+ * The 5m, 30m, 1h and 6h burn windows ending at `endTime` for many services in one minute-grain
+ * read over `[startTime, endTime]` (the 6h lookback), one row per (service, environment).
+ */
+export function sloBurnByServiceQuery(opts: SloGroupQueryOpts) {
+	return sloBuckets({ serviceNames: opts.serviceNames, environments: [], sli: opts.sli }, "minute")
+		.select(($) => ({
+			serviceName: $.bService,
+			environment: $.bEnv,
+			total5: CH.sumIf($.bTotal, inLastMinutes(5)),
+			good5: CH.sumIf($.bGood, inLastMinutes(5)),
+			total30: CH.sumIf($.bTotal, inLastMinutes(30)),
+			good30: CH.sumIf($.bGood, inLastMinutes(30)),
+			total60: CH.sumIf($.bTotal, inLastMinutes(60)),
+			good60: CH.sumIf($.bGood, inLastMinutes(60)),
+			total360: CH.sumIf($.bTotal, inLastMinutes(360)),
+			good360: CH.sumIf($.bGood, inLastMinutes(360)),
+		}))
+		.groupBy("serviceName", "environment")
+		.format("JSON")
+}