feat(billing): GCP cloud billing via BigQuery export

.claude/settings.local.json

@@ -206,7 +206,45 @@
       "Bash(git -C /Users/patrickwalukagga/Projects/Sunbirdai/sunbird-ai-api commit -m ' *)",
       "Bash(python -m pytest app/tests/test_billing_aggregation.py::test_summarize_active_services -v)",
       "Bash(python -m pytest app/tests/test_services/test_billing_analytics_service.py::test_cloud_category_uses_only_aws -v)",
-      "Bash(python -m pytest app/tests/test_services/test_billing_analytics_service.py app/tests/test_billing_aggregation.py app/tests/test_admin_billing.py -v)"
+      "Bash(python -m pytest app/tests/test_services/test_billing_analytics_service.py app/tests/test_billing_aggregation.py app/tests/test_admin_billing.py -v)",
+      "Bash(python -c \"import google.cloud.bigquery as b; print\\('bigquery', b.__version__\\)\")",
+      "Bash(pip show *)",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/task-brief docs/superpowers/plans/2026-07-02-gcp-cloud-billing.md 1)",
+      "Bash(pytest app/tests/test_integrations/test_billing_categories.py::test_gcp_is_in_cloud_category -v)",
+      "Bash(pytest app/tests/test_integrations/test_billing_categories.py -v)",
+      "Bash(timeout 120 bash -c 'while pgrep -f \"pip install google-cloud-bigquery\" > /dev/null; do sleep 1; done')",
+      "Bash(python -c \"import google.cloud.bigquery as b; print\\(b.__version__\\)\")",
+      "Bash(python -c \"from app.core.config import settings; print\\(settings.gcp_billing_bq_table\\)\")",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/review-package f1ee0be6 *)",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/task-brief docs/superpowers/plans/2026-07-02-gcp-cloud-billing.md 2)",
+      "Bash(pytest app/tests/test_integrations/test_gcp_billing.py -v)",
+      "Bash(black app/integrations/billing/gcp.py app/tests/test_integrations/test_gcp_billing.py)",
+      "Bash(isort app/integrations/billing/gcp.py app/tests/test_integrations/test_gcp_billing.py)",
+      "Bash(flake8 app/integrations/billing/gcp.py app/tests/test_integrations/test_gcp_billing.py)",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/review-package 760a87af HEAD)",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/task-brief docs/superpowers/plans/2026-07-02-gcp-cloud-billing.md 3)",
+      "Bash(pytest app/tests/test_billing_aggregation.py::test_active_services_counts_aws_and_gcp -v)",
+      "Bash(pytest \"app/tests/test_services/test_billing_analytics_service.py::test_cloud_category_routes_to_aws_and_gcp\" -v)",
+      "Bash(pytest app/tests/test_billing_aggregation.py app/tests/test_services/test_billing_analytics_service.py -v)",
+      "Bash(black app/services/billing_analytics/service.py app/services/billing_analytics/aggregation.py)",
+      "Bash(isort app/services/billing_analytics/service.py app/services/billing_analytics/aggregation.py)",
+      "Bash(flake8 app/services/billing_analytics/service.py app/services/billing_analytics/aggregation.py)",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/review-package 7b0a3800 HEAD)",
+      "Bash(pytest app/tests/test_services/test_billing_analytics_service.py -v)",
+      "Bash(black app/tests/test_services/test_billing_analytics_service.py --check)",
+      "Bash(isort app/tests/test_services/test_billing_analytics_service.py --check-only)",
+      "Bash(flake8 app/tests/test_services/test_billing_analytics_service.py)",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/task-brief docs/superpowers/plans/2026-07-02-gcp-cloud-billing.md 4)",
+      "Bash(git check-ignore *)",
+      "Bash(/Users/patrickwalukagga/.claude/plugins/cache/claude-plugins-official/superpowers/6.0.3/skills/subagent-driven-development/scripts/review-package d1477f4a *)",
+      "Bash(python -m pytest app/tests/test_billing_categories.py -q)",
+      "Bash(black --check app/tests/test_billing_categories.py)",
+      "Bash(isort --check-only app/tests/test_billing_categories.py)",
+      "Bash(flake8 app/tests/test_billing_categories.py)",
+      "Bash(python -m pytest app/tests/test_integrations/test_gcp_billing.py -q)",
+      "Bash(black --check app/integrations/billing/gcp.py app/core/config.py app/tests/test_integrations/test_gcp_billing.py)",
+      "Bash(isort --check-only app/integrations/billing/gcp.py app/core/config.py app/tests/test_integrations/test_gcp_billing.py)",
+      "Bash(flake8 app/integrations/billing/gcp.py app/core/config.py app/tests/test_integrations/test_gcp_billing.py)"
     ]
   }
 }

app/core/config.py

@@ -181,6 +181,30 @@ def vast_contract_types(self) -> list[str]:
         description="Cost Explorer metric to report (e.g. UnblendedCost, AmortizedCost).",
     )
 
+    # GCP (cloud) billing — BigQuery billing export
+    gcp_billing_bq_table: Optional[str] = Field(
+        default=None,
+        description=(
+            "Fully-qualified BigQuery billing-export table "
+            "(project.dataset.table), e.g. my-proj.billing.gcp_billing_export_v1_XXXX."
+        ),
+    )
+    gcp_billing_project_id: Optional[str] = Field(
+        default=None,
+        description=(
+            "GCP project to run the BigQuery job in (billed for the query). "
+            "Falls back to gcp_project_id when unset."
+        ),
+    )
+    gcp_billing_location: Optional[str] = Field(
+        default=None,
+        description=(
+            "BigQuery job location for the billing query. Required for single-region "
+            "export datasets (e.g. europe-west1); leave unset for US/EU multi-region "
+            "(the client auto-detects)."
+        ),
+    )
+
     billing_cache_ttl_seconds: int = Field(
         default=3600, description="TTL for cached normalized billing records."
     )

app/integrations/billing/categories.py

@@ -12,6 +12,7 @@
     "modal": "inference",
     "vastai": "training",
     "aws": "cloud",
+    "gcp": "cloud",
 }
 
 CATEGORIES: tuple[str, ...] = ("inference", "training", "cloud")

app/integrations/billing/gcp.py

@@ -0,0 +1,127 @@
+"""GCP billing analytics provider (Cloud Billing export in BigQuery)."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+
+from app.core.config import settings
+from app.integrations.billing.base import (
+    AnalyticsProvider,
+    ProviderQuery,
+    ProviderUnavailable,
+)
+from app.schemas.billing_analytics import BillingRecord
+
+logger = logging.getLogger(__name__)
+
+# project.dataset.table — trusted config only, validated before it reaches SQL.
+_TABLE_RE = re.compile(r"^[A-Za-z0-9_-]+\.[A-Za-z0-9_]+\.[A-Za-z0-9_]+$")
+
+
+class GcpAnalyticsProvider(AnalyticsProvider):
+    name = "gcp"
+
+    def __init__(self) -> None:
+        self.table = settings.gcp_billing_bq_table
+        self.project = settings.gcp_billing_project_id or settings.gcp_project_id
+        # BigQuery job location. Required for single-region export datasets
+        # (e.g. europe-west1); None lets the client auto-detect (US/EU multi-region).
+        self.location = settings.gcp_billing_location
+
+    async def is_available(self) -> bool:
+        return bool(self.table)
+
+    def _client(self):
+        """Construct the BigQuery client. Patched in tests."""
+        from google.cloud import bigquery  # lazy: import-time never needs BigQuery
+
+        return bigquery.Client(project=self.project)
+
+    def _granularity(self, base_resolution: str) -> str:
+        return "HOUR" if base_resolution == "hour" else "DAY"
+
+    def _build_sql(self, granularity: str) -> str:
+        # self.table is validated by the caller and backtick-quoted; @start/@end
+        # are bound parameters; granularity is a fixed internal literal.
+        return f"""
+            SELECT
+              TIMESTAMP_TRUNC(usage_start_time, {granularity}) AS bucket,
+              service.description AS service,
+              sku.description     AS sku,
+              SUM(cost + IFNULL((SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS net_cost,
+              ANY_VALUE(usage.unit) AS unit,
+              ANY_VALUE(currency)   AS currency
+            FROM `{self.table}`
+            WHERE usage_start_time >= @start AND usage_start_time < @end
+            GROUP BY bucket, service, sku
+            HAVING net_cost != 0
+            ORDER BY bucket
+        """
+
+    def _run_query(self, query: ProviderQuery) -> list:
+        """Blocking BigQuery call. Runs in a worker thread."""
+        from google.cloud import bigquery
+
+        granularity = self._granularity(query.base_resolution)
+        sql = self._build_sql(granularity)
+        job_config = bigquery.QueryJobConfig(
+            query_parameters=[
+                bigquery.ScalarQueryParameter("start", "TIMESTAMP", query.start),
+                bigquery.ScalarQueryParameter("end", "TIMESTAMP", query.end),
+            ]
+        )
+        client = self._client()
+        # location pins the job to the export dataset's region; None => auto-detect.
+        return list(
+            client.query(sql, job_config=job_config, location=self.location).result()
+        )
+
+    def _normalize(self, rows: list) -> list[BillingRecord]:
+        records: list[BillingRecord] = []
+        for row in rows:
+            service = row["service"]
+            if not service:
+                continue
+            cost = float(row["net_cost"] or 0.0)
+            if cost == 0.0:
+                continue
+            records.append(
+                BillingRecord(
+                    provider="gcp",
+                    object_id=str(service),
+                    object_name=str(service),
+                    timestamp=row["bucket"],
+                    cost=cost,
+                    runtime_ms=None,
+                    storage_gb=None,
+                    metadata={
+                        "kind": "gcp_service",
+                        "usage_type": row["sku"],
+                        "unit": row["unit"],
+                        "currency": row["currency"],
+                    },
+                )
+            )
+        return records
+
+    async def fetch_records(self, query: ProviderQuery) -> list[BillingRecord]:
+        if not self.table:
+            raise ProviderUnavailable(
+                "gcp", "GCP billing export table is not configured"
+            )
+        if not _TABLE_RE.match(self.table):
+            raise ProviderUnavailable(
+                "gcp",
+                "GCP billing export table must be a fully-qualified "
+                "project.dataset.table identifier",
+            )
+        try:
+            rows = await asyncio.to_thread(self._run_query, query)
+            return self._normalize(rows)
+        except ProviderUnavailable:
+            raise
+        except Exception as exc:  # BigQuery / auth errors — never leak details
+            logger.exception("gcp_billing_query_failed")
+            raise ProviderUnavailable("gcp", "BigQuery billing query failed") from exc

app/schemas/billing_analytics.py

@@ -11,7 +11,7 @@
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
-Provider = Literal["runpod", "modal", "vastai", "aws"]
+Provider = Literal["runpod", "modal", "vastai", "aws", "gcp"]
 
 
 class BillingRecord(BaseModel):

app/services/billing_analytics/aggregation.py

@@ -129,7 +129,7 @@ def summarize(records: list[BillingRecord], num_days: int) -> dict:
     endpoints = {r.object_id for r in runpod_endpoint_records}
     apps = {r.object_id for r in records if r.provider == "modal"}
     instances = {r.object_id for r in records if r.provider == "vastai"}
-    services = {r.object_id for r in records if r.provider == "aws"}
+    services = {r.object_id for r in records if r.provider in ("aws", "gcp")}
 
     endpoint_rows = group_records(runpod_endpoint_records, "endpoint")
     highest_endpoint = (

app/services/billing_analytics/service.py

@@ -16,6 +16,7 @@
     ProviderUnavailable,
 )
 from app.integrations.billing.categories import providers_in_category
+from app.integrations.billing.gcp import GcpAnalyticsProvider
 from app.integrations.billing.modal import ModalAnalyticsProvider
 from app.integrations.billing.runpod import RunpodAnalyticsProvider
 from app.integrations.billing.vastai import VastaiAnalyticsProvider
@@ -37,7 +38,7 @@
 
 @dataclass
 class BillingQueryParams:
-    provider: str  # "all" | "runpod" | "modal" | "vastai"
+    provider: str  # "all" | "runpod" | "modal" | "vastai" | "aws" | "gcp"
     start: datetime
     end: datetime
     resolution: str  # display: hour | day | week | month | year
@@ -67,6 +68,7 @@ def __init__(
         self.modal = modal_provider or ModalAnalyticsProvider()
         self.vastai = VastaiAnalyticsProvider()
         self.aws = AwsAnalyticsProvider()
+        self.gcp = GcpAnalyticsProvider()
         self.cache = cache or get_cache_backend()
         self.ttl = settings.billing_cache_ttl_seconds
         # Runpod endpoint IDs to scope billing to (empty = all account endpoints).
@@ -93,6 +95,7 @@ def _provider_by_name(self, name: str) -> AnalyticsProvider:
             "modal": self.modal,
             "vastai": self.vastai,
             "aws": self.aws,
+            "gcp": self.gcp,
         }[name]
 
     def _providers_for(self, category: str, provider: str) -> list[AnalyticsProvider]:

app/static/react_build/index.html

@@ -5,7 +5,7 @@
     <link rel="icon" href="/sunbird-logo.jpg" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>Sunbird AI API</title>
-    <script type="module" crossorigin src="/assets/index-Sk8NKe7P.js"></script>
+    <script type="module" crossorigin src="/assets/index-bL__p5db.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-BCYAYHKo.css">
   </head>
   <body>

app/tests/test_billing_aggregation.py

@@ -381,3 +381,24 @@ def test_summarize_active_services():
     s = summarize(recs, num_days=1)
     assert s["active_services"] == 2
     assert s["active_endpoints"] == 1
+
+
+def test_active_services_counts_aws_and_gcp():
+    from datetime import datetime
+
+    from app.schemas.billing_analytics import BillingRecord
+    from app.services.billing_analytics import aggregation
+
+    records = [
+        BillingRecord(provider="aws", object_id="Amazon EC2", object_name="Amazon EC2",
+                      timestamp=datetime(2026, 6, 1), cost=1.0,
+                      metadata={"kind": "aws_service", "usage_type": "BoxUsage"}),
+        BillingRecord(provider="gcp", object_id="Compute Engine", object_name="Compute Engine",
+                      timestamp=datetime(2026, 6, 1), cost=2.0,
+                      metadata={"kind": "gcp_service", "usage_type": "core"}),
+        BillingRecord(provider="gcp", object_id="Cloud Storage", object_name="Cloud Storage",
+                      timestamp=datetime(2026, 6, 1), cost=3.0,
+                      metadata={"kind": "gcp_service", "usage_type": "std"}),
+    ]
+    out = aggregation.summarize(records, num_days=1)
+    assert out["active_services"] == 3  # EC2 + Compute Engine + Cloud Storage

app/tests/test_billing_categories.py

@@ -27,7 +27,8 @@ def test_providers_in_category_unknown_raises():
         providers_in_category("nope")
 
 
-def test_cloud_category_maps_to_aws():
+def test_cloud_category_maps_to_aws_and_gcp():
     assert PROVIDER_CATEGORY["aws"] == "cloud"
+    assert PROVIDER_CATEGORY["gcp"] == "cloud"
     assert "cloud" in CATEGORIES
-    assert providers_in_category("cloud") == ["aws"]
+    assert providers_in_category("cloud") == ["aws", "gcp"]

app/tests/test_integrations/test_billing_categories.py

@@ -0,0 +1,11 @@
+"""Tests for billing provider category mapping."""
+
+
+def test_gcp_is_in_cloud_category():
+    from app.integrations.billing.categories import (
+        PROVIDER_CATEGORY,
+        providers_in_category,
+    )
+
+    assert PROVIDER_CATEGORY["gcp"] == "cloud"
+    assert providers_in_category("cloud") == ["aws", "gcp"]