feat(gcs): Implement google gcp bucket provider

backend/app/core/batch/__init__.py

@@ -11,6 +11,7 @@
     extract_text_from_response_dict,
 )
 from .openai import OpenAIBatchProvider
+from .google_gcp import GoogleGCPBatchProvider
 from .operations import (
     download_batch_results,
     process_completed_batch,
@@ -28,6 +29,7 @@
     "GeminiClient",
     "GeminiClientError",
     "GeminiBatchProvider",
+    "GoogleGCPBatchProvider",
     "OpenAIBatchProvider",
     "create_stt_batch_requests",
     "create_tts_batch_requests",

backend/app/core/batch/google_gcp.py

@@ -0,0 +1,257 @@
+"""Google GCP Vertex AI batch provider implementation.
+
+Vertex batch prediction reads its input JSONL from GCS and writes results back to
+GCS (no File API, unlike the AI-Studio ``GeminiBatchProvider``). Input/output
+therefore ride the project's ``google-gcp`` credential (SA key + ``gcs_bucket``).
+"""
+
+import json
+import logging
+import time
+from typing import Any, cast
+from uuid import uuid4
+
+from google import genai
+from google.genai import types
+from google.cloud import storage as gcs
+
+from app.core.cloud.storage import CloudStorageError, build_gcp_sa_credentials
+from app.core.providers import (
+    GoogleGcpCredentials,
+    Provider,
+    parse_provider_credentials,
+)
+
+from .base import BATCH_KEY, BatchProvider
+from .gemini import BatchJobState
+
+logger = logging.getLogger(__name__)
+
+# Terminal Vertex job states (superset of AI-Studio: Vertex adds PAUSED).
+_TERMINAL_STATES = {
+    BatchJobState.SUCCEEDED.value,
+    BatchJobState.FAILED.value,
+    BatchJobState.CANCELLED.value,
+    BatchJobState.EXPIRED.value,
+    "JOB_STATE_PAUSED",
+}
+_FAILED_STATES = {
+    BatchJobState.FAILED.value,
+    BatchJobState.CANCELLED.value,
+    BatchJobState.EXPIRED.value,
+}
+
+_DEFAULT_INPUT_PREFIX = "batch-input"
+_DEFAULT_OUTPUT_PREFIX = "batch-output"
+
+
+def _parse_gs_uri(uri: str) -> tuple[str, str]:
+    """Split ``gs://bucket/key`` into ``(bucket, key)``."""
+    if not uri.startswith("gs://"):
+        raise ValueError(f"Expected a gs:// URI, got '{uri}'.")
+    bucket, _, key = uri[len("gs://") :].partition("/")
+    return bucket, key
+
+
+class GoogleGCPBatchProvider(BatchProvider):
+    """Vertex AI implementation of the BatchProvider interface (GCS in/out).
+
+    Each JSONL line is the Vertex request schema, e.g.
+        {"request": {"contents": [{"parts": [...], "role": "user"}]}}
+    """
+
+    DEFAULT_MODEL = "gemini-3.1-pro-preview"
+
+    def __init__(
+        self,
+        client: genai.Client,
+        storage_client: gcs.Client,
+        gcs_bucket: str,
+        model: str | None = None,
+        input_prefix: str = _DEFAULT_INPUT_PREFIX,
+        output_prefix: str = _DEFAULT_OUTPUT_PREFIX,
+    ) -> None:
+        self._client = client
+        self._storage = storage_client
+        self._bucket = gcs_bucket
+        self._model = model or self.DEFAULT_MODEL
+        self._input_prefix = input_prefix
+        self._output_prefix = output_prefix
+
+    @classmethod
+    def from_credentials(
+        cls, credentials: dict[str, Any], model: str | None = None
+    ) -> "GoogleGCPBatchProvider":
+        """Build a Vertex batch provider from a ``google-gcp`` credential dict."""
+        creds_model = cast(
+            GoogleGcpCredentials,
+            parse_provider_credentials(Provider.GOOGLE_GCP, credentials),
+        )
+
+        creds = build_gcp_sa_credentials(cast(dict[str, Any], creds_model.sa_key))
+        client = genai.Client(
+            vertexai=True,
+            project=creds_model.project_id,
+            location=creds_model.location,
+            credentials=creds,
+        )
+        storage_client = gcs.Client(project=creds_model.project_id, credentials=creds)
+        return cls(
+            client=client,
+            storage_client=storage_client,
+            gcs_bucket=creds_model.gcs_bucket,
+            model=model,
+        )
+
+    def create_batch(
+        self, jsonl_data: list[dict[str, Any]], config: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Upload input JSONL to GCS and start a Vertex batch prediction job."""
+        model = config.get("model", self._model)
+        display_name = config.get("display_name", f"batch-{int(time.time())}")
+
+        jsonl_content = "\n".join(
+            json.dumps(item, ensure_ascii=False) for item in jsonl_data
+        )
+        src_uri = self.upload_file(jsonl_content, purpose="batch")
+        dest_uri = f"gs://{self._bucket}/{self._output_prefix}/{uuid4().hex}/"
+
+        logger.info(
+            f"[create_batch] Creating Vertex batch | items={len(jsonl_data)} | "
+            f"model={model} | src={src_uri} | dest={dest_uri}"
+        )
+
+        try:
+            batch_job = self._client.batches.create(
+                model=model,
+                src=src_uri,
+                config=types.CreateBatchJobConfig(
+                    dest=dest_uri, display_name=display_name
+                ),
+            )
+            initial_state = batch_job.state.name if batch_job.state else "UNKNOWN"
+            result = {
+                "provider_batch_id": batch_job.name,
+                "provider_file_id": src_uri,
+                "provider_output_prefix": dest_uri,
+                "provider_status": initial_state,
+                "total_items": len(jsonl_data),
+            }
+            logger.info(
+                f"[create_batch] Created Vertex batch | batch_id={batch_job.name} | "
+                f"status={initial_state} | items={len(jsonl_data)}"
+            )
+            return result
+        except Exception as e:
+            logger.error(f"[create_batch] Failed to create Vertex batch | {e}")
+            raise
+
+    def get_batch_status(self, batch_id: str) -> dict[str, Any]:
+        """Poll Vertex for batch job status."""
+        logger.info(f"[get_batch_status] Polling Vertex batch | batch_id={batch_id}")
+        try:
+            batch_job = self._client.batches.get(name=batch_id)
+            state = batch_job.state.name if batch_job.state else "UNKNOWN"
+            # Results live in the job's GCS dest; get_batch_status re-fetches it.
+            output_uri = (
+                batch_job.dest.gcs_uri
+                if batch_job.dest and batch_job.dest.gcs_uri
+                else None
+            )
+            result: dict[str, Any] = {
+                "provider_status": state,
+                "provider_output_file_id": output_uri or batch_id,
+            }
+            if state in _FAILED_STATES:
+                message = batch_job.error.message if batch_job.error else state
+                result["error_message"] = message
+            logger.info(
+                f"[get_batch_status] Vertex batch status | batch_id={batch_id} | "
+                f"status={state}"
+            )
+            return result
+        except Exception as e:
+            logger.error(
+                f"[get_batch_status] Failed to poll Vertex batch | "
+                f"batch_id={batch_id} | {e}"
+            )
+            raise
+
+    def download_batch_results(self, output_file_id: str) -> list[dict[str, Any]]:
+        """Read prediction JSONL files from the batch job's GCS output prefix.
+
+        Vertex echoes the input ``key`` per line; line order is only the fallback.
+        """
+        logger.info(
+            f"[download_batch_results] Reading Vertex results | src={output_file_id}"
+        )
+        output_uri = output_file_id
+        if not output_uri.startswith("gs://"):
+            batch_job = self._client.batches.get(name=output_file_id)
+            state = batch_job.state.name if batch_job.state else "UNKNOWN"
+            if state != BatchJobState.SUCCEEDED.value:
+                raise ValueError(f"Batch job not complete. Current state: {state}")
+            if not (batch_job.dest and batch_job.dest.gcs_uri):
+                raise ValueError(f"Batch job has no GCS output | id={output_file_id}")
+            output_uri = batch_job.dest.gcs_uri
+
+        try:
+            bucket_name, prefix = _parse_gs_uri(output_uri)
+            bucket = self._storage.bucket(bucket_name)
+            results: list[dict[str, Any]] = []
+            index = 0
+            for blob in self._storage.list_blobs(bucket, prefix=prefix):
+                if not blob.name.endswith(".jsonl"):
+                    continue
+                content = blob.download_as_text()
+                for line in content.strip().split("\n"):
+                    if not line:
+                        continue
+                    parsed = json.loads(line)
+                    custom_id = parsed.get("key") or str(index)
+                    response_obj = parsed.get("response")
+                    error_obj = parsed.get("error") or parsed.get("status")
+                    results.append(
+                        {
+                            BATCH_KEY: custom_id,
+                            "response": response_obj,
+                            "error": str(error_obj) if error_obj else None,
+                        }
+                    )
+                    index += 1
+            logger.info(
+                f"[download_batch_results] Read Vertex results | src={output_uri} | "
+                f"results={len(results)}"
+            )
+            return results
+        except Exception as e:
+            logger.error(
+                f"[download_batch_results] Failed to read Vertex results | "
+                f"src={output_uri} | {e}"
+            )
+            raise
+
+    def upload_file(self, content: str, purpose: str = "batch") -> str:
+        """Upload a JSONL string to GCS and return its ``gs://`` URI."""
+        key = f"{self._input_prefix}/{int(time.time())}-{uuid4().hex}.jsonl"
+        logger.info(f"[upload_file] Uploading batch input to GCS | key={key}")
+        try:
+            blob = self._storage.bucket(self._bucket).blob(key)
+            blob.upload_from_string(content, content_type="application/jsonl")
+            return f"gs://{self._bucket}/{key}"
+        except Exception as e:
+            logger.error(f"[upload_file] Failed to upload batch input to GCS | {e}")
+            raise CloudStorageError(f"GCS upload failed: {e}") from e
+
+    def download_file(self, file_id: str) -> str:
+        """Download a ``gs://`` object's content as text."""
+        logger.info(f"[download_file] Downloading from GCS | uri={file_id}")
+        try:
+            bucket_name, key = _parse_gs_uri(file_id)
+            blob = self._storage.bucket(bucket_name).blob(key)
+            return blob.download_as_text()
+        except Exception as e:
+            logger.error(
+                f"[download_file] Failed to download from GCS | uri={file_id} | {e}"
+            )
+            raise CloudStorageError(f"GCS download failed: {e}") from e

backend/app/core/cloud/storage.py

@@ -10,6 +10,7 @@
 from urllib.parse import ParseResult, urlparse, urlunparse
 
 from abc import ABC, abstractmethod
+from typing import Any
 import boto3
 from fastapi import UploadFile
 from botocore.exceptions import ClientError
@@ -329,6 +330,14 @@ def get_cloud_storage(session: Session, project_id: int) -> CloudStorage:
 
 GCS_SCOPES = ("https://www.googleapis.com/auth/cloud-platform",)
 
+
+def build_gcp_sa_credentials(sa_key: dict[str, Any]) -> service_account.Credentials:
+    """Build signing-capable SA credentials from a service-account key dict."""
+    return service_account.Credentials.from_service_account_info(
+        sa_key, scopes=list(GCS_SCOPES)
+    )
+
+
 MAX_AUDIO_UPLOAD_BYTES = 50 * 1024 * 1024  # 50 MB
 
 _MIME_TO_EXT = {
@@ -400,9 +409,7 @@ def upload_audio_to_gcs(
     key = f"{key_prefix}/{uuid4().hex}{ext}"
 
     try:
-        creds = service_account.Credentials.from_service_account_info(
-            sa_info, scopes=list(GCS_SCOPES)
-        )
+        creds = build_gcp_sa_credentials(sa_info)
         client = gcs.Client(
             project=project_id or sa_info.get("project_id"), credentials=creds
         )

backend/app/core/config.py

@@ -118,6 +118,8 @@ def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn:
     # Used by the registry fallback when a project has no ``google`` row.
     GCP_SA_KEY: str = ""
     GCS_AUDIO_BUCKET: str = ""
+    # A batch can run for hours; sign attachment URLs for 24h so they don't expire mid-run.
+    MAX_SIGNED_URL_EXPIRY_SECONDS: int = 86400
 
     # RabbitMQ configuration for Celery broker
     RABBITMQ_HOST: str = "localhost"

backend/app/crud/assessment/batch.py

@@ -42,6 +42,7 @@
     build_anthropic_attachment_parts,
     build_gemini_attachment_parts,
     resolve_attachment_values,
+    rewrite_gcs_attachment_urls,
 )
 from app.services.llm.mappers import kaapi_params_as_dict
 from app.services.llm.providers.registry import LLMProvider
@@ -423,6 +424,16 @@ def submit_assessment_batch(
     # Determine the base provider (openai or google)
     base_provider = provider_name.replace("-native", "")
 
+    # Resolve attachments url to provider-reachable URLs before building JSONL.
+    rows = rewrite_gcs_attachment_urls(
+        session=session,
+        rows=rows,
+        attachments=attachments,
+        llm_provider=provider_name,
+        project_id=project_id,
+        organization_id=organization_id,
+    )
+
     if base_provider == LLMProvider.OPENAI:
         mapped_params, warnings = map_kaapi_to_openai_params(
             session=session,
@@ -464,7 +475,7 @@ def submit_assessment_batch(
             config=batch_config,
         )
 
-    elif base_provider == LLMProvider.GOOGLE:
+    elif base_provider in (LLMProvider.GOOGLE, LLMProvider.GOOGLE_AISTUDIO):
         mapped_params, warnings = map_kaapi_to_google_params(params)
         if warnings:
             logger.info("[submit_assessment_batch] Mapper warnings: %s", warnings)

backend/app/crud/assessment/processing.py

@@ -236,6 +236,8 @@ def parse_assessment_output(
         elif provider_name in (
             LLMProvider.GOOGLE,
             LLMProvider.GOOGLE_NATIVE,
+            LLMProvider.GOOGLE_AISTUDIO,
+            LLMProvider.GOOGLE_AISTUDIO_NATIVE,
         ):
             response = result.get("response")
             error = result.get("error")

backend/app/models/llm/constants.py

@@ -186,6 +186,7 @@ def normalize_bcp47_language(value: str) -> str:
     "anthropic": "claude-sonnet-4-6",
     "openai": "gpt-4.1-mini",
     "google": "gemini-2.5-pro",
+    "google-gcp": "gemini-3.1-pro-preview",
 }
 
 DEFAULT_ANTHROPIC_MAX_TOKENS = 4096