perf(ci): read the CI analytics from ci_runs and ci_jobs

.github/workflows/ci.yml

@@ -395,7 +395,7 @@ jobs:
                   src/services/warehouse/alert-window-alignment.clickhouse.e2e.test.ts
                   src/services/warehouse/web-analytics-parity.clickhouse.e2e.test.ts
                   src/services/warehouse/trace-facets-hourly-materialization.clickhouse.e2e.test.ts
-                  src/services/warehouse/ci-runs-materialization.clickhouse.e2e.test.ts
+                  src/services/warehouse/ci-runs.clickhouse.e2e.test.ts
 
             # Every SQL shape the product emits, against the real analyzer: unit tests
             # only compare SQL text, which is how a NO_COMMON_TYPE reached production.

apps/ai/src/mcp/__evals__/integration-world.ts

@@ -245,7 +245,13 @@ export const seedIntegrationWorld = async (rt: EvalRuntime) => {
 		 FROM scrape_targets t WHERE t.source_id = s.id AND s.platform = 'prometheus'`,
 		[],
 	)
-	const reported = (platform: string, okMinutesAgo: number, error: string | null, total: number, unhealthy: number) =>
+	const reported = (
+		platform: string,
+		okMinutesAgo: number,
+		error: string | null,
+		total: number,
+		unhealthy: number,
+	) =>
 		sql(
 			`UPDATE sources SET last_poll_at = $2, last_poll_ok_at = $2, last_poll_error = $3,
 				poll_resources = $4::jsonb WHERE platform = $1`,
@@ -410,13 +416,13 @@ const inWindow = (sql: string, hoursAgo: number) => {
 
 export const integrationFixtureRules = (): FixtureRule[] => [
 	{
-		match: (sql) => sql.includes(" AS pipeline") && sql.includes("cicd.pipeline.name"),
+		match: (sql) => sql.includes("FROM ci_runs"),
 		rows: (sql) =>
 			CI_RUNS.filter((run) => inWindow(sql, run.hoursAgo))
 				.filter((run) => !sql.includes("'failure'") || run.result === "failure")
 				.filter(
 					(run) =>
-						!sql.includes("vcs.repository.name'] = '") || sql.includes(`= '${run.repository}'`),
+						!sql.includes("Repository = '") || sql.includes(`Repository = '${run.repository}'`),
 				)
 				.map((run) => ({
 					traceId: run.traceId,
@@ -427,16 +433,16 @@ export const integrationFixtureRules = (): FixtureRule[] => [
 					repository: run.repository,
 					branch: "main",
 					revision: run.revision,
-					attempt: "1",
+					attempt: 1,
 					trigger: "push",
 					actor: "ada",
-					pullRequest: "",
+					pullRequest: 0,
 					durationMs: run.minutes * 60_000,
 					startedAt: wire(Date.now() - run.hoursAgo * 3_600_000),
 				})),
 	},
 	{
-		match: (sql) => sql.includes("maple.cicd.failure.reason"),
+		match: (sql) => sql.includes("FROM ci_jobs"),
 		rows: CI_RUNS.filter((run) => run.result === "failure").map((run) => ({
 			traceId: run.traceId,
 			job: CI_WORLD.failingJob,

apps/ai/src/mcp/tools/ci-runs.ts

@@ -2,7 +2,8 @@ import { DateTime, Effect, Schema } from "effect"
 import { CiRunsOutput } from "@maple/domain/mcp-outputs"
 import { CH, formatWarehouseDateTime } from "@maple/query-engine"
 import { WarehouseExecutor } from "@maple/query-engine/observability"
-import type { McpToolRegistrar } from "./types"
+import { isMissingCiTables } from "@maple/backend/services/warehouse/missing-table"
+import { McpUnavailableError, type McpToolRegistrar } from "./types"
 import { warehouseToMcpHandlers } from "../lib/map-warehouse-error"
 import { withTenantExecutor, CurrentMcpTenant } from "../lib/query-warehouse"
 import { MCP_DISCOVERY_MAX_HOURS } from "../lib/time"
@@ -29,8 +30,17 @@ const formatRunDuration = (ms: number) => {
 const quantile = (sorted: ReadonlyArray<number>, q: number) =>
 	sorted.length === 0 ? 0 : (sorted[Math.min(sorted.length - 1, Math.floor(q * sorted.length))] ?? 0)
 
-const summarize = (runs: ReadonlyArray<CH.CiRunsOutput>): Output["pipelines"] => {
-	const groups = new Map<string, Array<CH.CiRunsOutput>>()
+interface SummarizedRun {
+	readonly repository: string
+	readonly pipeline: string
+	readonly result: string
+	readonly revision: string
+	readonly durationMs: number
+	readonly startedAt: DateTime.Utc
+}
+
+const summarize = (runs: ReadonlyArray<SummarizedRun>): Output["pipelines"] => {
+	const groups = new Map<string, Array<SummarizedRun>>()
 	for (const run of runs) {
 		const key = `${run.repository}\u0000${run.pipeline}`
 		groups.set(key, [...(groups.get(key) ?? []), run])
@@ -146,7 +156,19 @@ export function registerCiRunsTool(server: McpToolRegistrar) {
 						failedJobsCap: failedIds.length * FAILED_JOBS_PER_RUN,
 					}
 				}),
-			).pipe(Effect.catchTags(warehouseToMcpHandlers(TOOL)))
+			).pipe(
+				// A self-hosted ClickHouse gets the CI tables only once an admin applies the schema.
+				Effect.catchIf(isMissingCiTables, () =>
+					Effect.fail(
+						new McpUnavailableError({
+							message:
+								"This org's warehouse has no CI tables yet. On a self-hosted ClickHouse, apply the schema from its ClickHouse settings page.",
+							capability: "ci_runs",
+						}),
+					),
+				),
+				Effect.catchTags(warehouseToMcpHandlers(TOOL)),
+			)
 			yield* Effect.annotateCurrentSpan("result.rowCount", runs.length)
 
 			return {
@@ -164,7 +186,7 @@ export function registerCiRunsTool(server: McpToolRegistrar) {
 					branch: run.branch,
 					revision: run.revision,
 					runId: run.runId,
-					attempt: run.attempt,
+					attempt: Math.max(run.attempt, 1),
 					trigger: run.trigger,
 					result: run.result,
 					durationMs: run.durationMs,
@@ -316,7 +338,7 @@ export function registerCiRunsTool(server: McpToolRegistrar) {
 										run.pipeline,
 										run.branch,
 										run.revision,
-										run.attempt !== "" && run.attempt !== "1"
+										run.attempt > 1
 											? `${run.result} (attempt ${run.attempt})`
 											: run.result,
 										formatRunDuration(run.durationMs),

apps/api/scripts/bench-suites/ci.ts

@@ -0,0 +1,64 @@
+/**
+ * The CI page's reads (`ci_runs` / `ci_jobs`) over a week and a month. Edit the org and windows
+ * to a populated snapshot; the 2026-10-22 table layout choice was measured on this shape.
+ */
+import * as CH from "@maple/query-engine/ch"
+import * as Bench from "@maple-dev/effect-orm/benchmark"
+import { OrgId } from "@maple/domain"
+
+const END = "2026-10-22 22:58:00"
+const WINDOWS = { "7d": "2026-10-15 22:58:00", "30d": "2026-09-22 22:58:00" } as const
+
+const cases = (label: keyof typeof WINDOWS) => {
+	const inputs = { orgId: OrgId.make("org_sql_catalog"), startTime: WINDOWS[label], endTime: END }
+	return [
+		Bench.query({
+			id: `ci/run-totals/${label}`,
+			inputs,
+			compile: (i) => CH.compile(CH.ciRunTotalsQuery({}), i),
+			results: "unordered",
+		}),
+		Bench.query({
+			id: `ci/run-series/${label}`,
+			inputs: { ...inputs, bucketSeconds: 21_600 },
+			compile: (i) => CH.compile(CH.ciRunSeriesQuery({}), i),
+			results: "unordered",
+		}),
+		Bench.query({
+			id: `ci/pipeline-stats/${label}`,
+			inputs,
+			compile: (i) => CH.compile(CH.ciPipelineStatsQuery({}), i),
+			results: "unordered",
+		}),
+		Bench.query({
+			id: `ci/job-stats/${label}`,
+			inputs,
+			compile: (i) => CH.compile(CH.ciJobStatsQuery({}), i),
+			results: "unordered",
+		}),
+		Bench.query({
+			id: `ci/queue-totals/${label}`,
+			inputs,
+			compile: (i) => CH.compile(CH.ciQueueTotalsQuery({}), i),
+			results: "unordered",
+		}),
+		Bench.query({
+			id: `ci/runs-list/${label}`,
+			inputs,
+			compile: (i) => CH.compile(CH.ciRunsQuery({ limit: 51 }), i),
+			results: "ordered",
+		}),
+		Bench.query({
+			id: `ci/pull-request-runs/${label}`,
+			inputs,
+			compile: (i) => CH.compile(CH.ciRunsQuery({ pullRequest: 1234, limit: 100 }), i),
+			results: "ordered",
+		}),
+	]
+}
+
+export default Bench.defineSuite({
+	name: "ci-workload",
+	dataset: "replace-with-your-snapshot-revision",
+	cases: [...cases("7d"), ...cases("30d")],
+})

packages/backend/src/services/integrations/ci-traces/otlp.ts

@@ -14,8 +14,8 @@ import {
 	type WorkflowRunJob,
 	type WorkflowStep,
 } from "@maple/domain/http"
+import { CI_SERVICE_NAME } from "@maple/domain/tinybird"
 
-export const CI_SERVICE_NAME = "github-actions"
 export const SCOPE_NAME = "@maple/ci-traces"
 
 const SPAN_KIND_INTERNAL = 1

packages/backend/src/services/pr-review/CiAnalyticsService.test.ts

@@ -40,10 +40,10 @@ const run = (n: number, result: string, startedAt: number, extra: Record<string,
 	repository: "web",
 	branch: "main",
 	revision: `sha${n}`,
-	attempt: "1",
+	attempt: 1,
 	trigger: "push",
 	actor: "ada",
-	pullRequest: "",
+	pullRequest: 0,
 	durationMs: 60_000,
 	startedAt: wire(startedAt),
 	...extra,
@@ -136,8 +136,8 @@ describe("CiAnalyticsService.analytics", () => {
 				})
 				assert.isAbove(reads.length, 0)
 				for (const read of reads) {
-					assert.include(read.sql, "ResourceAttributes['vcs.owner.name'] = 'acme'")
-					assert.include(read.sql, "ResourceAttributes['vcs.repository.name'] = 'web'")
+					assert.include(read.sql, ".Owner = 'acme'")
+					assert.include(read.sql, ".Repository = 'web'")
 				}
 			}),
 		),
@@ -205,11 +205,8 @@ describe("CiAnalyticsService.analytics", () => {
 					})
 					// The red workflow's own runs are read, skipped ones left out.
 					const runsRead = reads.find((read) => read.context === "ciWorkflowRuns")
-					assert.include(runsRead?.sql ?? "", "traces.SpanAttributes['cicd.pipeline.name'] = 'CI'")
-					assert.include(
-						runsRead?.sql ?? "",
-						"traces.ResourceAttributes['vcs.owner.name'] = 'acme'",
-					)
+					assert.include(runsRead?.sql ?? "", "ci_runs.Pipeline = 'CI'")
+					assert.include(runsRead?.sql ?? "", "ci_runs.Owner = 'acme'")
 					assert.include(runsRead?.sql ?? "", "!= 'skip'")
 					const red = analytics.pipelines[0]?.red
 					assert.strictEqual(red?.failedRuns, 2)
@@ -296,7 +293,7 @@ describe("CiAnalyticsService.listRuns", () => {
 		withService(
 			{
 				ciRuns: [
-					run(3, "failure", END - HOUR, { pullRequest: "42", attempt: "2" }),
+					run(3, "failure", END - HOUR, { pullRequest: 42, attempt: 2 }),
 					run(2, "success", END - 2 * HOUR),
 					run(1, "success", END - 3 * HOUR),
 				],
@@ -337,7 +334,7 @@ describe("CiAnalyticsService.listRuns", () => {
 					assert.notInclude(jobsRead?.sql ?? "", traceId(2))
 
 					// A failed run's jobs are read up to the end of the run, not the window.
-					assert.include(jobsRead?.sql ?? "", `traces.Timestamp <= '${wire(END - HOUR + 61_000)}'`)
+					assert.include(jobsRead?.sql ?? "", `ci_jobs.Timestamp <= '${wire(END - HOUR + 61_000)}'`)
 
 					yield* service.listRuns(tenant, {
 						startTime: END - 24 * HOUR,
@@ -346,7 +343,7 @@ describe("CiAnalyticsService.listRuns", () => {
 						cursor: page.nextCursor ?? undefined,
 					})
 					const lastRead = reads.filter((read) => read.context === "ciRuns").at(-1)
-					assert.include(lastRead?.sql ?? "", `traces.TraceId < '${traceId(2)}'`)
+					assert.include(lastRead?.sql ?? "", `ci_runs.TraceId < '${traceId(2)}'`)
 				}),
 		),
 	)

packages/backend/src/services/pr-review/CiAnalyticsService.ts

@@ -40,7 +40,7 @@ export interface CiAnalyticsServiceApi {
 
 const HOUR = 3_600
 const DAY = 86_400
-/** Raw CI traces are scanned per read; past two weeks a big org's scan outlasts the query budget. */
+/** The longest window the CI page offers: two weeks. */
 export const CI_MAX_SPAN_MS = 14 * DAY * 1000
 /** Runs read per possibly red workflow to find where its streak began. */
 const STREAK_RUNS = 200
@@ -72,14 +72,6 @@ const decodeCursor = (cursor: string) => {
 
 /** A quantile over no rows is NaN or null; the contract says null. */
 const finiteOrNull = (value: number | null) => (value !== null && Number.isFinite(value) ? value : null)
-const attemptOf = (attempt: string) => {
-	const parsed = Number(attempt)
-	return Number.isInteger(parsed) && parsed > 0 ? parsed : 1
-}
-const pullRequestOf = (value: string) => {
-	const parsed = Number(value)
-	return value !== "" && Number.isInteger(parsed) && parsed > 0 ? parsed : null
-}
 
 type CiSeriesRow = {
 	readonly bucket: DateTime.Utc
@@ -451,10 +443,11 @@ export class CiAnalyticsService extends Context.Service<CiAnalyticsService, CiAn
 								branch: run.branch,
 								revision: run.revision,
 								runId: run.runId,
-								attempt: attemptOf(run.attempt),
+								// 0 when the span carried none: an attempt counts from 1, and no run is pull request 0.
+								attempt: Math.max(run.attempt, 1),
 								trigger: run.trigger,
 								actor: run.actor,
-								pullRequest: pullRequestOf(run.pullRequest),
+								pullRequest: run.pullRequest === 0 ? null : run.pullRequest,
 								result: run.result,
 								durationMs: run.durationMs,
 								startedAt: DateTime.toEpochMillis(run.startedAt),