diff --git a/apps/api/src/routes/internal/ai-sessions.http.test.ts b/apps/api/src/routes/internal/ai-sessions.http.test.ts
index 362df25a9..d5bf8b1ef 100644
--- a/apps/api/src/routes/internal/ai-sessions.http.test.ts
+++ b/apps/api/src/routes/internal/ai-sessions.http.test.ts
@@ -358,6 +358,179 @@ describe("POST /internal/ai-sessions/spans", () => {
 	})
 })
 
+/**
+ * The list is two reads, and the handler is what holds them together: it derives
+ * the fan-out's window from the page, and it re-imposes the page's order on an
+ * aggregation that cannot know it. Both are invisible in either query alone.
+ */
+describe("POST /internal/ai-sessions/list", () => {
+	const LIST_BODY = { ...WINDOW, limit: 3 }
+
+	/** A stage-one row: a session id and the extent of its agent spans. */
+	const pageRow = (sessionId: string, agentStart: string, agentEnd: string) => ({
+		sessionId,
+		agentStart,
+		agentEnd,
+	})
+
+	/** A stage-two row, in the wire shape the aggregation's SELECT decodes. */
+	const listRow = (sessionId: string, startTime: string) => ({
+		sessionId,
+		vendorId: "eve",
+		vendorVersion: "1",
+		traceCount: "1",
+		spanCount: "12",
+		errorSpanCount: "0",
+		serviceNames: ["agent-runner"],
+		startTime,
+		endTime: "2026-08-19 10:45:00.000000000",
+		durationMs: "1000",
+	})
+
+	// Deliberately not in start order: the page ranks on the first AGENT span and
+	// the aggregation orders by the first span of any kind, so the handler must
+	// not be able to reconstruct one from the other.
+	const PAGE = [
+		pageRow("wrun_beta", "2026-08-19 10:20:00.000000000", "2026-08-19 10:30:00.000000000"),
+		pageRow("wrun_alpha", "2026-08-19 10:05:00.000000000", "2026-08-19 10:40:00.000000000"),
+		pageRow(`trace:${TRACE_ID}`, "2026-08-19 09:50:00.000000000", "2026-08-19 10:00:00.000000000"),
+	]
+
+	it("bounds the aggregation by the page's own agent spans, not the caller's window", async () => {
+		let pageSql: string | undefined
+		let listSql: string | undefined
+		const harness = makeHarness({
+			compiledQuery: (_tenant, compiled, options) => {
+				if (options?.context === "aiSessionsPage") {
+					pageSql = compiledQueryOf(compiled).sql
+					return compiledQueryOf(compiled).decodeRows(PAGE).pipe(Effect.orDie)
+				}
+				listSql = compiledQueryOf(compiled).sql
+				return compiledQueryOf(compiled)
+					.decodeRows(PAGE.map((row) => listRow(row.sessionId, row.agentStart)))
+					.pipe(Effect.orDie)
+			},
+		})
+
+		try {
+			const response = await harness.post("/internal/ai-sessions/list", LIST_BODY)
+			expect(response.status).toBe(200)
+			// Stage one is the only read that sees the caller's window.
+			expect(pageSql).toContain("FROM ai_trace_index")
+			expect(pageSql).not.toContain("trace_detail_spans")
+			expect(pageSql).toContain(`Timestamp <= '${WINDOW.endTime}'`)
+			expect(pageSql).toContain("LIMIT 3")
+			// Stage two reads the fan-out table over the page's extent — the min
+			// agentStart and the max agentEnd of the rows stage one returned, padded.
+			// The caller's window would be 30 days of partitions on the page the UI
+			// actually offers.
+			expect(listSql).toContain("FROM trace_detail_spans")
+			expect(listSql).toContain("Timestamp >= '2026-08-19 09:50:00.000000000' - INTERVAL 3600 SECOND")
+			expect(listSql).toContain("Timestamp <= '2026-08-19 10:40:00.000000000' + INTERVAL 3600 SECOND")
+			// The caller's range reaches NO level of stage two — not the fan-out and
+			// not either of its two `ai_trace_index` reads. The handler hands it
+			// `orgId` and the page's two bounds, and nothing else.
+			expect(listSql).not.toContain(WINDOW.startTime)
+			expect(listSql).not.toContain(WINDOW.endTime)
+			// Three levels take that lower bound — the fan-out padded, and each of
+			// the two `ai_trace_index` reads exactly.
+			expect(listSql?.split("Timestamp >= '2026-08-19 09:50:00.000000000'").length).toBe(4)
+			expect(listSql).not.toContain("__PARAM_")
+		} finally {
+			await harness.dispose()
+		}
+	})
+
+	it("seeks stage two by exactly the session ids stage one ranked", async () => {
+		let listSql: string | undefined
+		const harness = makeHarness({
+			compiledQuery: (_tenant, compiled, options) => {
+				if (options?.context === "aiSessionsPage") {
+					return compiledQueryOf(compiled).decodeRows(PAGE).pipe(Effect.orDie)
+				}
+				listSql = compiledQueryOf(compiled).sql
+				return compiledQueryOf(compiled)
+					.decodeRows(PAGE.map((row) => listRow(row.sessionId, row.agentStart)))
+					.pipe(Effect.orDie)
+			},
+		})
+
+		try {
+			await harness.post("/internal/ai-sessions/list", LIST_BODY)
+			for (const row of PAGE) {
+				expect(listSql).toContain(`'${row.sessionId}'`)
+			}
+		} finally {
+			await harness.dispose()
+		}
+	})
+
+	it("answers in the page's order, dropping a session the aggregation lost", async () => {
+		const harness = makeHarness({
+			compiledQuery: (_tenant, compiled, options) =>
+				options?.context === "aiSessionsPage"
+					? compiledQueryOf(compiled).decodeRows(PAGE).pipe(Effect.orDie)
+					: compiledQueryOf(compiled)
+							// Reversed, and one short: the aggregation's own ORDER BY is
+							// meaningless to the client, and a trace whose spans fell outside
+							// the padded window returns nothing at all.
+							.decodeRows([
+								listRow(`trace:${TRACE_ID}`, "2026-08-19 09:49:00.000000000"),
+								listRow("wrun_beta", "2026-08-19 10:19:00.000000000"),
+							])
+							.pipe(Effect.orDie),
+		})
+
+		try {
+			const response = await harness.post("/internal/ai-sessions/list", LIST_BODY)
+			expect(response.status).toBe(200)
+			// The page's order is the order that was paged; re-sorting here would
+			// let a row jump between pages on a scroll. A session with no row is
+			// dropped rather than shown with blank counts.
+			expect((response.body.data as ReadonlyArray<{ sessionId: string }>).map((r) => r.sessionId)).toEqual([
+				"wrun_beta",
+				`trace:${TRACE_ID}`,
+			])
+			// Three ranked, two returned. `ranked` is what the client pages on: on
+			// `data.length` this short page reads as the end of the list, and the
+			// next offset would be one too low and re-show a session. The gap is
+			// real — the two MVs are written one after the other from the same
+			// insert, so the newest session can be ranked before its spans land.
+			expect(response.body.ranked).toBe(PAGE.length)
+			expect(response.body.ranked).toBe(3)
+		} finally {
+			await harness.dispose()
+		}
+	})
+
+	it("answers an empty page without touching trace_detail_spans", async () => {
+		const contexts: Array<string | undefined> = []
+		const harness = makeHarness({
+			compiledQuery: (_tenant, compiled, options) => {
+				contexts.push(options?.context)
+				return compiledQueryOf(compiled).decodeRows([]).pipe(Effect.orDie)
+			},
+		})
+
+		try {
+			const response = await harness.post("/internal/ai-sessions/list", LIST_BODY)
+			expect(response.status).toBe(200)
+			// `ranked` is omitted entirely, not sent as 0: the handler short-circuits
+			// with `new ListAiSessionsResponse({ data: [] })`, and the field is an
+			// `optionalKey`. The client's `?? data.length` fallback reads it as 0
+			// either way, which is what ends the scroll.
+			expect(response.body).toEqual({ data: [] })
+			expect("ranked" in response.body).toBe(false)
+			// One read, not two. With no ids to seek by, the fan-out's `IN ()` is a
+			// builder defect, and the shape it would have compiled to reads the whole
+			// padded window for nothing.
+			expect(contexts).toEqual(["aiSessionsPage"])
+		} finally {
+			await harness.dispose()
+		}
+	})
+})
+
 describe("POST /internal/ai-sessions/facets", () => {
 	// `pick("vendor")` in the handler and `facet("vendor", …)` in the query are
 	// two independent string literals in two packages. If either drifts both
diff --git a/apps/api/src/routes/internal/ai-sessions.http.ts b/apps/api/src/routes/internal/ai-sessions.http.ts
index 314af07ad..79a1522b0 100644
--- a/apps/api/src/routes/internal/ai-sessions.http.ts
+++ b/apps/api/src/routes/internal/ai-sessions.http.ts
@@ -34,26 +34,65 @@ export const HttpAiSessionsInternalLive = HttpApiBuilder.group(
 					Effect.gen(function* () {
 						const tenant = yield* CurrentTenant.Context
 						yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId })
-						const compiled = CH.compile(
-							Integrations.aiSessionListQuery({
-								limit: payload.limit,
-								offset: payload.offset,
-								vendorIds: payload.vendorIds,
-								serviceNames: payload.serviceNames,
-							}),
-							{
-								orgId: tenant.orgId,
-								startTime: payload.startTime,
-								endTime: payload.endTime,
-							},
+						const filters = { vendorIds: payload.vendorIds, serviceNames: payload.serviceNames }
+						const window = { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }
+						// Two reads, not one: the page is ranked on `ai_trace_index` over the
+						// caller's whole window, and only then is that page aggregated over
+						// `trace_detail_spans` — inside the hours its own agent spans cover,
+						// never the caller's window. See `aiSessionListQuery` for what the
+						// single-read shape cost.
+						const page = yield* warehouse.compiledQuery(
+							tenant,
+							CH.compile(
+								Integrations.aiSessionPageQuery({
+									...filters,
+									limit: payload.limit,
+									offset: payload.offset,
+								}),
+								window,
+							),
+							{ profile: "list", context: "aiSessionsPage" },
 						)
+						if (page.length === 0) {
+							return new ListAiSessionsResponse({ data: [] })
+						}
+						// Fixed-width warehouse literals, so they sort as the instants do.
+						const fanOutStart = page.map((row) => row.agentStart).reduce((a, b) => (a < b ? a : b))
+						const fanOutEnd = page.map((row) => row.agentEnd).reduce((a, b) => (a < b ? b : a))
 						// The row schema already coerces the UInt64 aggregates and decodes
 						// exactly the response's fields, so rows pass through unmapped.
-						const rows = yield* warehouse.compiledQuery(tenant, compiled, {
-							profile: "list",
-							context: "listAiSessions",
+						const rows = yield* warehouse.compiledQuery(
+							tenant,
+							CH.compile(
+								Integrations.aiSessionListQuery({
+									...filters,
+									sessionIds: page.map((row) => row.sessionId),
+								}),
+								{ orgId: tenant.orgId, fanOutStart, fanOutEnd },
+							),
+							{ profile: "list", context: "listAiSessions" },
+						)
+						// The page's order is the order that was paged, so it is the order
+						// shown: the aggregation sorts by the true first span, which leads
+						// the first agent span the page ranked on by under a second.
+						//
+						// A session the aggregation did not return is dropped rather than
+						// shown blank, and `ranked` tells the client the page was still a
+						// full one. It happens: `ai_trace_index` and `trace_detail_spans`
+						// are two materialized views written one after the other from the
+						// same `traces` insert, so the newest session — ranked first — can
+						// have index rows a moment before it has span rows, and at the far
+						// end of retention the two tables' TTL merges run on their own
+						// clocks. The two counts on the span are how often.
+						yield* Effect.annotateCurrentSpan({
+							"maple.ai.page_size": page.length,
+							"maple.ai.aggregated": rows.length,
+						})
+						const byId = new Map(rows.map((row) => [row.sessionId, row]))
+						return new ListAiSessionsResponse({
+							data: page.flatMap((row) => byId.get(row.sessionId) ?? []),
+							ranked: page.length,
 						})
-						return new ListAiSessionsResponse({ data: rows })
 					}),
 				)
 				.handle("facets", ({ payload }) =>
diff --git a/apps/api/src/services/warehouse/ai-trace-index-materialization.clickhouse.e2e.test.ts b/apps/api/src/services/warehouse/ai-trace-index-materialization.clickhouse.e2e.test.ts
index 88493a1a0..5f0161768 100644
--- a/apps/api/src/services/warehouse/ai-trace-index-materialization.clickhouse.e2e.test.ts
+++ b/apps/api/src/services/warehouse/ai-trace-index-materialization.clickhouse.e2e.test.ts
@@ -39,10 +39,21 @@ const ORG_ID = "org_ai_trace_index_e2e"
 // carry a 30-day TTL enforced at insert, and a hardcoded date would one day
 // silently drop every seed and let the suite compare nothing to nothing.
 const HOUR_MS = 3_600_000
-const BASE_MS = Date.now() - 2 * HOUR_MS
+// Floored to a whole second so the ONE seed that carries a fraction carries it
+// on purpose — see `SESSIONLESS_SPAN`. `Date.now()` has millisecond precision,
+// and letting it through would make every bound fractional and prove nothing.
+const BASE_MS = Math.floor((Date.now() - 2 * HOUR_MS) / 1000) * 1000
 
+// Milliseconds kept, not truncated to the second: `agentEnd` comes back off a
+// DateTime64(9) column and the fan-out is bounded by that literal, so a seed at
+// a fractional instant is the only thing that proves `Timestamp <= '{fanOutEnd}'`
+// still admits the very row that produced it.
 const chDateTime = (epochMs: number): string =>
-	new Date(epochMs).toISOString().replace("T", " ").slice(0, 19)
+	new Date(epochMs).toISOString().replace("T", " ").slice(0, 23)
+
+/** The same instant as `ai_trace_index` renders it: DateTime64(9), so the
+ *  millisecond literal above padded out to nanoseconds. */
+const chTimestamp = (epochMs: number): string => `${chDateTime(epochMs)}000000`
 
 const quote = (value: string): string => `'${value.replaceAll("'", "\\'")}'`
 
@@ -52,6 +63,10 @@ const SESSION_ID = `${ORG_ID}:inv-e2e-1`
 const AGENT_TRACE = "aitraceindexe2e000000000000000001"
 const SESSIONLESS_TRACE = "aitraceindexe2e000000000000000002"
 const PLAIN_TRACE = "aitraceindexe2e000000000000000003"
+/** A second trace of the SAME session — the reason the list groups traces. */
+const AGENT_TRACE_2 = "aitraceindexe2e000000000000000005"
+/** A third, hours past the caller's `endTime`: same session id, out of range. */
+const AGENT_TRACE_3 = "aitraceindexe2e000000000000000006"
 
 interface SeedSpan {
 	readonly traceId: string
@@ -62,37 +77,107 @@ interface SeedSpan {
 	readonly attrs: Readonly<Record<string, string>>
 }
 
-// Three populations, keyed off the constants the MV's write filter is rendered
-// from: a session-bearing agent span, a sessionless agent span, and a plain
-// span that must NOT materialize.
-const SEED_SPANS: ReadonlyArray<SeedSpan> = [
-	{
-		traceId: AGENT_TRACE,
-		spanId: "span-agent-1",
-		ms: BASE_MS,
-		service: "agent-service",
-		status: "Ok",
-		attrs: {
-			[MAPLE_AI_VENDOR_ID_ATTR]: "eve",
-			[MAPLE_AI_SESSION_ID_ATTR]: SESSION_ID,
-		},
+// The turn-owning span of the eve session: the only one of its trace that
+// carries the session key, which is why resolution is per-TRACE.
+const AGENT_TURN_SPAN: SeedSpan = {
+	traceId: AGENT_TRACE,
+	spanId: "span-agent-1",
+	ms: BASE_MS,
+	service: "agent-service",
+	status: "Ok",
+	attrs: {
+		[MAPLE_AI_VENDOR_ID_ATTR]: "eve",
+		[MAPLE_AI_SESSION_ID_ATTR]: SESSION_ID,
 	},
-	{
-		traceId: SESSIONLESS_TRACE,
-		spanId: "span-agent-2",
-		ms: BASE_MS + 60_000,
-		service: "agent-service",
-		status: "Error",
-		attrs: { [MAPLE_AI_VENDOR_ID_ATTR]: "vercel_ai_sdk" },
+}
+
+// A second agent span on the SAME trace, stamped by the SDK the agent calls
+// through and carrying no session id — an index row whose `SessionId` is ''.
+// `max(SessionId)` per trace is what keeps the trace under the eve session, and
+// the vendor `argMin` is what keeps the row's vendor `eve` rather than the
+// alphabetically-later `vercel_ai_sdk`.
+const AGENT_SDK_SPAN: SeedSpan = {
+	traceId: AGENT_TRACE,
+	spanId: "span-agent-1b",
+	ms: BASE_MS + 5_000,
+	service: "agent-service",
+	status: "Ok",
+	attrs: { [MAPLE_AI_VENDOR_ID_ATTR]: "vercel_ai_sdk" },
+}
+
+// A plain child of the agent trace, BEFORE its first agent span: no `maple_ai.*`
+// at all, so it is in `trace_detail_spans` and NOT in the index. It is what the
+// fan-out's pad exists for, and what makes `spanCount` bigger than the number of
+// agent spans — the full agent context the page promises.
+const AGENT_CHILD_SPAN: SeedSpan = {
+	traceId: AGENT_TRACE,
+	spanId: "span-agent-1c",
+	ms: BASE_MS - 50,
+	service: "web-service",
+	status: "Ok",
+	attrs: { "http.request.method": "GET" },
+}
+
+// A second TRACE of the same session — the join that makes `traceCount` 2.
+const AGENT_TURN_2_SPAN: SeedSpan = {
+	traceId: AGENT_TRACE_2,
+	spanId: "span-agent-3",
+	ms: BASE_MS + 30_000,
+	service: "agent-service",
+	status: "Ok",
+	attrs: {
+		[MAPLE_AI_VENDOR_ID_ATTR]: "eve",
+		[MAPLE_AI_SESSION_ID_ATTR]: SESSION_ID,
 	},
-	{
-		traceId: PLAIN_TRACE,
-		spanId: "span-plain-1",
-		ms: BASE_MS + 120_000,
-		service: "web-service",
-		status: "Ok",
-		attrs: { "http.request.method": "GET" },
+}
+
+// The sessionless agent trace, at a FRACTIONAL instant: it is the page's latest
+// agent span, so its timestamp is `fanOutEnd`, and stage two's
+// `Timestamp <= '{fanOutEnd}'` has to admit the row it was measured from. A
+// millisecond dropped anywhere in that round trip erases this session.
+const SESSIONLESS_SPAN: SeedSpan = {
+	traceId: SESSIONLESS_TRACE,
+	spanId: "span-agent-2",
+	ms: BASE_MS + 60_123,
+	service: "agent-service",
+	status: "Error",
+	attrs: { [MAPLE_AI_VENDOR_ID_ATTR]: "vercel_ai_sdk" },
+}
+
+// No `maple_ai.*`: must NOT materialize, and must not be detected as a session.
+const PLAIN_SPAN: SeedSpan = {
+	traceId: PLAIN_TRACE,
+	spanId: "span-plain-1",
+	ms: BASE_MS + 120_000,
+	service: "web-service",
+	status: "Ok",
+	attrs: { "http.request.method": "GET" },
+}
+
+// The same session id, hours before the caller's `startTime`. The page cannot rank
+// it, so its trace must not reach the aggregation either — stage two's index
+// levels are bounded by the PAGE, and a trace merged in there would inflate a
+// count for a window the user did not ask about.
+const EARLY_TURN_SPAN: SeedSpan = {
+	traceId: AGENT_TRACE_3,
+	spanId: "span-agent-4",
+	ms: BASE_MS - 3 * HOUR_MS,
+	service: "agent-service",
+	status: "Ok",
+	attrs: {
+		[MAPLE_AI_VENDOR_ID_ATTR]: "eve",
+		[MAPLE_AI_SESSION_ID_ATTR]: SESSION_ID,
 	},
+}
+
+const SEED_SPANS: ReadonlyArray<SeedSpan> = [
+	AGENT_TURN_SPAN,
+	AGENT_SDK_SPAN,
+	AGENT_CHILD_SPAN,
+	AGENT_TURN_2_SPAN,
+	SESSIONLESS_SPAN,
+	PLAIN_SPAN,
+	EARLY_TURN_SPAN,
 ]
 
 // A vendor span under ANOTHER org: it must materialize under its own OrgId —
@@ -157,53 +242,103 @@ describe.skipIf(!clickhouseE2eEnabled)("ai_trace_index materialization", () => {
 			 FROM ai_trace_index ORDER BY Timestamp ASC`,
 		)
 
+		/** The index row a seed span is expected to produce, by name. */
+		const indexRow = (orgId: string, span: SeedSpan) => ({
+			OrgId: orgId,
+			Timestamp: chTimestamp(span.ms),
+			TraceId: span.traceId,
+			SessionId: span.attrs[MAPLE_AI_SESSION_ID_ATTR] ?? "",
+			VendorId: span.attrs[MAPLE_AI_VENDOR_ID_ATTR] ?? "",
+			ServiceName: span.service,
+		})
+
+		// Every vendor-stamped span and nothing else: `AGENT_CHILD_SPAN` and
+		// `PLAIN_SPAN` carry no `maple_ai.*` and must be absent, and the two rows
+		// whose `SessionId` is '' are the reason the read side resolves a session
+		// per TRACE rather than per span.
 		assert.deepStrictEqual(rows, [
-			{
-				OrgId: ORG_ID,
-				Timestamp: `${chDateTime(SEED_SPANS[0]!.ms)}.000000000`,
-				TraceId: AGENT_TRACE,
-				SessionId: SESSION_ID,
-				VendorId: "eve",
-				ServiceName: "agent-service",
-			},
-			{
-				OrgId: ORG_ID,
-				Timestamp: `${chDateTime(SEED_SPANS[1]!.ms)}.000000000`,
-				TraceId: SESSIONLESS_TRACE,
-				SessionId: "",
-				VendorId: "vercel_ai_sdk",
-				ServiceName: "agent-service",
-			},
-			{
-				OrgId: FOREIGN_ORG_ID,
-				Timestamp: `${chDateTime(FOREIGN_SPAN.ms)}.000000000`,
-				TraceId: FOREIGN_SPAN.traceId,
-				SessionId: "",
-				VendorId: "eve",
-				ServiceName: "agent-service",
-			},
+			indexRow(ORG_ID, EARLY_TURN_SPAN),
+			indexRow(ORG_ID, AGENT_TURN_SPAN),
+			indexRow(ORG_ID, AGENT_SDK_SPAN),
+			indexRow(ORG_ID, AGENT_TURN_2_SPAN),
+			indexRow(ORG_ID, SESSIONLESS_SPAN),
+			indexRow(FOREIGN_ORG_ID, FOREIGN_SPAN),
 		])
 	})
 
-	it("feeds the real compiled list query end to end", async () => {
-		const compiled = compileUnsafe(Integrations.aiSessionListQuery(), {
+	// Both stages, wired the way the route wires them: the page is ranked on the
+	// index over the caller's window, and its own agent-span bounds are what the
+	// fan-out is then run over. Running the second with the first's real output
+	// is the only thing that proves the bounds it reports are a window
+	// ClickHouse accepts back as a param — the compiled SQL cannot say that.
+	it("feeds the real compiled page and list queries end to end", async () => {
+		const window = {
 			orgId: ORG_ID,
 			startTime: chDateTime(BASE_MS - HOUR_MS),
 			endTime: chDateTime(BASE_MS + HOUR_MS),
-		})
+		}
+		const compiledPage = compileUnsafe(Integrations.aiSessionPageQuery(), window)
 		// Decoded through the query's own row schema, exactly as `compiledQuery`
 		// does in production — the raw JSON alone would not catch a wire shape
 		// the schema refuses.
-		const rows = Effect.runSync(compiled.decodeRows(await runJson(compiled.sql)))
+		const page = Effect.runSync(compiledPage.decodeRows(await runJson(compiledPage.sql)))
 
 		// Newest session first: the sessionless agent trace files under its own
 		// `trace:` key, the session-bearing one under the vendor's id, and the
 		// plain trace and the foreign org's trace must not appear at all.
 		assert.deepStrictEqual(
-			rows.map((row) => [row.sessionId, row.vendorId, row.traceCount]),
+			page.map((row) => row.sessionId),
+			[`${MAPLE_AI_TRACE_SESSION_PREFIX}${SESSIONLESS_TRACE}`, SESSION_ID],
+		)
+
+		// The eve session's bounds span BOTH its traces: it starts on the first
+		// trace's turn span and ends on the second trace's, which is what makes the
+		// fan-out window a property of the session rather than of one trace.
+		// `agentEnd` is the SECOND trace's turn span, not the first trace's — and
+		// not `EARLY_TURN_SPAN`, which carries the same session id three hours out
+		// and which the page never saw, so it did not stretch the bounds.
+		const eve = page.find((row) => row.sessionId === SESSION_ID)
+		assert.deepStrictEqual(
+			[eve?.agentStart, eve?.agentEnd],
+			[chTimestamp(BASE_MS), chTimestamp(BASE_MS + 30_000)],
+		)
+
+		// The route's own derivation, character for character — string bounds that
+		// sort as the instants do.
+		const fanOutStart = page.map((row) => row.agentStart).reduce((a, b) => (a < b ? a : b))
+		const fanOutEnd = page.map((row) => row.agentEnd).reduce((a, b) => (a < b ? b : a))
+		// The upper bound lands on a fractional instant, and stage two compares
+		// `Timestamp <= '{fanOutEnd}'` against the DateTime64(9) column it came
+		// from. Truncate the literal anywhere and the row that SET the bound falls
+		// outside it — the sessionless session below is the canary.
+		assert.strictEqual(fanOutStart, chTimestamp(BASE_MS))
+		assert.strictEqual(fanOutEnd, chTimestamp(BASE_MS + 60_123))
+		assert.ok(fanOutEnd.endsWith(".123000000"))
+		// `orgId` and the page's two bounds — stage two takes no window param from
+		// the caller, so there is nothing else to pass.
+		const compiled = compileUnsafe(
+			Integrations.aiSessionListQuery({ sessionIds: page.map((row) => row.sessionId) }),
+			{ orgId: ORG_ID, fanOutStart, fanOutEnd },
+		)
+		const rows = Effect.runSync(compiled.decodeRows(await runJson(compiled.sql)))
+
+		// Reordered into the page's order and dropped where the aggregation has no
+		// row, as the handler does — same two sessions, now with the facts the
+		// index cannot answer.
+		const byId = new Map(rows.map((row) => [row.sessionId, row]))
+		assert.deepStrictEqual(
+			page
+				.flatMap((row) => byId.get(row.sessionId) ?? [])
+				.map((row) => [row.sessionId, row.vendorId, row.traceCount, row.spanCount]),
 			[
-				[`${MAPLE_AI_TRACE_SESSION_PREFIX}${SESSIONLESS_TRACE}`, "vercel_ai_sdk", 1],
-				[SESSION_ID, "eve", 1],
+				// Survived the `<= fanOutEnd` boundary it defined.
+				[`${MAPLE_AI_TRACE_SESSION_PREFIX}${SESSIONLESS_TRACE}`, "vercel_ai_sdk", 1, 1],
+				// Two traces merged, four spans: the turn span, the SDK span that
+				// carries no session id, the plain child that is not in the index at
+				// all, and the second trace's turn span. `eve` and not the
+				// alphabetically-later `vercel_ai_sdk`, because the vendor is the
+				// earliest SESSION-BEARING span's. `EARLY_TURN_SPAN` is not among them.
+				[SESSION_ID, "eve", 2, 4],
 			],
 		)
 	})
diff --git a/apps/web/src/hooks/use-infinite-ai-sessions.ts b/apps/web/src/hooks/use-infinite-ai-sessions.ts
index 4b822db26..8ddc9e9c6 100644
--- a/apps/web/src/hooks/use-infinite-ai-sessions.ts
+++ b/apps/web/src/hooks/use-infinite-ai-sessions.ts
@@ -26,8 +26,15 @@ export interface AiSessionsFilterInputs {
 
 interface AiSessionsPage {
 	data: ReadonlyArray<AgentSessionRow>
+	/** Sessions the server ranked for this page — what paging counts, since
+	 *  `data` can run short of it (see `ListAiSessionsResponse.ranked`). */
+	ranked: number
 }
 
+/** A page's ranked count, falling back to its row count for a server that
+ *  predates the field. */
+const rankedOf = (page: { data: ReadonlyArray<unknown>; ranked?: number }) => page.ranked ?? page.data.length
+
 /**
  * Offset-based infinite scroll for the agent-sessions list, mirroring
  * `useInfiniteReplays`. The first page flows through the cached result atom (so
@@ -66,15 +73,24 @@ export function useInfiniteAiSessions(filterInputs: AiSessionsFilterInputs) {
 	}, [firstPageResult, additionalPages])
 	const isCapped = allData.length >= MAX_RETAINED_AI_SESSIONS
 
+	// Paged on what the server RANKED, not on the rows it returned: a page can
+	// come back a row short of a full one and still not be the last (see
+	// `ListAiSessionsResponse.ranked`), so the row count would end the scroll
+	// early and the next offset would re-show a session.
+	const rankedCount = React.useMemo(() => {
+		const first = Result.isSuccess(firstPageResult) ? rankedOf(firstPageResult.value) : 0
+		return additionalPages.reduce((sum, page) => sum + page.ranked, first)
+	}, [firstPageResult, additionalPages])
+
 	const hasNextPage = React.useMemo(() => {
 		if (isCapped) return false
 		if (paginationStopped) return false
 		if (!Result.isSuccess(firstPageResult)) return false
 		if (additionalPages.length === 0) {
-			return firstPageResult.value.data.length === PAGE_SIZE
+			return rankedOf(firstPageResult.value) === PAGE_SIZE
 		}
 		const lastPage = additionalPages[additionalPages.length - 1]
-		return lastPage.data.length === PAGE_SIZE
+		return lastPage.ranked === PAGE_SIZE
 	}, [firstPageResult, additionalPages, paginationStopped, isCapped])
 
 	const fetchNextPage = React.useCallback(() => {
@@ -83,13 +99,13 @@ export function useInfiniteAiSessions(filterInputs: AiSessionsFilterInputs) {
 		setIsFetchingNextPage(true)
 
 		const currentKey = filterKeyRef.current
-		const offset = allData.length
+		const offset = rankedCount
 
 		mapleRuntime
 			.runPromise(listAiSessions({ data: { ...filterInputs, limit: PAGE_SIZE, offset } }))
 			.then((result) => {
 				if (filterKeyRef.current !== currentKey) return
-				setAdditionalPages((prev) => [...prev, { data: result.data }])
+				setAdditionalPages((prev) => [...prev, { data: result.data, ranked: rankedOf(result) }])
 			})
 			.catch((error) => {
 				if (filterKeyRef.current !== currentKey) return
@@ -104,7 +120,7 @@ export function useInfiniteAiSessions(filterInputs: AiSessionsFilterInputs) {
 				}
 				isFetchingRef.current = false
 			})
-	}, [filterInputs, allData.length, hasNextPage])
+	}, [filterInputs, rankedCount, hasNextPage])
 
 	return {
 		firstPageResult,
diff --git a/packages/domain/src/clickhouse/migrations/0024_ai_trace_index.ts b/packages/domain/src/clickhouse/migrations/0024_ai_trace_index.ts
index f20af284f..6922847f5 100644
--- a/packages/domain/src/clickhouse/migrations/0024_ai_trace_index.ts
+++ b/packages/domain/src/clickhouse/migrations/0024_ai_trace_index.ts
@@ -12,10 +12,11 @@
  * `ai_trace_index` is a filtered projection (the `error_events` shape): only
  * the vendor-stamped spans, with the `maple_ai.*` identity pre-extracted to
  * plain columns. Roughly 10k narrow rows per day at current volume, against
- * 70M raw spans. `aiSessionListQuery`'s detection subquery and
- * `aiSessionFacetsQuery` read it; the per-trace fan-out still reads
- * `trace_detail_spans`, which is where every other fact about an agent span
- * (its status, its failure attributes, its vendor version) is read from.
+ * 70M raw spans. `aiSessionPageQuery` (which ranks a page of sessions),
+ * `aiSessionListQuery`'s index levels and `aiSessionFacetsQuery` read it; the
+ * per-trace fan-out reads `trace_detail_spans` for the page's traces alone,
+ * which is where every other fact about an agent span (its status, its failure
+ * attributes, its vendor version) is read from.
  *
  * NOTHING IS BACKFILLED here: a materialized view sees inserts from creation
  * forward, so windows predating this migration under-report until the raw
diff --git a/packages/domain/src/http/ai-sessions.ts b/packages/domain/src/http/ai-sessions.ts
index a2225fb7c..7112f1417 100644
--- a/packages/domain/src/http/ai-sessions.ts
+++ b/packages/domain/src/http/ai-sessions.ts
@@ -10,7 +10,8 @@ import { warehouseReadHttpErrors } from "./warehouse"
 //
 // Backed by the `maple_ai.*` span attributes the ingest gateway stamps at
 // decode time; a session is resolved at trace granularity by
-// `aiSessionListQuery` in the query-engine integrations layer. The Agent
+// `aiSessionPageQuery` (which ranks a page) and `aiSessionListQuery` (which
+// aggregates it) in the query-engine integrations layer. The Agent
 // Sessions page is behind the `agent_tracing` org rollout flag and these
 // shapes exist for it alone, so they live in the internal tier where they can
 // follow the UI.
@@ -22,16 +23,18 @@ export class ListAiSessionsRequest extends Schema.Class<ListAiSessionsRequest>("
 		Schema.Number.check(Schema.isInt(), Schema.isBetween({ minimum: 1, maximum: 100 })),
 	),
 	/**
-	 * Rows to skip, for the list's infinite scroll. Offset-based like the
-	 * replays list: the page re-runs the aggregation and drops the first `offset`
-	 * sessions, which is fine at the volumes an org's agent traffic reaches
-	 * (~10k index rows a day) and lets the client keep a single ordered list
-	 * without a session-keyed cursor the aggregation cannot cheaply seek to.
+	 * Rows to skip, for the list's infinite scroll. Offset-based like the replays
+	 * list, and applied on the index-only page ranking (`aiSessionPageQuery`),
+	 * which is cheap to re-run at the volumes an org's agent traffic reaches
+	 * (~10k index rows a day). The span aggregation only ever covers the page
+	 * that ranking returned, so the offset costs nothing there.
 	 */
 	offset: Schema.optional(Schema.Number.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(0))),
-	// Both filters land on the session-detection subquery, so `serviceNames`
-	// means "the session-bearing spans came from this service", not "the trace
-	// touched it" — see `aiSessionListQuery`.
+	// Both filters land on the `ai_trace_index` level of both stages, each as
+	// its own per-trace existence test: `serviceNames` means "some agent span
+	// of the trace came from this service", not "the trace touched it", and a
+	// vendor and a service given together may be matched by different spans of
+	// the trace — see `aiSessionPageQuery`.
 	vendorIds: Schema.optional(Schema.Array(Schema.String)),
 	serviceNames: Schema.optional(Schema.Array(Schema.String)),
 }) {}
@@ -57,6 +60,18 @@ export const AiSessionListItem = Schema.Struct({
 
 export class ListAiSessionsResponse extends Schema.Class<ListAiSessionsResponse>("ListAiSessionsResponse")({
 	data: Schema.Array(AiSessionListItem),
+	/**
+	 * How many sessions the page RANKED, which `data` can fall short of: the
+	 * ranking reads `ai_trace_index` and the rows read `trace_detail_spans`,
+	 * two materialized views written one after the other from the same insert,
+	 * so the newest session can be ranked a moment before its spans are
+	 * readable. A client paging on `data.length` would take that short page for
+	 * the end of the list and skip nothing but stop scrolling; it should page on
+	 * this instead — the next offset is the sum of `ranked`, and a page is the
+	 * last one when `ranked` is under the limit. Optional only for a client
+	 * built before the field existed.
+	 */
+	ranked: Schema.optionalKey(Schema.Number),
 }) {}
 
 export class ListAiSessionsFacetsRequest extends Schema.Class<ListAiSessionsFacetsRequest>(
diff --git a/packages/domain/src/tinybird/datasources.ts b/packages/domain/src/tinybird/datasources.ts
index 3b16e65cc..c48ee15a7 100644
--- a/packages/domain/src/tinybird/datasources.ts
+++ b/packages/domain/src/tinybird/datasources.ts
@@ -1082,7 +1082,8 @@ export type TraceDetailSpansRow = InferRow<typeof traceDetailSpans>
 /**
  * Filtered projection of GenAI agent spans — every span the ingest gateway
  * stamped with `maple_ai.vendor.id` — for the Agent Sessions read path
- * (`aiSessionListQuery` detection + `aiSessionFacetsQuery`).
+ * (`aiSessionPageQuery`, `aiSessionListQuery`'s index levels, and
+ * `aiSessionFacetsQuery`).
  *
  * Why it exists: detecting agent traces by `mapContains(SpanAttributes, …)` on
  * raw `traces` cannot be indexed at this shape. GenAI spans are ~0.01% of rows
@@ -1092,10 +1093,11 @@ export type TraceDetailSpansRow = InferRow<typeof traceDetailSpans>
  * one hour, timeout at a day). This table holds only those spans, pre-extracted
  * to plain columns, so the same detection is a scan of ~10k narrow rows per day.
  *
- * The columns are exactly what the two readers need — the trace-id set, the
- * grouping key, and the two filter dimensions. Everything else about an agent
- * span (its failure attributes, its vendor version) is read per-trace off
- * `trace_detail_spans`, which the fan-out already touches.
+ * The columns are exactly what its readers need — the trace-id set, the
+ * grouping key, the two filter dimensions, and the agent-span bounds that tell
+ * the fan-out which hours to read. Everything else about an agent span (its
+ * failure attributes, its vendor version) is read per-trace off
+ * `trace_detail_spans`, over the page's bounds rather than the caller's window.
  *
  * Session ids live only on the turn-owning spans, so `SessionId` is '' for most
  * rows — resolution to a session key stays per-TRACE at read time, exactly as
diff --git a/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql b/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql
index 585d0f261..e0150f573 100644
--- a/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql
+++ b/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql
@@ -38,19 +38,18 @@ FORMAT JSON
 
 -- builder:ai-sessions:aiSessionListQuery:default
 SELECT
-          if(rawSessionId = '', concat('trace:', traceId), rawSessionId) AS sessionId,
-          argMin(vendorId, sessionStart) AS vendorId,
-          argMin(vendorVersion, sessionStart) AS vendorVersion,
+          if(index_traces.rawSessionId = '', concat('trace:', session_traces.traceId), index_traces.rawSessionId) AS sessionId,
+          argMin(session_traces.vendorId, session_traces.sessionStart) AS vendorId,
+          argMin(session_traces.vendorVersion, session_traces.sessionStart) AS vendorVersion,
           count() AS traceCount,
-          sum(spanCount) AS spanCount,
-          sum(errorSpanCount) AS errorSpanCount,
-          groupUniqArrayArray(serviceNames) AS serviceNames,
-          toString(min(traceStart)) AS startTime,
-          toString(fromUnixTimestamp64Nano(max(traceEndNanos))) AS endTime,
-          intDiv(max(traceEndNanos) - toUnixTimestamp64Nano(min(traceStart)), 1000000) AS durationMs
+          sum(session_traces.spanCount) AS spanCount,
+          sum(session_traces.errorSpanCount) AS errorSpanCount,
+          groupUniqArrayArray(session_traces.serviceNames) AS serviceNames,
+          toString(min(session_traces.traceStart)) AS startTime,
+          toString(fromUnixTimestamp64Nano(max(session_traces.traceEndNanos))) AS endTime,
+          intDiv(max(session_traces.traceEndNanos) - toUnixTimestamp64Nano(min(session_traces.traceStart)), 1000000) AS durationMs
         FROM (SELECT
           TraceId AS traceId,
-          max(SpanAttributes['maple_ai.session.id']) AS rawSessionId,
           argMin(SpanAttributes['maple_ai.vendor.id'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorId,
           argMin(SpanAttributes['maple_ai.vendor.version'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorVersion,
           min(if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS sessionStart,
@@ -61,35 +60,54 @@ SELECT
           max(toUnixTimestamp64Nano(Timestamp) + toInt64(Duration)) AS traceEndNanos
         FROM trace_detail_spans
         WHERE OrgId = 'org_sql_catalog'
-          AND Timestamp >= '2026-01-01 10:30:00' - INTERVAL 86400 SECOND
-          AND Timestamp <= '2026-01-03 14:15:00' + INTERVAL 86400 SECOND
+          AND Timestamp >= '2026-01-02 10:30:00' - INTERVAL 3600 SECOND
+          AND Timestamp <= '2026-01-02 12:30:00' + INTERVAL 3600 SECOND
           AND TraceId IN (SELECT
-          TraceId AS TraceId
+          traceId AS traceId
+        FROM (SELECT
+          TraceId AS traceId,
+          max(SessionId) AS rawSessionId,
+          min(Timestamp) AS traceAgentStart,
+          max(Timestamp) AS traceAgentEnd
         FROM ai_trace_index
         WHERE OrgId = 'org_sql_catalog'
-          AND Timestamp >= '2026-01-01 10:30:00'
-          AND Timestamp <= '2026-01-03 14:15:00')
+          AND Timestamp >= '2026-01-02 10:30:00'
+          AND Timestamp <= '2026-01-02 12:30:00'
+        GROUP BY traceId) AS agent_traces
+        WHERE if(rawSessionId = '', concat('trace:', traceId), rawSessionId) IN ('wrun_sql_catalog', 'trace:7f3a4b5c6d7e8f901234567890abcdef'))
         GROUP BY traceId) AS session_traces
+        INNER JOIN (SELECT
+          traceId AS traceId,
+          rawSessionId AS rawSessionId
+        FROM (SELECT
+          TraceId AS traceId,
+          max(SessionId) AS rawSessionId,
+          min(Timestamp) AS traceAgentStart,
+          max(Timestamp) AS traceAgentEnd
+        FROM ai_trace_index
+        WHERE OrgId = 'org_sql_catalog'
+          AND Timestamp >= '2026-01-02 10:30:00'
+          AND Timestamp <= '2026-01-02 12:30:00'
+        GROUP BY traceId) AS agent_traces
+        WHERE if(rawSessionId = '', concat('trace:', traceId), rawSessionId) IN ('wrun_sql_catalog', 'trace:7f3a4b5c6d7e8f901234567890abcdef')) AS index_traces ON session_traces.traceId = index_traces.traceId
         GROUP BY sessionId
         ORDER BY startTime DESC
-        LIMIT 50
         FORMAT JSON
 
 -- builder:ai-sessions:aiSessionListQuery:filtered
 SELECT
-          if(rawSessionId = '', concat('trace:', traceId), rawSessionId) AS sessionId,
-          argMin(vendorId, sessionStart) AS vendorId,
-          argMin(vendorVersion, sessionStart) AS vendorVersion,
+          if(index_traces.rawSessionId = '', concat('trace:', session_traces.traceId), index_traces.rawSessionId) AS sessionId,
+          argMin(session_traces.vendorId, session_traces.sessionStart) AS vendorId,
+          argMin(session_traces.vendorVersion, session_traces.sessionStart) AS vendorVersion,
           count() AS traceCount,
-          sum(spanCount) AS spanCount,
-          sum(errorSpanCount) AS errorSpanCount,
-          groupUniqArrayArray(serviceNames) AS serviceNames,
-          toString(min(traceStart)) AS startTime,
-          toString(fromUnixTimestamp64Nano(max(traceEndNanos))) AS endTime,
-          intDiv(max(traceEndNanos) - toUnixTimestamp64Nano(min(traceStart)), 1000000) AS durationMs
+          sum(session_traces.spanCount) AS spanCount,
+          sum(session_traces.errorSpanCount) AS errorSpanCount,
+          groupUniqArrayArray(session_traces.serviceNames) AS serviceNames,
+          toString(min(session_traces.traceStart)) AS startTime,
+          toString(fromUnixTimestamp64Nano(max(session_traces.traceEndNanos))) AS endTime,
+          intDiv(max(session_traces.traceEndNanos) - toUnixTimestamp64Nano(min(session_traces.traceStart)), 1000000) AS durationMs
         FROM (SELECT
           TraceId AS traceId,
-          max(SpanAttributes['maple_ai.session.id']) AS rawSessionId,
           argMin(SpanAttributes['maple_ai.vendor.id'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorId,
           argMin(SpanAttributes['maple_ai.vendor.version'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorVersion,
           min(if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS sessionStart,
@@ -100,19 +118,83 @@ SELECT
           max(toUnixTimestamp64Nano(Timestamp) + toInt64(Duration)) AS traceEndNanos
         FROM trace_detail_spans
         WHERE OrgId = 'org_sql_catalog'
-          AND Timestamp >= '2026-01-01 10:30:00' - INTERVAL 86400 SECOND
-          AND Timestamp <= '2026-01-03 14:15:00' + INTERVAL 86400 SECOND
+          AND Timestamp >= '2026-01-02 10:30:00' - INTERVAL 3600 SECOND
+          AND Timestamp <= '2026-01-02 12:30:00' + INTERVAL 3600 SECOND
           AND TraceId IN (SELECT
-          TraceId AS TraceId
+          traceId AS traceId
+        FROM (SELECT
+          TraceId AS traceId,
+          max(SessionId) AS rawSessionId,
+          min(Timestamp) AS traceAgentStart,
+          max(Timestamp) AS traceAgentEnd
         FROM ai_trace_index
         WHERE OrgId = 'org_sql_catalog'
-          AND Timestamp >= '2026-01-01 10:30:00'
-          AND Timestamp <= '2026-01-03 14:15:00'
-          AND VendorId IN ('eve')
-          AND ServiceName IN ('maple-slack-agent'))
+          AND Timestamp >= '2026-01-02 10:30:00'
+          AND Timestamp <= '2026-01-02 12:30:00'
+        GROUP BY traceId
+        HAVING countIf(VendorId IN ('eve')) > 0
+          AND countIf(ServiceName IN ('maple-slack-agent')) > 0) AS agent_traces
+        WHERE if(rawSessionId = '', concat('trace:', traceId), rawSessionId) IN ('wrun_sql_catalog', 'trace:7f3a4b5c6d7e8f901234567890abcdef'))
         GROUP BY traceId) AS session_traces
+        INNER JOIN (SELECT
+          traceId AS traceId,
+          rawSessionId AS rawSessionId
+        FROM (SELECT
+          TraceId AS traceId,
+          max(SessionId) AS rawSessionId,
+          min(Timestamp) AS traceAgentStart,
+          max(Timestamp) AS traceAgentEnd
+        FROM ai_trace_index
+        WHERE OrgId = 'org_sql_catalog'
+          AND Timestamp >= '2026-01-02 10:30:00'
+          AND Timestamp <= '2026-01-02 12:30:00'
+        GROUP BY traceId
+        HAVING countIf(VendorId IN ('eve')) > 0
+          AND countIf(ServiceName IN ('maple-slack-agent')) > 0) AS agent_traces
+        WHERE if(rawSessionId = '', concat('trace:', traceId), rawSessionId) IN ('wrun_sql_catalog', 'trace:7f3a4b5c6d7e8f901234567890abcdef')) AS index_traces ON session_traces.traceId = index_traces.traceId
         GROUP BY sessionId
         ORDER BY startTime DESC
+        FORMAT JSON
+
+-- builder:ai-sessions:aiSessionPageQuery:default
+SELECT
+          if(rawSessionId = '', concat('trace:', traceId), rawSessionId) AS sessionId,
+          toString(min(traceAgentStart)) AS agentStart,
+          toString(max(traceAgentEnd)) AS agentEnd
+        FROM (SELECT
+          TraceId AS traceId,
+          max(SessionId) AS rawSessionId,
+          min(Timestamp) AS traceAgentStart,
+          max(Timestamp) AS traceAgentEnd
+        FROM ai_trace_index
+        WHERE OrgId = 'org_sql_catalog'
+          AND Timestamp >= '2026-01-01 10:30:00'
+          AND Timestamp <= '2026-01-03 14:15:00'
+        GROUP BY traceId) AS index_traces
+        GROUP BY sessionId
+        ORDER BY agentStart DESC, sessionId ASC
+        LIMIT 50
+        FORMAT JSON
+
+-- builder:ai-sessions:aiSessionPageQuery:filtered
+SELECT
+          if(rawSessionId = '', concat('trace:', traceId), rawSessionId) AS sessionId,
+          toString(min(traceAgentStart)) AS agentStart,
+          toString(max(traceAgentEnd)) AS agentEnd
+        FROM (SELECT
+          TraceId AS traceId,
+          max(SessionId) AS rawSessionId,
+          min(Timestamp) AS traceAgentStart,
+          max(Timestamp) AS traceAgentEnd
+        FROM ai_trace_index
+        WHERE OrgId = 'org_sql_catalog'
+          AND Timestamp >= '2026-01-01 10:30:00'
+          AND Timestamp <= '2026-01-03 14:15:00'
+        GROUP BY traceId
+        HAVING countIf(VendorId IN ('eve')) > 0
+          AND countIf(ServiceName IN ('maple-slack-agent')) > 0) AS index_traces
+        GROUP BY sessionId
+        ORDER BY agentStart DESC, sessionId ASC
         LIMIT 25
         OFFSET 25
         FORMAT JSON
diff --git a/packages/query-engine-integrations/src/ai/ai-sessions.test.ts b/packages/query-engine-integrations/src/ai/ai-sessions.test.ts
index e4f3b078e..e3ed33bb4 100644
--- a/packages/query-engine-integrations/src/ai/ai-sessions.test.ts
+++ b/packages/query-engine-integrations/src/ai/ai-sessions.test.ts
@@ -1,9 +1,15 @@
 import { describe, expect, it } from "vitest"
 import { Effect } from "effect"
-import { compileUnsafe, compileUnionUnsafe, type CompiledQuery } from "@maple-dev/clickhouse-builder"
+import {
+	compileUnsafe,
+	compileUnionUnsafe,
+	QueryBuilderDefect,
+	type CompiledQuery,
+} from "@maple-dev/clickhouse-builder"
 import {
 	aiSessionFacetsQuery,
 	aiSessionListQuery,
+	aiSessionPageQuery,
 	aiSessionSpansQuery,
 	aiSessionSpansRowSchema,
 	aiSessionWindowQuery,
@@ -25,20 +31,190 @@ const traceParams = { ...params, traceId: TRACE_ID }
 /** The trace's session id, or the synthesized one — the grouping key. */
 const SESSION_KEY = "if(rawSessionId = '', concat('trace:', traceId), rawSessionId)"
 
+/** The same key on the list's joined level, where both sides are qualified. */
+const LIST_SESSION_KEY =
+	"if(index_traces.rawSessionId = '', concat('trace:', session_traces.traceId), index_traces.rawSessionId)"
+
+/** The extent of one page's agent spans — what `aiSessionPageQuery` reports and
+ *  the only window the fan-out is ever run over. Inside the caller's, by
+ *  construction: the page was ranked within it. */
+const FAN_OUT_START = "2026-08-18 10:00:00"
+const FAN_OUT_END = "2026-08-18 12:00:00"
+
+/** Stage 2's ENTIRE param set — the caller's window is not among them. */
+const listParams = { orgId: params.orgId, fanOutStart: FAN_OUT_START, fanOutEnd: FAN_OUT_END }
+
+/** A page of two sessions, one of each kind — the list never runs without one. */
+const listOpts = {
+	sessionIds: ["wrun_01M0CSAEW96BH2W9185XZPRPKH", `trace:${TRACE_ID}`],
+}
+
 const decodeRows = <T>(compiled: CompiledQuery<T>, rows: ReadonlyArray<Record<string, unknown>>) =>
 	Effect.runSync(compiled.decodeRows(rows))
 
-/** `OrgId = 'x'` on the detection level AND on the fan-out level. */
+/** `OrgId = 'x'` on every level that reads a table — a subquery contributes
+ *  nothing to the outer query's scope. */
 const orgPredicateCount = (sql: string) => sql.split("OrgId = 'org_1'").length - 1
 
+describe("aiSessionPageQuery", () => {
+	it("ranks the page on ai_trace_index alone, never on trace_detail_spans", () => {
+		const { sql } = compileUnsafe(aiSessionPageQuery(), params)
+
+		// The whole point of the split: this is the only read that sees the
+		// caller's window, so it must stay on the filtered projection. The moment
+		// it touches the fan-out table it costs what the single-read shape cost —
+		// 5–15s on a day, killed on a month (see the file header).
+		expect(sql).toContain("FROM ai_trace_index")
+		expect(sql).not.toContain("trace_detail_spans")
+		expect(sql).not.toContain("FROM traces")
+		expect(sql).not.toContain("SpanAttributes")
+	})
+
+	it("resolves the trace's session key, then groups the traces into sessions", () => {
+		const { sql } = compileUnsafe(aiSessionPageQuery(), params)
+
+		// Per trace first, because sessionless-ness is a property of the TRACE:
+		// keyed per index row, every non-turn agent span would become its own
+		// `trace:` session. Only then is the key grouped over.
+		expect(sql).toContain("max(SessionId) AS rawSessionId")
+		expect(sql).toContain(`${SESSION_KEY} AS sessionId`)
+		expect(sql).toContain("GROUP BY traceId")
+		expect(sql).toContain("GROUP BY sessionId")
+	})
+
+	it("returns the page's agent-span bounds and nothing else", () => {
+		const { sql } = compileUnsafe(aiSessionPageQuery(), params)
+
+		// These three columns ARE the contract with the fan-out: the ids it seeks
+		// by, and the window it seeks in.
+		expect(sql).toContain("toString(min(traceAgentStart)) AS agentStart")
+		expect(sql).toContain("toString(max(traceAgentEnd)) AS agentEnd")
+		expect(sql).not.toContain("spanCount")
+		expect(sql).not.toContain("serviceNames")
+	})
+
+	it("orders by the first agent span, with the session id breaking ties", () => {
+		const { sql } = compileUnsafe(aiSessionPageQuery(), params)
+
+		// `agentStart` is a fixed-width literal, so the String order is the
+		// instant order. The tiebreak is what stops a page boundary splitting two
+		// sessions that share a start — one would be shown twice and one never.
+		expect(sql).toContain("ORDER BY agentStart DESC, sessionId ASC")
+		expect(sql).toContain("LIMIT 50")
+	})
+
+	it("skips past the previous pages on the ordered session rows", () => {
+		const { sql } = compileUnsafe(aiSessionPageQuery({ limit: 25, offset: 100 }), params)
+
+		// The per-trace derived table has no order to page over, so the offset
+		// belongs to the level that ranked the sessions.
+		expect(sql).toContain("LIMIT 25\n        OFFSET 100")
+		expect(sql.split("OFFSET").length - 1).toBe(1)
+	})
+
+	it("emits no OFFSET clause for the first page", () => {
+		expect(compileUnsafe(aiSessionPageQuery({ offset: 0 }), params).sql).not.toContain("OFFSET")
+		expect(compileUnsafe(aiSessionPageQuery(), params).sql).not.toContain("OFFSET")
+	})
+
+	it("applies both filters as trace-level existence tests, after the grouping", () => {
+		const { sql } = compileUnsafe(
+			aiSessionPageQuery({ vendorIds: ["eve"], serviceNames: ["maple-slack-agent"] }),
+			params,
+		)
+
+		// HAVING, not WHERE: a row predicate would also narrow the rows
+		// `rawSessionId` is read from, so a vendor filter would file a trace under
+		// `trace:` whenever its turn-owning span belongs to the other vendor it
+		// calls through. It also has to match `aiSessionFacetsQuery`'s any-span
+		// counting, or the sidebar's number and the page's length disagree.
+		expect(sql).toContain("HAVING countIf(VendorId IN ('eve')) > 0")
+		expect(sql).toContain("AND countIf(ServiceName IN ('maple-slack-agent')) > 0")
+		expect(sql).not.toContain("WHERE VendorId IN")
+	})
+
+	it("tests each filter dimension separately, not one row against both", () => {
+		const { sql } = compileUnsafe(
+			aiSessionPageQuery({ vendorIds: ["eve"], serviceNames: ["maple-slack-agent"] }),
+			params,
+		)
+		const [where] = sql.split("GROUP BY traceId")
+
+		// Two `countIf`s, not one: the semantics are "SOME agent span of the trace
+		// is eve" AND "SOME agent span of the trace is maple-slack-agent" — which
+		// need not be the same span. A single row-level
+		// `WHERE VendorId IN (…) AND ServiceName IN (…)` would demand one span
+		// satisfying both, and would drop an eve session whose eve spans and whose
+		// maple-slack-agent spans are different spans — the ordinary case, since a
+		// trace's spans come from several services. Neither name may appear in the
+		// index read's WHERE at all.
+		expect(sql.split("countIf(").length - 1).toBe(2)
+		expect(where).not.toContain("VendorId")
+		expect(where).not.toContain("ServiceName")
+		expect(sql).not.toContain("VendorId IN ('eve') AND ServiceName")
+	})
+
+	it("omits the optional filters when none are given", () => {
+		const { sql } = compileUnsafe(aiSessionPageQuery(), params)
+
+		expect(sql).not.toContain("HAVING")
+		expect(sql).not.toContain("VendorId IN")
+		expect(sql).not.toContain("ServiceName IN")
+	})
+
+	it("is org-scoped, and escapes an org id carrying a quote", () => {
+		const compiled = compileUnsafe(aiSessionPageQuery(), params)
+		expect(compiled.tenantScope).toBe("single-tenant")
+		expect(orgPredicateCount(compiled.sql)).toBe(1)
+
+		expect(compileUnsafe(aiSessionPageQuery(), { ...params, orgId: "org'evil" }).sql).toContain(
+			"OrgId = 'org\\'evil'",
+		)
+	})
+
+	it("bounds the read by the caller's window, unpadded", () => {
+		const { sql } = compileUnsafe(aiSessionPageQuery(), params)
+
+		// The pad belongs to the fan-out, which reads whole traces. The page reads
+		// agent spans only, and widening it would change which sessions the range
+		// reports.
+		expect(sql).toContain(`Timestamp >= '${params.startTime}'`)
+		expect(sql).toContain(`Timestamp <= '${params.endTime}'`)
+		expect(sql).not.toContain("INTERVAL")
+	})
+
+	it("leaves no unresolved param placeholder", () => {
+		expect(compileUnsafe(aiSessionPageQuery(), params).sql).not.toContain("__PARAM_")
+	})
+
+	it("decodes the bounds the fan-out takes back as params", () => {
+		const compiled = compileUnsafe(aiSessionPageQuery(), params)
+
+		expect(
+			decodeRows(compiled, [
+				{
+					sessionId: "wrun_01M0CSAEW96BH2W9185XZPRPKH",
+					agentStart: "2026-08-19 10:33:25.825000000",
+					agentEnd: "2026-08-19 10:33:36.242000000",
+				},
+			]),
+		).toEqual([
+			{
+				sessionId: "wrun_01M0CSAEW96BH2W9185XZPRPKH",
+				agentStart: "2026-08-19 10:33:25.825000000",
+				agentEnd: "2026-08-19 10:33:36.242000000",
+			},
+		])
+	})
+})
+
 describe("aiSessionListQuery", () => {
-	it("detects sessions on ai_trace_index, then fans out over trace_detail_spans", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+	it("aggregates one page of sessions, seeking trace_detail_spans by trace id", () => {
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
-		// The tier is the point: detection must read the filtered projection, not
-		// raw `traces` — the raw scan reads the fat Map column for every span in
-		// the window and cannot be saved by the bloom index (see the file header).
-		// The fan-out reads the MV whose sort key starts (OrgId, TraceId).
+		// The fan-out reads the MV whose sort key starts (OrgId, TraceId), and the
+		// id set is pushed into that read by `IN` rather than joined — the same
+		// reason `errorDetailTracesQuery` uses it.
 		expect(sql).toContain("FROM trace_detail_spans")
 		expect(sql).toContain("TraceId IN (SELECT")
 		expect(sql).toContain("FROM ai_trace_index")
@@ -46,41 +222,81 @@ describe("aiSessionListQuery", () => {
 		expect(sql).toContain("GROUP BY traceId")
 		expect(sql).toContain("GROUP BY sessionId")
 		expect(sql).toContain("ORDER BY startTime DESC")
-		expect(sql).toContain("LIMIT 50")
+	})
+
+	it("pages nowhere itself — the page it was handed is the page", () => {
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
+
+		// A LIMIT here would cut the page the caller already ranked, and the
+		// missing sessions would silently vanish from a scroll that had room.
+		expect(sql).not.toContain("LIMIT")
+		expect(sql).not.toContain("OFFSET")
+	})
+
+	it("restricts both index reads to the page's sessions, escaping the ids", () => {
+		const { sql } = compileUnsafe(
+			aiSessionListQuery({ sessionIds: ["wrun_01M0CSAEW96BH2W9185XZPRPKH", "sess'evil"] }),
+			listParams,
+		)
+
+		// The ids come back off the page's own rows, but they are session ids a
+		// vendor chose, so the escaping is what stands between one and the query.
+		expect(sql).toContain(
+			`${SESSION_KEY} IN ('wrun_01M0CSAEW96BH2W9185XZPRPKH', 'sess\\'evil')`,
+		)
+		expect(sql.split(`${SESSION_KEY} IN (`).length - 1).toBe(2)
+	})
+
+	it("takes the session key from the index, not from the spans", () => {
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
+
+		// The one JOIN in the file, and it is here so the aggregation files a trace
+		// under exactly the key the page ranked it by: two derivations over two
+		// windows can disagree, and a disagreement drops the row from its own page.
+		expect(sql).toContain("INNER JOIN")
+		expect(sql).toContain("AS index_traces ON session_traces.traceId = index_traces.traceId")
+		expect(sql).toContain(`${LIST_SESSION_KEY} AS sessionId`)
+		expect(sql).not.toContain("max(SpanAttributes['maple_ai.session.id'])")
+		// The index read nested inside each side is `agent_traces`, so the JOIN's
+		// own `index_traces` alias is the only thing that name resolves to — the
+		// two used to collide one level apart.
+		expect(sql.split("AS agent_traces").length - 1).toBe(2)
+		expect(sql.split("AS index_traces").length - 1).toBe(1)
 	})
 
 	it("repeats the org predicate on every level that reads a table", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
-		expect(orgPredicateCount(sql)).toBe(2)
+		// Three now, not two: the fan-out plus both reads of the index — one for
+		// the id set, one for the key. A subquery contributes nothing to the outer
+		// query's scope.
+		expect(orgPredicateCount(sql)).toBe(3)
 	})
 
 	it("is org-scoped", () => {
-		expect(compileUnsafe(aiSessionListQuery(), params).tenantScope).toBe("single-tenant")
+		expect(compileUnsafe(aiSessionListQuery(listOpts), listParams).tenantScope).toBe(
+			"single-tenant",
+		)
 	})
 
-	it("detects on index membership, with no attribute predicate at all", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+	it("selects the page's traces on index membership, with no attribute predicate", () => {
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 		const [, detection] = sql.split("TraceId IN (SELECT")
 
-		// The session id is sparse by vendor — several frameworks never emit one —
-		// so detection keys on the vendor stamp. That predicate now lives in
-		// `ai_trace_index_mv`'s write filter: every row of the index carries a
-		// non-empty vendor id, so being in the table IS the guard and the read
-		// touches no Map column.
+		// The vendor predicate lives in `ai_trace_index_mv`'s write filter: every
+		// row of the index carries a non-empty vendor id, so being in the table IS
+		// the guard and the read touches no Map column.
 		expect(detection).toContain("FROM ai_trace_index")
 		expect(detection).not.toContain("mapContains")
 		expect(detection).not.toContain("SpanAttributes")
 	})
 
 	it("keys a trace with no session id on the trace itself", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
-		// One session per sessionless trace, and the per-trace derived table is the
-		// only level that can say so: a span of a session-bearing trace carries no
-		// session id of its own either.
-		expect(sql).toContain(`${SESSION_KEY} AS sessionId`)
-		expect(sql).toContain("max(SpanAttributes['maple_ai.session.id']) AS rawSessionId")
+		// One session per sessionless trace, resolved on the per-trace level: a
+		// span of a session-bearing trace carries no session id of its own either.
+		expect(sql).toContain("max(SessionId) AS rawSessionId")
 		expect(sql).toContain("GROUP BY sessionId")
 		// The guard that used to drop them. The key is never empty now, and a
 		// blank one would have swallowed every such trace into one session.
@@ -88,7 +304,7 @@ describe("aiSessionListQuery", () => {
 	})
 
 	it("tests session-id presence with mapContains AND a non-empty value", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
 		// ClickHouse yields '' for a missing Map key, so mapContains alone would
 		// rank spans carrying an empty session id as session-bearing.
@@ -98,21 +314,23 @@ describe("aiSessionListQuery", () => {
 	})
 
 	it("resolves the vendor from the earliest session-bearing span, not max()", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
 		// max(vendorId) picked `vercel_ai_sdk` alphabetically over the `eve` that
 		// actually ran the turn — see the builder's doc comment.
 		expect(sql).not.toContain("max(SpanAttributes['maple_ai.vendor.id'])")
 		expect(sql).toContain("argMin(SpanAttributes['maple_ai.vendor.id'], tuple(multiIf(")
 		expect(sql).toContain("argMin(SpanAttributes['maple_ai.vendor.version'], tuple(multiIf(")
-		expect(sql).toContain("argMin(vendorId, sessionStart) AS vendorId")
-		expect(sql).toContain("argMin(vendorVersion, sessionStart) AS vendorVersion")
+		expect(sql).toContain("argMin(session_traces.vendorId, session_traces.sessionStart) AS vendorId")
+		expect(sql).toContain(
+			"argMin(session_traces.vendorVersion, session_traces.sessionStart) AS vendorVersion",
+		)
 		// The sentinel must stay inside DateTime's range or toDateTime won't parse.
 		expect(sql).toContain("toDateTime('2106-01-01 00:00:00')")
 	})
 
 	it("ranks a sessionless trace's spans so a vendor-stamped one wins", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
 		// Every span of a sessionless trace ties at the sentinel under the session
 		// ordering alone, and argMin over ties is non-deterministic — it handed
@@ -123,69 +341,96 @@ describe("aiSessionListQuery", () => {
 		)
 	})
 
+	it("counts an attribute-declared failure on an Ok span as an error", () => {
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
+
+		// Frameworks record failed model and tool calls as values on `Ok` spans,
+		// and the list badge has to count what the detail page's Failures panel
+		// counts — `spanFailed` in `session-turns.ts` is the other half.
+		expect(sql).toContain(
+			"countIf((StatusCode = 'Error' OR (SpanAttributes['maple_ai.vendor.id'] != '' AND (SpanAttributes['error.type'] != '' OR SpanAttributes['gen_ai.response.status'] IN ('failed', 'error')))))",
+		)
+	})
+
 	it("escapes an org id carrying a quote", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), { ...params, orgId: "org'evil" })
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), {
+			...listParams,
+			orgId: "org'evil",
+		})
 
 		expect(sql).toContain("OrgId = 'org\\'evil'")
 	})
 
 	it("omits the optional filters when none are given", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
 		expect(sql).not.toContain("VendorId IN")
 		expect(sql).not.toContain("ServiceName IN")
 	})
 
-	it("puts both optional filters on the detection level only", () => {
+	it("puts both optional filters on the index level only", () => {
 		const { sql } = compileUnsafe(
-			aiSessionListQuery({ limit: 25, vendorIds: ["eve"], serviceNames: ["maple-slack-agent"] }),
-			params,
+			aiSessionListQuery({
+				...listOpts,
+				vendorIds: ["eve"],
+				serviceNames: ["maple-slack-agent"],
+			}),
+			listParams,
 		)
 
 		// Filtering the fan-out instead would drop spans and under-count spanCount.
+		// They must also be the SAME filters the page ran under, or the two stages
+		// resolve traces differently and the join silently loses rows.
 		const [fanOut, detection] = sql.split("TraceId IN (SELECT")
-		expect(detection).toContain("VendorId IN ('eve')")
-		expect(detection).toContain("ServiceName IN ('maple-slack-agent')")
+		expect(detection).toContain("HAVING countIf(VendorId IN ('eve')) > 0")
+		expect(detection).toContain("AND countIf(ServiceName IN ('maple-slack-agent')) > 0")
 		expect(fanOut).not.toContain("IN ('eve')")
-		expect(sql).toContain("LIMIT 25")
 	})
 
-	it("skips past the previous pages on the outermost level only", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery({ limit: 50, offset: 100 }), params)
+	it("reads every level over the page's bounds — padded for the fan-out only", () => {
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
+		const [fanOut, detection] = sql.split("TraceId IN (SELECT")
 
-		// The offset must apply to the ordered SESSION rows — the derived per-trace
-		// level has no order to page over.
-		expect(sql).toContain("LIMIT 50\n        OFFSET 100")
-		expect(sql.split("OFFSET").length - 1).toBe(1)
+		// `trace_detail_spans` is PARTITION BY toDate(Timestamp), so this predicate
+		// is the only thing that prunes partitions there — and the bounds are the
+		// page's own agent spans, which span hours, not the caller's 30 days.
+		expect(fanOut).toContain(`Timestamp >= '${FAN_OUT_START}' - INTERVAL 3600 SECOND`)
+		expect(fanOut).toContain(`Timestamp <= '${FAN_OUT_END}' + INTERVAL 3600 SECOND`)
+		// The index levels take the same bounds unpadded: a page trace's index rows
+		// lie between its own session's agentStart and agentEnd by construction, so
+		// the key and the filters come out of hours of the index rather than the
+		// caller's month, and the two index scans stop being the cost they were.
+		expect(detection).toContain(`Timestamp >= '${FAN_OUT_START}'`)
+		expect(detection).toContain(`Timestamp <= '${FAN_OUT_END}'`)
+		expect(detection).not.toContain("INTERVAL")
 	})
 
-	it("emits no OFFSET clause for the first page", () => {
-		expect(compileUnsafe(aiSessionListQuery({ offset: 0 }), params).sql).not.toContain("OFFSET")
-		expect(compileUnsafe(aiSessionListQuery(), params).sql).not.toContain("OFFSET")
-	})
+	it("takes no window param from the caller at all", () => {
+		const { sql } = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
-	it("pads the fan-out window rather than dropping it", () => {
-		const { sql } = compileUnsafe(aiSessionListQuery(), params)
-		const [fanOut, detection] = sql.split("TraceId IN (SELECT")
+		// `orgId`, `fanOutStart`, `fanOutEnd` and nothing else: a `startTime` param
+		// left in the query would compile against a value this call never passes,
+		// and the whole point is that no level here sees the caller's range.
+		expect(sql).not.toContain(params.startTime)
+		expect(sql).not.toContain(params.endTime)
+		expect(sql).not.toContain("__PARAM_")
+	})
 
-		// `trace_detail_spans` is PARTITION BY toDate(Timestamp), so this predicate
-		// is the only thing standing between a seek over the window's partitions
-		// and a seek over every partition the 30-day TTL retains.
-		expect(fanOut).toContain(`Timestamp >= '${params.startTime}' - INTERVAL 86400 SECOND`)
-		expect(fanOut).toContain(`Timestamp <= '${params.endTime}' + INTERVAL 86400 SECOND`)
-		// Detection stays exact: the pad keeps a straddling trace whole, it does not
-		// widen which sessions the range reports.
-		expect(detection).toContain(`Timestamp >= '${params.startTime}'`)
-		expect(detection).toContain(`Timestamp <= '${params.endTime}'`)
-		expect(detection).not.toContain("INTERVAL")
+	it("refuses an empty page rather than compiling `IN ()`", () => {
+		// Not a failure the caller recovers from: `IN ()` is not SQL, and a caller
+		// holding an empty page already knows to answer it without this read.
+		expect(() => aiSessionListQuery({ sessionIds: [] })).toThrow(QueryBuilderDefect)
+		expect(() => aiSessionListQuery({ sessionIds: [] })).toThrow(
+			/needs the page's session ids/,
+		)
 	})
 
 	it("leaves no unresolved param placeholder", () => {
-		expect(compileUnsafe(aiSessionListQuery(), params).sql).not.toContain("__PARAM_")
+		expect(compileUnsafe(aiSessionListQuery(listOpts), listParams).sql).not.toContain("__PARAM_")
 	})
 
 	it("decodes quoted 64-bit aggregates and the service-name array", () => {
-		const compiled = compileUnsafe(aiSessionListQuery(), params)
+		const compiled = compileUnsafe(aiSessionListQuery(listOpts), listParams)
 
 		const [row] = decodeRows(compiled, [
 			{
diff --git a/packages/query-engine-integrations/src/ai/ai-sessions.ts b/packages/query-engine-integrations/src/ai/ai-sessions.ts
index 5c4accb0d..c43680f2b 100644
--- a/packages/query-engine-integrations/src/ai/ai-sessions.ts
+++ b/packages/query-engine-integrations/src/ai/ai-sessions.ts
@@ -23,49 +23,68 @@
 // classified as GenAI, so it is the marker that finds both populations, and the
 // session id becomes a grouping key rather than an admission test.
 //
-// Both queries are that fan-out, in two stages against two different tables:
+// The list is that fan-out, in two stages against two different tables — and
+// two round trips, because the second is bounded by what the first found:
 //
-//   detect  — `ai_trace_index`, the filtered projection holding ONLY the
-//     vendor-stamped spans (~0.01% of rows), pre-extracted to plain columns by
-//     `ai_trace_index_mv`. Detection used to run the vendor predicate against
-//     raw `traces` behind the `mapKeys(SpanAttributes)` bloom skip index, and
-//     that shape cannot be saved: GenAI spans arrive continuously — about one
-//     per index granule at production volume — so the bloom prunes nothing and
-//     the scan reads the fat Map column for EVERY span in the window. Measured
-//     2026-08-29 in production: ~3.6s for a one-hour window, dead at the 15s
-//     kill by a day. The index is the same predicate applied at insert time;
-//     scanning it costs ~10k narrow rows per day. This stage yields the
-//     qualifying trace-id set and nothing else.
-//   fan out — `trace_detail_spans`, restricted by `TraceId IN (…)`. `TraceId` is
-//     a sort-key prefix there (`(OrgId, TraceId, SpanId)`), so this is a seek.
-//     The same fan-out against raw `traces` times out at 10s on a 7-day window
-//     in production: that table is sorted `(OrgId, ServiceName, SpanName,
-//     Timestamp)` and `idx_trace_id` is only a bloom skip index, which prunes
-//     far too little at this org's volume.
+//   page — `aiSessionPageQuery`, over `ai_trace_index`: the filtered
+//     projection holding ONLY the vendor-stamped spans (~0.01% of rows),
+//     pre-extracted to plain columns by `ai_trace_index_mv`. It is the one
+//     level that sees the caller's whole window, and the one that can afford
+//     to: ~10k narrow rows a day against 70M raw spans, ~600ms cold over 30
+//     days of production. It resolves every trace to its session key, ranks
+//     the sessions by their first agent span, and yields one page of session
+//     ids with the bounds of their agent spans — nothing else.
+//   fan out — `aiSessionListQuery`, over `trace_detail_spans`, restricted to
+//     that page's traces by `TraceId IN (…)`. `TraceId` is a sort-key prefix
+//     there (`(OrgId, TraceId, SpanId)`), so this is a seek. The same fan-out
+//     against raw `traces` times out at 10s on a 7-day window in production:
+//     that table is sorted `(OrgId, ServiceName, SpanName, Timestamp)` and
+//     `idx_trace_id` is only a bloom skip index, which prunes far too little
+//     at this org's volume.
+//
+// Detection used to run the vendor predicate against raw `traces` behind the
+// `mapKeys(SpanAttributes)` bloom skip index, and that shape cannot be saved:
+// GenAI spans arrive continuously — about one per index granule at production
+// volume — so the bloom prunes nothing and the scan reads the fat Map column
+// for EVERY span in the window. Measured 2026-08-29 in production: ~3.6s for a
+// one-hour window, dead at the 15s kill by a day. The index is the same
+// predicate applied at insert time.
+//
+// The fan-out used to run over every qualifying trace in the window and page
+// with LIMIT/OFFSET afterwards, and that shape cannot be saved either — by the
+// index or by anything else. `trace_detail_spans` sits on object storage, and
+// what a seek costs there is set by the partitions it touches, not by the rows
+// it returns: measured 2026-09-02 in production, five trace ids across the 32
+// retained partitions ran past 10s, while a page's worth of sessions inside
+// one partition took 1.3–2.8s cold and ~300ms warm. The old shape read three
+// partitions for a one-day window (5–15s, eight of 31 reads killed at the 15s
+// ceiling over three days) and all of them for the 30-day window the page
+// offers (killed, every time). Paging first is what bounds the fan-out to the
+// hours a page spans.
 //
 // `IN` rather than a JOIN for the fan-out, the same reason
 // `errorDetailTracesQuery` uses it — ClickHouse pushes the id set into the
-// read, which a JOIN does not do.
+// read, which a JOIN does not do. The JOIN the list DOES carry is one level up,
+// between two derived tables of at most a page of traces each, and it is there
+// so the fan-out takes the session key the page resolved rather than deriving
+// its own from the spans: two derivations over two windows can disagree, and
+// a disagreement would drop the row from the page it was ranked into.
 //
 // The index fills forward from its deploy: rows already in `traces` when the MV
-// was created are not in it until a backfill runs, so detection (and the
+// was created are not in it until a backfill runs, so the page (and the
 // facets) can under-report windows that predate the deploy. The fan-out and the
-// per-session reads still see every span of any trace detection finds.
-//
-// The window predicate sits on BOTH levels, and the fan-out's copy is PADDED
-// rather than exact. That is what reconciles the two demands on it:
-// `trace_detail_spans` is `PARTITION BY toDate(Timestamp)`, so the predicate is
-// the only thing that prunes partitions there, while an exact copy of the
-// window would clamp the reported start/end of every session that began before
-// the range. A day of padding costs one extra partition on each side and
-// contains any trace shorter than 24h.
+// per-session reads still see every span of any trace the page finds.
 //
-// What omitting it costs is invisible warm and severe cold. Measured against
-// production, one fixed set of 20 trace ids whose parts were not cached:
-// 328ms with an exact window, 1,089ms with the padded one, 8,389ms with no
-// predicate at all — while re-running all three against warm parts puts them
-// within ~100ms of each other. Cold is the normal state of a dashboard query
-// against a month of partitions, and the `list` profile kills it at 15s.
+// The window predicate on the fan-out is the PAGE's, not the caller's: the
+// bounds of the page's agent spans, padded by `FAN_OUT_PAD_SECONDS` so a
+// trace's non-agent spans on either side are counted too. `trace_detail_spans`
+// is `PARTITION BY toDate(Timestamp)`, so the predicate is the only thing that
+// prunes partitions there. What that buys depends on how densely an org runs
+// agents: at production volume a page of sessions ordered by start spans
+// hours — one partition, two around midnight — while an org with a few
+// sessions a day has a first page that spans weeks, and its fan-out probes
+// every partition in between exactly as the old shape did (no worse, no
+// better; a chunked or per-partition fan-out is the follow-up if that bites).
 //
 // A caller that has no window — a deep link carrying only a session id —
 // resolves one with `aiSessionWindowQuery` first, rather than running the
@@ -93,6 +112,7 @@ import {
 	fromQuery,
 	inSubquery,
 	param,
+	QueryBuilderDefect,
 	unionAll,
 	type CHUnionQuery,
 	type ColumnAccessor,
@@ -130,13 +150,29 @@ const FAILED_RESPONSE_STATUSES = ["failed", "error"]
 const SESSION_ORDER_SENTINEL = "2106-01-01 00:00:00"
 
 /**
- * How far past the caller's window the `trace_detail_spans` fan-out reads, in
- * seconds. See this file's header: the point is a predicate ClickHouse can
- * prune partitions with, not an exact bound, so the pad is chosen to be a whole
- * partition (`PARTITION BY toDate(Timestamp)`) and to contain any trace that
- * straddles the window edge.
+ * How far past the page's own agent-span bounds the `trace_detail_spans`
+ * fan-out reads, in seconds — see `aiSessionListQuery`.
+ *
+ * The pad exists because a trace's non-agent spans lie outside its agent
+ * spans: measured over two days of production (4,920 agent traces,
+ * 2026-09-02), the first span leads the first agent span by at most 0.1s and
+ * the last span trails the last agent span by at most 10 minutes. An hour
+ * contains both with room to spare and keeps the read inside one partition
+ * (`PARTITION BY toDate(Timestamp)`) for every page but the ones nearest
+ * midnight; a day would make it three partitions every time, which is what
+ * the fan-out's cost is made of. A trace whose spans reach further than an
+ * hour past its agent spans is clamped in the list row alone.
+ */
+const FAN_OUT_PAD_SECONDS = 3_600
+
+/**
+ * The pad on the bounds `aiSessionWindowQuery`/`aiTraceWindowQuery` report for
+ * a deep link, in seconds. A whole partition rather than the list's hour: that
+ * path resolves ONE session, so the extra partitions cost one read, and the
+ * detail page shows the spans themselves — clamping there would cut a
+ * transcript, where the list would only under-count a cell.
  */
-const FAN_OUT_PAD_SECONDS = 86_400
+const WINDOW_PAD_SECONDS = 86_400
 
 /** ClickHouse returns `''` for a missing Map key, so presence needs both halves. */
 const hasSessionId = (attrs: CH.Expr<Record<string, string>>, get: CH.Expr<string>) =>
@@ -164,15 +200,45 @@ const orderTuple = (...parts: ReadonlyArray<unknown>): CH.Expr<unknown> =>
 const sessionKey = (rawSessionId: CH.Expr<string>, traceId: CH.Expr<string>): CH.Expr<string> =>
 	CH.if_(rawSessionId.eq(""), CH.concat(MAPLE_AI_TRACE_SESSION_PREFIX, traceId), rawSessionId)
 
-export interface AiSessionListOpts {
+/** The filters the page and the list share; both apply them on `ai_trace_index`. */
+export interface AiSessionFilterOpts {
+	readonly vendorIds?: readonly string[]
+	readonly serviceNames?: readonly string[]
+}
+
+export interface AiSessionPageOpts extends AiSessionFilterOpts {
 	/** Sessions returned, most recently started first. */
 	readonly limit?: number
 	/** Sessions skipped before `limit` applies — the list's next page. */
 	readonly offset?: number
-	readonly vendorIds?: readonly string[]
-	readonly serviceNames?: readonly string[]
 }
 
+export interface AiSessionPageOutput {
+	/** The vendor's own session id, or `trace:<TraceId>` for a trace that has
+	 *  none — see `MAPLE_AI_TRACE_SESSION_PREFIX`. */
+	readonly sessionId: string
+	/** Bounds of the session's agent spans inside the window — warehouse
+	 *  datetime literals, the shape `aiSessionListQuery`'s `fanOutStart` and
+	 *  `fanOutEnd` params take back. */
+	readonly agentStart: string
+	readonly agentEnd: string
+}
+
+export interface AiSessionListOpts extends AiSessionFilterOpts {
+	/**
+	 * The page to aggregate, as `aiSessionPageQuery` ranked it — under the same
+	 * filters, or the two stages resolve traces differently. Never empty: an
+	 * empty page is answered without this read, and an empty list here is a
+	 * defect rather than an empty result.
+	 */
+	readonly sessionIds: readonly string[]
+}
+
+/** Which pair of params bounds an `ai_trace_index` read: the caller's window
+ *  (`startTime`/`endTime`) or the page's (`fanOutStart`/`fanOutEnd`). */
+type IndexBounds = "window" | "page"
+
+
 export interface AiSessionListOutput {
 	/** The vendor's own session id, or `trace:<TraceId>` for a trace that has
 	 *  none — see `MAPLE_AI_TRACE_SESSION_PREFIX`. */
@@ -191,7 +257,63 @@ export interface AiSessionListOutput {
 }
 
 /**
- * One row per AI agent session in the window.
+ * One row per agent trace in the window, off `ai_trace_index` alone: the
+ * trace's session key and the bounds of its agent spans. The level both stages
+ * share, so a trace resolves to the same session in the page and in the
+ * aggregation.
+ *
+ * The page reads it over the caller's window; the aggregation over the page's
+ * own bounds, which is the same thing for every trace ON the page: a page
+ * trace's index rows all lie between its session's `agentStart` and `agentEnd`,
+ * and the page's bounds contain every session's. So the key and the filters
+ * come out identical, from hours of the index rather than the caller's month.
+ * A trace that is NOT on the page can key differently over the narrower read,
+ * and is then discarded by the page's key list like any other — unless it
+ * carries two session ids and only the lesser one falls inside the page's
+ * bounds, which no vendor has been seen to produce.
+ *
+ * `rawSessionId` is `max` over the trace's index rows because the session id
+ * sits on the turn-owning span alone (see the file header) and every other row
+ * reads `''`, which `max` discards.
+ *
+ * A filter is a TRACE-level existence test — "some agent span of the trace
+ * carries this value" — applied after the grouping, not a row predicate before
+ * it. A row predicate would also narrow the rows `rawSessionId` is read from,
+ * and a vendor filter would then file a trace under `trace:` whenever its
+ * session-bearing span belongs to another vendor — an eve agent calling through
+ * the Vercel AI SDK carries both. The population is the one the facets count:
+ * `aiSessionFacetsQuery` also collects per trace and counts any-span.
+ */
+const indexTraces = (opts: AiSessionFilterOpts, bounds: IndexBounds) =>
+	from(AiTraceIndex)
+		.select(($) => ({
+			traceId: $.TraceId,
+			rawSessionId: CH.max_($.SessionId),
+			// Named apart from the page's `agentStart`/`agentEnd`: an outer alias
+			// shadows the derived table's column of the same name, so `min(…)` of
+			// it would resolve to the outer `toString(…)` String and fail — see
+			// `traceStart` in `aiSessionListQuery`.
+			traceAgentStart: CH.min_($.Timestamp),
+			traceAgentEnd: CH.max_($.Timestamp),
+		}))
+		.where(($) => [
+			$.OrgId.eq(param.string("orgId")),
+			$.Timestamp.gte(param.dateTimeString(bounds === "window" ? "startTime" : "fanOutStart")),
+			$.Timestamp.lte(param.dateTimeString(bounds === "window" ? "endTime" : "fanOutEnd")),
+		])
+		.groupBy("traceId")
+		.having(($) => [
+			opts.vendorIds?.length ? CH.countIf(CH.inList($.VendorId, opts.vendorIds)).gt(0) : undefined,
+			opts.serviceNames?.length
+				? CH.countIf(CH.inList($.ServiceName, opts.serviceNames)).gt(0)
+				: undefined,
+		])
+
+/**
+ * The page: which sessions the list shows, in what order, and where their
+ * agent spans lie — the first of the list's two reads, and the only one that
+ * sees the caller's whole window. See the file header for why the fan-out
+ * cannot.
  *
  * Detection admits any trace with a GenAI span, and the session id then groups
  * rather than admits: a trace that carries one is filed under it — with the
@@ -199,6 +321,50 @@ export interface AiSessionListOutput {
  * its own, keyed `trace:<TraceId>`. Sessionless is the normal state for whole
  * vendors, not an edge case; see this file's header.
  *
+ * Ordered by the first AGENT span, not the first span of any kind: the index
+ * carries only agent spans, and the two differ by under a second in practice
+ * (see `FAN_OUT_PAD_SECONDS`). The row's `startTime` still reports the true
+ * first span; the caller keeps the page in this order rather than re-sorting
+ * by it, so what is shown is the order that was paged. `sessionId` breaks
+ * ties, so a page boundary never splits two sessions that share a start.
+ *
+ * The remaining gap is between traces, not inside one: a session whose OTHER
+ * traces lie entirely outside the range is still found only by the traces that
+ * touched it, which needs a session-keyed table to fix and not a wider window.
+ */
+export function aiSessionPageQuery(opts: AiSessionPageOpts = {}) {
+	const limit = opts.limit ?? 50
+	const offset = opts.offset ?? 0
+
+	const page = fromQuery(indexTraces(opts, "window"), "index_traces")
+		.select(($) => ({
+			// The grouping key, and the only level that can compute it: the
+			// derived table is one row per trace, so a trace with no session id
+			// of its own becomes a session of one trace here rather than joining
+			// every other sessionless trace under `''`.
+			sessionId: sessionKey($.rawSessionId, $.traceId),
+			agentStart: CH.toString_(CH.min_($.traceAgentStart)),
+			agentEnd: CH.toString_(CH.max_($.traceAgentEnd)),
+		}))
+		.groupBy("sessionId")
+		// A String order, and a correct one: the literal is fixed-width
+		// `YYYY-MM-DD hh:mm:ss.nnnnnnnnn`, so it sorts as the instant does.
+		.orderBy(["agentStart", "desc"], ["sessionId", "asc"])
+		.limit(limit)
+	// Only a positive offset is emitted: `OFFSET 0` is a no-op that would still
+	// change the compiled SQL of every first-page read.
+	return (offset > 0 ? page.offset(offset) : page).format("JSON")
+}
+
+/**
+ * One row per session of the page `aiSessionPageQuery` ranked, with every
+ * fact the list shows that the index cannot answer — the second of the list's
+ * two reads. Bounded by the page alone: `fanOutStart`/`fanOutEnd` are the
+ * extent of the page's agent spans, and both the index reads and the fan-out
+ * run inside them (padded, for the fan-out) — see `indexTraces` for why the
+ * index level resolves a page trace exactly as the page did without the
+ * caller's window.
+ *
  * `vendorId` is the vendor of the EARLIEST span that carries a session id, not
  * `max(vendorId)`. A single trace legitimately carries several vendors — an eve
  * agent calls through the Vercel AI SDK — and `max` picked `vercel_ai_sdk`
@@ -208,48 +374,45 @@ export interface AiSessionListOutput {
  * with no session-bearing span falls to the next rank of the same ordering —
  * its earliest vendor-stamped span, which is that trace's root-most agent span.
  *
- * The vendor filter goes on the detection subquery: it is the level the index
- * serves, and it is the only place `maple_ai.vendor.id` is unambiguous —
- * a trace's other spans carry other vendors, or none.
+ * The filters land on the index level, which means "the trace's agent spans
+ * came from this service" rather than "the trace touched this service". A
+ * trace spans services by definition, so the alternative — filtering the
+ * fan-out — would silently drop spans and under-count `spanCount`. The agent
+ * spans come from the agent's own service, which is the one a user filtering by
+ * service means.
  *
- * The service filter goes there too, which means "the trace's agent spans came
- * from this service" rather than "the trace touched this service". A trace spans
- * services by definition, so the alternative — filtering the fan-out — would
- * silently drop spans and under-count `spanCount`. The agent spans come from the
- * agent's own service, which is the one a user filtering by service means.
+ * Once a trace is on the page it is aggregated across the padded fan-out window
+ * rather than the caller's, so `startTime`/`endTime`/`durationMs`/`spanCount`/
+ * `errorSpanCount`/`serviceNames` describe the whole trace rather than the
+ * slice of it that fell inside the range — a session that began an hour before
+ * the range no longer reports the range edge as its start, and the detail page
+ * can read the bounds this row carries as the session's own.
  *
- * The time window bounds DETECTION exactly and the fan-out loosely. Once a trace
- * qualifies it is aggregated across the padded window rather than the caller's,
- * so `startTime`/`endTime`/`durationMs`/`spanCount`/`errorSpanCount`/
- * `serviceNames` describe the whole trace rather than the slice of it that fell
- * inside the range — a session that began an hour before the range no longer
- * reports the range edge as its start, and the detail page can read the bounds
- * this row carries as the session's own. A trace longer than `FAN_OUT_PAD_SECONDS`
- * is clamped again, which no observed trace comes close to.
- *
- * The remaining gap is between traces, not inside one: a session whose OTHER
- * traces lie entirely outside the range is still found only by the traces that
- * touched it, which needs a session-keyed table to fix and not a wider window.
+ * Ordered by `startTime` for a caller that reads it alone; the list's caller
+ * re-imposes the page's order, which this row cannot know.
  */
-export function aiSessionListQuery(opts: AiSessionListOpts = {}) {
-	const limit = opts.limit ?? 50
-	const offset = opts.offset ?? 0
-
-	// No vendor-presence predicate: `ai_trace_index_mv` admits only spans with a
-	// non-empty vendor id, so membership in the table IS the detection predicate.
-	const sessionTraceIds = from(AiTraceIndex)
-		.select(($) => ({ TraceId: $.TraceId }))
-		.where(($) => [
-			$.OrgId.eq(param.string("orgId")),
-			$.Timestamp.gte(param.dateTimeString("startTime")),
-			$.Timestamp.lte(param.dateTimeString("endTime")),
-			opts.vendorIds?.length ? CH.inList($.VendorId, opts.vendorIds) : undefined,
-			opts.serviceNames?.length ? CH.inList($.ServiceName, opts.serviceNames) : undefined,
-		])
-
-	// Per trace: every span of a qualifying trace, session-bearing or not. The
-	// window is padded here rather than dropped, so "every span" means every span
-	// the trace has, while the read still prunes to a handful of partitions.
+export function aiSessionListQuery(opts: AiSessionListOpts) {
+	// A defect, not a failure: `IN ()` is not SQL, and the caller already knows
+	// its page is empty — see `AiSessionListOpts`.
+	if (opts.sessionIds.length === 0) {
+		throw new QueryBuilderDefect({
+			message: "aiSessionListQuery needs the page's session ids; an empty page is answered without it",
+		})
+	}
+	// The page's traces, keyed as the page keyed them — read twice below, once
+	// as the id set the fan-out seeks by and once joined for the key, over the
+	// page's bounds rather than the caller's window (see `indexTraces`).
+	const onPage = ($: { rawSessionId: CH.Expr<string>; traceId: CH.Expr<string> }) =>
+		CH.inList(sessionKey($.rawSessionId, $.traceId), opts.sessionIds)
+	const pageTraceIds = fromQuery(indexTraces(opts, "page"), "agent_traces")
+		.select(($) => ({ traceId: $.traceId }))
+		.where(($) => [onPage($)])
+	const pageTraces = fromQuery(indexTraces(opts, "page"), "agent_traces")
+		.select(($) => ({ traceId: $.traceId, rawSessionId: $.rawSessionId }))
+		.where(($) => [onPage($)])
+
+	// Per trace: every span of a page trace, session-bearing or not, inside the
+	// page's padded window.
 	const perTrace = from(TraceDetailSpans)
 		.select(($) => {
 			const sessionOrder = CH.if_(
@@ -274,11 +437,6 @@ export function aiSessionListQuery(opts: AiSessionListOpts = {}) {
 			)
 			return {
 				traceId: $.TraceId,
-				// A trace belongs to one session; the non-bearing spans read `''`,
-				// which `max` discards. Named apart from the outer `sessionId`: that
-				// one is this value or a synthesized `trace:` id, and an alias that
-				// referred to itself would be a cyclic alias rather than a fallback.
-				rawSessionId: CH.max_($.SpanAttributes.get(SESSION_ID_ATTR)),
 				vendorId: CH.argMin($.SpanAttributes.get(VENDOR_ID_ATTR), vendorOrder),
 				vendorVersion: CH.argMin($.SpanAttributes.get(VENDOR_VERSION_ATTR), vendorOrder),
 				// Carried so the outer level can order traces by their first
@@ -323,49 +481,42 @@ export function aiSessionListQuery(opts: AiSessionListOpts = {}) {
 		})
 		.where(($) => [
 			$.OrgId.eq(param.string("orgId")),
-			$.Timestamp.gte(CH.intervalSub(param.dateTimeString("startTime"), FAN_OUT_PAD_SECONDS)),
-			$.Timestamp.lte(CH.intervalAdd(param.dateTimeString("endTime"), FAN_OUT_PAD_SECONDS)),
-			inSubquery($.TraceId, sessionTraceIds),
+			$.Timestamp.gte(CH.intervalSub(param.dateTimeString("fanOutStart"), FAN_OUT_PAD_SECONDS)),
+			$.Timestamp.lte(CH.intervalAdd(param.dateTimeString("fanOutEnd"), FAN_OUT_PAD_SECONDS)),
+			inSubquery($.TraceId, pageTraceIds),
 		])
 		.groupBy("traceId")
 
-	const sessions = fromQuery(perTrace, "session_traces")
-		.select(($) => ({
-			// The grouping key, and the only level that can compute it: the
-			// derived table is one row per trace, so a trace with no session id
-			// of its own becomes a session of one trace here rather than joining
-			// every other sessionless trace under `''`.
-			sessionId: sessionKey($.rawSessionId, $.traceId),
-			vendorId: CH.argMin($.vendorId, $.sessionStart),
-			vendorVersion: CH.argMin($.vendorVersion, $.sessionStart),
-			// `count()`, not `uniq()`: the derived table already emits exactly one
-			// row per trace, so this is exact and cheaper — `uniq` is an
-			// approximate HLL that would start drifting on a very large session.
-			traceCount: CH.count(),
-			spanCount: CH.sum($.spanCount),
-			errorSpanCount: CH.sum($.errorSpanCount),
-			serviceNames: CH.groupUniqArrayArray($.serviceNames),
-			startTime: CH.toString_(CH.min_($.traceStart)),
-			endTime: CH.toString_(fromUnixTimestamp64Nano(CH.max_($.traceEndNanos))),
-			// Nanoseconds first: `Timestamp` is DateTime64(9), and subtracting two
-			// of them yields a Decimal whose scale the wire format then quotes.
-			// Wrapped in `intDiv` because `Expr.sub`/`div` do not parenthesize.
-			durationMs: CH.intDiv(
-				CH.max_($.traceEndNanos).sub(CH.toUnixTimestamp64Nano(CH.min_($.traceStart))),
-				1_000_000,
-			),
-		}))
-		// No `sessionId != ''` guard any more, and none needed: the key is never
-		// empty. It used to keep two unrelated populations apart — a trace whose
-		// session-bearing span has landed in `traces` but not yet in the
-		// `trace_detail_spans` MV read back empty, and every such trace grouped
-		// together under one blank session. Both now key on their own trace id.
-		.groupBy("sessionId")
-		.orderBy(["startTime", "desc"])
-		.limit(limit)
-	// Only a positive offset is emitted: `OFFSET 0` is a no-op that would still
-	// change the compiled SQL of every first-page read.
-	return (offset > 0 ? sessions.offset(offset) : sessions).format("JSON")
+	return (
+		fromQuery(perTrace, "session_traces")
+			.innerJoinQuery(pageTraces, "index_traces", (t, i) => t.traceId.eq(i.traceId))
+			.select(($) => ({
+				// The key the page resolved, not one re-derived from the spans — see
+				// the file header for why the two must not be allowed to differ.
+				sessionId: sessionKey($.index_traces.rawSessionId, $.traceId),
+				vendorId: CH.argMin($.vendorId, $.sessionStart),
+				vendorVersion: CH.argMin($.vendorVersion, $.sessionStart),
+				// `count()`, not `uniq()`: the derived table already emits exactly one
+				// row per trace, so this is exact and cheaper — `uniq` is an
+				// approximate HLL that would start drifting on a very large session.
+				traceCount: CH.count(),
+				spanCount: CH.sum($.spanCount),
+				errorSpanCount: CH.sum($.errorSpanCount),
+				serviceNames: CH.groupUniqArrayArray($.serviceNames),
+				startTime: CH.toString_(CH.min_($.traceStart)),
+				endTime: CH.toString_(fromUnixTimestamp64Nano(CH.max_($.traceEndNanos))),
+				// Nanoseconds first: `Timestamp` is DateTime64(9), and subtracting two
+				// of them yields a Decimal whose scale the wire format then quotes.
+				// Wrapped in `intDiv` because `Expr.sub`/`div` do not parenthesize.
+				durationMs: CH.intDiv(
+					CH.max_($.traceEndNanos).sub(CH.toUnixTimestamp64Nano(CH.min_($.traceStart))),
+					1_000_000,
+				),
+			}))
+			.groupBy("sessionId")
+			.orderBy(["startTime", "desc"])
+			.format("JSON")
+	)
 }
 
 // List facets (UNION ALL — vendor / service)
@@ -458,7 +609,7 @@ export interface AiSessionWindowOutput {
  * `mapValues(SpanAttributes)` for the id to prune with, and the table's 30-day
  * TTL caps what is left. The fan-out has neither and must not be run that way.
  *
- * The bounds come back padded by `FAN_OUT_PAD_SECONDS`, because they are
+ * The bounds come back padded by `WINDOW_PAD_SECONDS`, because they are
  * measured over the session-BEARING spans while the read they bound returns
  * every span of those spans' traces — a trace whose first span is not the
  * session-bearing one starts earlier than any window this could report exactly.
@@ -469,8 +620,8 @@ export interface AiSessionWindowOutput {
 export function aiSessionWindowQuery() {
 	return from(Traces)
 		.select(($) => ({
-			startTime: CH.toString_(CH.intervalSub(CH.min_($.Timestamp), FAN_OUT_PAD_SECONDS)),
-			endTime: CH.toString_(CH.intervalAdd(CH.max_($.Timestamp), FAN_OUT_PAD_SECONDS)),
+			startTime: CH.toString_(CH.intervalSub(CH.min_($.Timestamp), WINDOW_PAD_SECONDS)),
+			endTime: CH.toString_(CH.intervalAdd(CH.max_($.Timestamp), WINDOW_PAD_SECONDS)),
 			spanCount: CH.count(),
 		}))
 		.where(($) => [
@@ -500,8 +651,8 @@ export function aiSessionWindowQuery() {
 export function aiTraceWindowQuery() {
 	return from(Traces)
 		.select(($) => ({
-			startTime: CH.toString_(CH.intervalSub(CH.min_($.Timestamp), FAN_OUT_PAD_SECONDS)),
-			endTime: CH.toString_(CH.intervalAdd(CH.max_($.Timestamp), FAN_OUT_PAD_SECONDS)),
+			startTime: CH.toString_(CH.intervalSub(CH.min_($.Timestamp), WINDOW_PAD_SECONDS)),
+			endTime: CH.toString_(CH.intervalAdd(CH.max_($.Timestamp), WINDOW_PAD_SECONDS)),
 			spanCount: CH.count(),
 		}))
 		.where(($) => [$.OrgId.eq(param.string("orgId")), $.TraceId.eq(param.string("traceId"))])
diff --git a/packages/query-engine-integrations/src/ai/index.ts b/packages/query-engine-integrations/src/ai/index.ts
index 9db68d007..16092bddc 100644
--- a/packages/query-engine-integrations/src/ai/index.ts
+++ b/packages/query-engine-integrations/src/ai/index.ts
@@ -10,14 +10,18 @@
 export {
 	aiSessionFacetsQuery,
 	aiSessionListQuery,
+	aiSessionPageQuery,
 	aiSessionSpansQuery,
 	aiSessionSpansRowSchema,
 	aiSessionWindowQuery,
 	aiTraceSpansQuery,
 	aiTraceWindowQuery,
 	type AiSessionFacetsOutput,
+	type AiSessionFilterOpts,
 	type AiSessionListOpts,
 	type AiSessionListOutput,
+	type AiSessionPageOpts,
+	type AiSessionPageOutput,
 	type AiSessionSpansOpts,
 	type AiSessionSpansOutput,
 	type AiSessionWindowOutput,
diff --git a/packages/query-engine-integrations/src/catalog.ts b/packages/query-engine-integrations/src/catalog.ts
index 7c4487f9e..6875ff4b1 100644
--- a/packages/query-engine-integrations/src/catalog.ts
+++ b/packages/query-engine-integrations/src/catalog.ts
@@ -10,6 +10,7 @@
 
 import { Effect } from "effect"
 import { compileUnionUnsafe, compileUnsafe, type CompiledQuery } from "@maple/query-engine/ch"
+import { MAPLE_AI_TRACE_SESSION_PREFIX } from "@maple/domain/gen-ai"
 import * as CH from "./index"
 
 export interface IntegrationFixture {
@@ -56,7 +57,45 @@ const traceWindow = {
 	parentStart: "2026-01-01 08:30:00",
 }
 
+/** One page of sessions, as `aiSessionPageQuery` hands them to the fan-out: the
+ *  ids, and the extent of their agent spans. The bounds sit inside `window`
+ *  because the page was ranked inside it. */
+const AI_PAGE_SESSION_IDS = ["wrun_sql_catalog", `${MAPLE_AI_TRACE_SESSION_PREFIX}${AI_TRACE_ID}`]
+
+/** Stage two's whole param set — it never sees the caller's window. */
+const aiPageBounds = {
+	orgId: ORG_ID,
+	fanOutStart: "2026-01-02 10:30:00",
+	fanOutEnd: "2026-01-02 12:30:00",
+}
+
 export const integrationFixtures: ReadonlyArray<IntegrationFixture> = [
+	{
+		// The AI sessions list is two reads. This one ranks the page on
+		// `ai_trace_index` alone and is the only one that sees the caller's window.
+		module: "ai-sessions",
+		name: "aiSessionPageQuery",
+		label: "default",
+		compile: () => compileUnsafe(CH.aiSessionPageQuery(), window),
+	},
+	{
+		// The vendor/service filters the AI sessions list page sends, on its
+		// second page. They land here and are repeated verbatim on the fan-out, or
+		// the two stages resolve traces differently.
+		module: "ai-sessions",
+		name: "aiSessionPageQuery",
+		label: "filtered",
+		compile: () =>
+			compileUnsafe(
+				CH.aiSessionPageQuery({
+					limit: 25,
+					offset: 25,
+					vendorIds: ["eve"],
+					serviceNames: ["maple-slack-agent"],
+				}),
+				window,
+			),
+	},
 	{
 		module: "ai-sessions",
 		name: "aiSessionListQuery",
@@ -64,23 +103,23 @@ export const integrationFixtures: ReadonlyArray<IntegrationFixture> = [
 		// The ClickHouse e2e sweep runs its quoted/unquoted 64-bit decode assertion
 		// for every fixture whose compiled query carries a row schema — which the
 		// builder derives from the SELECT, so nothing is declared here.
-		compile: () => compileUnsafe(CH.aiSessionListQuery(), window),
+		compile: () =>
+			compileUnsafe(CH.aiSessionListQuery({ sessionIds: AI_PAGE_SESSION_IDS }), aiPageBounds),
 	},
 	{
-		// The vendor/service filters the AI sessions list page sends, on its
-		// second page.
+		// The same page under the list's filters — the aggregation runs with the
+		// filters the page ranked under, never without them.
 		module: "ai-sessions",
 		name: "aiSessionListQuery",
 		label: "filtered",
 		compile: () =>
 			compileUnsafe(
 				CH.aiSessionListQuery({
-					limit: 25,
-					offset: 25,
+					sessionIds: AI_PAGE_SESSION_IDS,
 					vendorIds: ["eve"],
 					serviceNames: ["maple-slack-agent"],
 				}),
-				window,
+				aiPageBounds,
 			),
 	},
 	{