feat(gcp): query any Google Cloud metric live through the query engine

apps/ai/src/mcp/lib/map-warehouse-error.test.ts

@@ -1,5 +1,10 @@
 import { describe, expect, it } from "vitest"
 import { WarehouseQueryError, WarehouseResultDecodeError, WarehouseSchemaDriftError } from "@maple/domain"
+import {
+	GcpLiveQueryFailedError,
+	GcpLiveQueryRejectedError,
+	GcpLiveQueryUnavailableError,
+} from "@maple/domain/http/gcp-live-query-errors"
 import { toMcpQueryError } from "./map-warehouse-error"
 
 describe("toMcpQueryError", () => {
@@ -37,4 +42,39 @@ describe("toMcpQueryError", () => {
 		expect(mcp.message).toBe("row did not decode")
 		expect(mcp.message).not.toContain("schema apply")
 	})
+
+	it("makes a query Google refused the caller's to fix", () => {
+		const err = new GcpLiveQueryRejectedError({
+			message: "The query reaches 34 projects; filter on project_id",
+			reason: "too_many_projects",
+		})
+		const mcp = toMcpQueryError("query_data")(err)
+		expect(mcp._tag).toBe("@maple/mcp/errors/McpInvalidInputError")
+		expect(mcp.message).toBe("The query reaches 34 projects; filter on project_id")
+	})
+
+	it("reports Google Cloud metrics as unavailable when no connection reads them", () => {
+		const err = new GcpLiveQueryUnavailableError({
+			message: "No Google Cloud connection has metrics on.",
+		})
+		const mcp = toMcpQueryError("query_data")(err)
+		expect(mcp).toMatchObject({
+			_tag: "@maple/mcp/errors/McpUnavailableError",
+			message: "No Google Cloud connection has metrics on.",
+			capability: "google_cloud_metrics",
+		})
+	})
+
+	it("asks for a retry when Google did not answer", () => {
+		const err = new GcpLiveQueryFailedError({
+			message: "Google Cloud could not be read.",
+			failures: [{ scope: "projects/acme", message: "503" }],
+		})
+		const mcp = toMcpQueryError("query_data")(err)
+		expect(mcp).toMatchObject({
+			_tag: "@maple/mcp/errors/McpNotReadyError",
+			message: "Google Cloud could not be read.",
+			retryAfterSeconds: 60,
+		})
+	})
 })

apps/ai/src/mcp/lib/map-warehouse-error.ts

@@ -1,10 +1,25 @@
 import { Effect } from "effect"
-import { type WarehouseError, WarehouseQuotaExceededError, WarehouseSchemaDriftError } from "@maple/domain"
+import {
+	type WarehouseError,
+	type WarehouseQuotaExceededError,
+	WarehouseSchemaDriftError,
+} from "@maple/domain"
 import {
 	warehouseHandlers,
 	warehouseReadHandlers,
 } from "@maple/backend/services/warehouse/warehouse-error-handlers"
-import { McpQueryBudgetError, McpQueryError } from "../tools/types"
+import type {
+	GcpLiveQueryFailedError,
+	GcpLiveQueryRejectedError,
+	GcpLiveQueryUnavailableError,
+} from "@maple/domain/http/gcp-live-query-errors"
+import {
+	McpInvalidInputError,
+	McpNotReadyError,
+	McpQueryBudgetError,
+	McpQueryError,
+	McpUnavailableError,
+} from "../tools/types"
 
 export { warehouseHandlers, warehouseReadHandlers }
 
@@ -33,23 +48,55 @@ const BUDGET_GUIDANCE = {
 	max_threads: "The query needed more parallelism than it is allowed. Scan less data.",
 } satisfies Record<WarehouseQuotaExceededError["setting"], string>
 
+/** Seconds the model waits before asking Google again. */
+const GCP_RETRY_AFTER_SECONDS = 60
+
+/** The MCP error a warehouse error becomes, per member, so reads that cannot reach Google keep a narrow type. */
+type McpErrorFor<E> = E extends GcpLiveQueryRejectedError
+	? McpInvalidInputError
+	: E extends GcpLiveQueryUnavailableError
+		? McpUnavailableError
+		: E extends GcpLiveQueryFailedError
+			? McpNotReadyError
+			: McpQueryError | McpQueryBudgetError
+
+const toMcpError = (pipe: string, error: WarehouseError) => {
+	switch (error._tag) {
+		case "@maple/http/errors/GcpLiveQueryRejectedError":
+			return new McpInvalidInputError({ message: error.message })
+		case "@maple/http/errors/GcpLiveQueryUnavailableError":
+			return new McpUnavailableError({ message: error.message, capability: "google_cloud_metrics" })
+		case "@maple/http/errors/GcpLiveQueryFailedError":
+			return new McpNotReadyError({
+				message: error.message,
+				retryAfterSeconds: GCP_RETRY_AFTER_SECONDS,
+			})
+		case "@maple/http/errors/WarehouseQuotaExceededError":
+			return new McpQueryBudgetError({
+				message: BUDGET_GUIDANCE[error.setting],
+				pipeName: pipe,
+				setting: error.setting,
+			})
+		default:
+			return new McpQueryError({ message: warehouseErrorText(error), pipeName: pipe, cause: error })
+	}
+}
+
 /**
  * Curry the pipe label so call sites read as
  * `Effect.mapError(toMcpQueryError("service_overview"))`.
  *
  * A budget breach gets its own error with guidance: the raw text is the vendor's ("Upgrade your
- * plan for higher capacity"), which no model can act on.
+ * plan for higher capacity"), which no model can act on. A metrics read answered live by Google
+ * fails like the tool's other dependencies: a refused query is the caller's to fix, no connection
+ * or one Google refuses (a missing role, API or billing account) is unavailable, and Google not
+ * answering is worth a retry.
  */
 export const toMcpQueryError =
 	(pipe: string) =>
-	(error: WarehouseError): McpQueryError | McpQueryBudgetError =>
-		error instanceof WarehouseQuotaExceededError
-			? new McpQueryBudgetError({
-					message: BUDGET_GUIDANCE[error.setting],
-					pipeName: pipe,
-					setting: error.setting,
-				})
-			: new McpQueryError({ message: warehouseErrorText(error), pipeName: pipe, cause: error })
+	<E extends WarehouseError>(error: E): McpErrorFor<E> =>
+		// TypeScript cannot narrow a conditional return type through the switch.
+		toMcpError(pipe, error) as McpErrorFor<E>
 
 /**
  * `Effect.catchTags` handler map that converts every warehouse error tag into an

apps/ai/src/mcp/tools/list-metrics.ts

@@ -7,6 +7,8 @@ import { doc } from "../lib/tool-doc"
 import { Effect, Schema } from "effect"
 import { ListMetricsOutput } from "@maple/domain/mcp-outputs"
 import { MetricType } from "@maple/query-engine"
+import { isGcpMetricType } from "@maple/domain/gcp-live-metrics"
+import { QueryEngineService } from "@maple/backend/services/warehouse/QueryEngineService"
 
 const WINDOW = P.timeWindow({ defaultHours: 6, maxHours: MCP_DISCOVERY_MAX_HOURS })
 
@@ -30,7 +32,8 @@ export function registerListMetricsTool(server: McpToolRegistrar) {
 		name: "list_metrics",
 		title: "List Metrics",
 		description:
-			"Discover custom metrics with their type, unit, monotonicity and data volume. Pass a discovered `metric_name` and `metric_type` to query_data source=metrics.",
+			"Discover custom metrics with their type, unit, monotonicity and data volume. Pass a discovered `metric_name` and `metric_type` to query_data source=metrics. " +
+			"Google Cloud Monitoring metrics with data in the last day are listed too, under service `Google Cloud`, in the room the first page leaves (search e.g. `cloudsql.googleapis.com` to find them), and are read live from Google.",
 		parameters: Schema.Struct({
 			...WINDOW.fields,
 			service: P.service(),
@@ -46,6 +49,7 @@ export function registerListMetricsTool(server: McpToolRegistrar) {
 		handler: Effect.fn("McpTool.listMetrics")(function* (params) {
 			const { st, et } = yield* WINDOW.resolve(params, "list_metrics")
 			const tenant = yield* CurrentMcpTenant
+			const queryEngine = yield* QueryEngineService
 			yield* Effect.annotateCurrentSpan({
 				orgId: tenant.orgId,
 				"maple.ai.service": params.service ?? "all",
@@ -54,7 +58,7 @@ export function registerListMetricsTool(server: McpToolRegistrar) {
 				"maple.ai.offset": params.offset,
 			})
 
-			const [metricsResult, summaryResult] = yield* Effect.all(
+			const [metricsResult, summaryResult, gcpMetrics] = yield* Effect.all(
 				[
 					queryWarehouse<MetricListRow>("list_metrics", {
 						start_time: st,
@@ -70,21 +74,32 @@ export function registerListMetricsTool(server: McpToolRegistrar) {
 						end_time: et,
 						service: params.service,
 					}),
+					queryEngine.listGcpMetrics(tenant, {
+						service: params.service,
+						metricType: params.metric_type,
+						search: params.search,
+						limit: params.limit,
+						offset: params.offset,
+					}),
 				],
 				{ concurrency: "unbounded" },
 			)
 
-			const metrics = metricsResult.data
+			// Google rows fill what the first page leaves free; paging counts the warehouse's alone.
+			const hasMore = metricsResult.data.length === params.limit
+			const metrics: ReadonlyArray<MetricListRow> = [
+				...metricsResult.data,
+				...gcpMetrics.slice(0, Math.max(0, params.limit - metricsResult.data.length)),
+			]
 			yield* Effect.annotateCurrentSpan("result.rowCount", metrics.length)
 
-			const hasMore = metrics.length === params.limit
 			return {
 				timeRange: { start: st, end: et },
 				pagination: {
 					offset: params.offset,
 					limit: params.limit,
 					hasMore,
-					...(hasMore ? { nextOffset: params.offset + metrics.length } : undefined),
+					...(hasMore ? { nextOffset: params.offset + metricsResult.data.length } : undefined),
 				},
 				summary: summaryResult.data.map((s) => ({
 					metricType: s.metricType,
@@ -164,7 +179,9 @@ export function registerListMetricsTool(server: McpToolRegistrar) {
 										m.isMonotonic ? "yes" : "-",
 										m.serviceName,
 										m.metricUnit || "-",
-										formatNumber(m.dataPointCount),
+										isGcpMetricType(m.metricName)
+											? "live"
+											: formatNumber(m.dataPointCount),
 									]),
 								),
 							]),

apps/ai/src/mcp/tools/query-data.ts

@@ -98,7 +98,9 @@ type Params = typeof queryDataSchema.Type
 const queryDataDescription =
 	"Aggregate traces, logs, metrics or product events into a timeseries or a top-N breakdown. " +
 	"Start here for trends, comparisons and top-N; for errors prefer find_errors and error_detail; " +
-	"for a shape this tool cannot express, run_sql. Applied defaults are listed in the result."
+	"for a shape this tool cannot express, run_sql. Applied defaults are listed in the result. " +
+	"A Google Cloud Monitoring metric type (e.g. cloudsql.googleapis.com/database/cpu/utilization) works as metric_name and is read live from Google; " +
+	"it has no service, so filter and group by its labels with attribute_key (e.g. database_id, project_id, monitored_resource)."
 
 /** Default metric/group_by per source, as `[default, available]` for the decisions list. */
 const DEFAULT_METRIC = {

apps/ai/src/worker.ts

@@ -62,6 +62,8 @@ const configuredEnv = ({ stage, region, domains, profile }: MapleStackContext) =
 		// The source and sandbox tools resolve repositories through the GitHub App, so
 		// they need its reader credentials here too, not only on api.
 		githubAppSourceEnv,
+		// Tools that query Google Cloud metrics read them live; unset leaves them unavailable.
+		optionalSecret("MAPLE_GCP_SERVICE_ACCOUNT_KEY"),
 		// Required wherever a sandbox Worker deploys: without it every sandbox call is refused.
 		...(profile.deploys.sandbox ? [requireSecretEntry("SANDBOX_INTERNAL_SERVICE_TOKEN")] : []),
 		// Dev-only escape hatch from per-org BYO rows (see apps/api/src/resources/env.ts).

apps/api/src/resources/env.ts

@@ -82,4 +82,6 @@ export const apiConfiguredEnv = (stage: MapleStage, region: MapleRegion, domains
 		cloudflareOAuthEnv,
 		planetScaleOAuthEnv,
 		optionalPlain("MAPLE_GCP_SERVICE_ACCOUNT_EMAIL"),
+		// Sign-in for live Google Cloud metrics queries; unset leaves them unavailable.
+		optionalSecret("MAPLE_GCP_SERVICE_ACCOUNT_KEY"),
 	)

apps/api/src/routes/internal/query-engine.http.ts

@@ -1484,8 +1484,15 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleInternalApi, "query
 				.handle("listMetrics", ({ payload }) =>
 					Effect.gen(function* () {
 						const tenant = yield* CurrentTenant.Context
-						const rows = yield* runQuery(Queries.listMetrics, tenant, payload)
-						return new ListMetricsResponse({ data: rows })
+						const [rows, gcpRows] = yield* Effect.all(
+							[
+								runQuery(Queries.listMetrics, tenant, payload),
+								queryEngine.listGcpMetrics(tenant, payload),
+							],
+							{ concurrency: "unbounded" },
+						)
+						// One page of `limit`: the table reads `limit + 1` rows to tell whether more exist.
+						return new ListMetricsResponse({ data: [...rows, ...gcpRows].slice(0, payload.limit ?? 100) })
 					}),
 				)
 				.handle("metricsSummary", ({ payload }) =>

apps/api/src/routes/v2/alchemy-provider.integration.test.ts

@@ -50,6 +50,7 @@ import { HazelOAuthService } from "@maple/backend/services/auth/HazelOAuthServic
 import { OrgClickHouseSettingsService } from "@maple/backend/services/org/OrgClickHouseSettingsService"
 import { OrgMembersService } from "@maple/backend/services/org/OrgMembersService"
 import { QueryEngineService } from "@maple/backend/services/warehouse/QueryEngineService"
+import { GcpLiveMetricsService } from "@maple/backend/services/integrations/GcpLiveMetricsService"
 import { V2TransportErrorBoundaryLive } from "./error-envelope"
 import {
 	AllV2GroupLayersLive,
@@ -105,6 +106,7 @@ const makeHarness = () => {
 	const edgeCacheLive = EdgeCacheService.layer.pipe(Layer.provide(CacheBackendLive))
 	const bucketCacheLive = BucketCacheService.layer.pipe(Layer.provide(edgeCacheLive))
 	const queryEngineLive = Layer.effect(QueryEngineService, QueryEngineService.make).pipe(
+		Layer.provide(Layer.mock(GcpLiveMetricsService, {})),
 		Layer.provide(warehouseLive),
 		Layer.provide(edgeCacheLive),
 		Layer.provide(bucketCacheLive),

apps/api/src/routes/v2/alerts.http.test.ts

@@ -36,6 +36,7 @@ import { HazelOAuthService, type HazelOAuthServiceApi } from "@maple/backend/ser
 import { OrgClickHouseSettingsService } from "@maple/backend/services/org/OrgClickHouseSettingsService"
 import { OrgMembersService, type OrgMembersServiceApi } from "@maple/backend/services/org/OrgMembersService"
 import { QueryEngineService } from "@maple/backend/services/warehouse/QueryEngineService"
+import { GcpLiveMetricsService } from "@maple/backend/services/integrations/GcpLiveMetricsService"
 import { V2TransportErrorBoundaryLive } from "./error-envelope"
 import {
 	AllV2GroupLayersLive,
@@ -89,6 +90,7 @@ const makeHarness = (
 	const edgeCacheLive = EdgeCacheService.layer.pipe(Layer.provide(CacheBackendLive))
 	const bucketCacheLive = BucketCacheService.layer.pipe(Layer.provide(edgeCacheLive))
 	const queryEngineLive = Layer.effect(QueryEngineService, QueryEngineService.make).pipe(
+		Layer.provide(Layer.mock(GcpLiveMetricsService, {})),
 		Layer.provide(warehouseLive),
 		Layer.provide(edgeCacheLive),
 		Layer.provide(bucketCacheLive),

apps/api/src/routes/v2/telemetry.http.test.ts

@@ -2,7 +2,7 @@ import { afterEach, describe, expect, it } from "@effect/vitest"
 import { OrgId, UserId, WarehouseQueryError } from "@maple/domain/http"
 import { MapleApiV2 } from "@maple/domain/http/v2"
 import { QueryEngineExecuteResponse, type QueryEngineExecuteRequest } from "@maple/query-engine"
-import { ConfigProvider, Context, Effect, Layer, ManagedRuntime, Option, Schema } from "effect"
+import { ConfigProvider, Context, DateTime, Effect, Layer, ManagedRuntime, Option, Schema } from "effect"
 import { HttpRouter } from "effect/http"
 import { HttpApiBuilder } from "effect/http-api"
 import { Env } from "@maple/backend/platform/Env"
@@ -296,6 +296,7 @@ const queryEngineStub = {
 	evaluate: () => Effect.die(new Error("not used")),
 	evaluateSeries: () => Effect.die(new Error("not used")),
 	cachedDirect: (_tenant, _route, _payload, effect) => effect,
+	listGcpMetrics: () => Effect.succeed([]),
 } satisfies QueryEngineServiceApi
 
 const makeHarness = (
@@ -697,6 +698,39 @@ describe("v2 telemetry reads over HTTP", () => {
 		await harness.dispose()
 	})
 
+	it("leaves Google Cloud metrics out of /v2/metrics, so its pages stay the warehouse's", async () => {
+		let gcpReads = 0
+		const queryEngine: QueryEngineServiceApi = {
+			...queryEngineStub,
+			listGcpMetrics: () => {
+				gcpReads += 1
+				return Effect.succeed([
+					{
+						metricName: "cloudsql.googleapis.com/database/cpu/utilization",
+						metricType: "gauge",
+						serviceName: "Google Cloud",
+						metricDescription: "CPU utilization",
+						metricUnit: "10^2.%",
+						dataPointCount: 0,
+						firstSeen: DateTime.makeUnsafe("2026-07-14T11:00:00Z"),
+						lastSeen: DateTime.makeUnsafe("2026-07-15T12:00:00Z"),
+						isMonotonic: 0,
+					},
+				])
+			},
+		}
+		const harness = makeHarness(warehouseStub, queryEngine)
+		const key = await harness.bootstrapKey()
+
+		const metrics = await harness.request("GET", `/v2/metrics?${windowQuery}&search=cpu`, key.secret)
+		expect(metrics.status).toBe(200)
+		expect(metrics.body.data.map((metric: { name: string }) => metric.name)).toEqual([
+			"http.server.duration",
+		])
+		expect(gcpReads).toBe(0)
+		await harness.dispose()
+	})
+
 	// Regression: timeseries and breakdowns read the hourly rollups whenever the
 	// filters allow, and those floor their bounds through `toDateTime` or compare
 	// them against plain `DateTime` columns. Millisecond bounds made every

apps/api/src/routes/v2/v2-test-support.ts

@@ -326,6 +326,7 @@ export const TelemetryServiceStubsLayer = Layer.mergeAll(
 		evaluate: die,
 		evaluateSeries: die,
 		cachedDirect: die,
+		listGcpMetrics: die,
 	}),
 )
 

apps/api/src/routes/v2/widget-summary.http.test.ts

@@ -128,6 +128,7 @@ const queryEngineStub = (execute: QueryEngineServiceApi["execute"]): QueryEngine
 	evaluate: () => Effect.die(new Error("not used")),
 	evaluateSeries: () => Effect.die(new Error("not used")),
 	cachedDirect: (_tenant, _route, _payload, effect) => effect,
+	listGcpMetrics: () => Effect.die(new Error("not used")),
 })
 
 /** Grouped requests carry a `service` group-by; the ungrouped one does not. */