package clients import ( "bytes" "context" "encoding/base64" "encoding/json" "fmt" "io" "net/http" "net/url" "strings" "time" ) const geminiVeoMaxVideoBytes int64 = 256 << 20 type geminiVeoImage struct { URI string MimeType string } func geminiVeoRequest(request Request) bool { return request.Kind == "videos.generations" && strings.Contains(strings.ToLower(strings.TrimSpace(upstreamModelName(request.Candidate))), "veo") } func (c GeminiClient) runVeo(ctx context.Context, request Request, apiKey string) (Response, error) { startedAt := time.Now() operationName := strings.TrimSpace(request.RemoteTaskID) lastRequestID := operationName var operation map[string]any var wire *WireResponse if operationName == "" { body, err := geminiVeoBody(request) if err != nil { return Response{}, err } operation, lastRequestID, wire, err = c.geminiVeoPost(ctx, request, apiKey, body) if err != nil { return Response{}, annotateResponseError(err, lastRequestID, startedAt, time.Now()) } operationName = strings.TrimSpace(stringFromAny(operation["name"])) if err := validateGeminiVeoOperationName(operationName); err != nil { return Response{}, err } if request.OnRemoteTaskSubmitted != nil { if err := request.OnRemoteTaskSubmitted(operationName, geminiVeoOperationCheckpoint(operation)); err != nil { return Response{}, err } } } else if err := validateGeminiVeoOperationName(operationName); err != nil { return Response{}, err } interval := durationFromConfig(request.Candidate.PlatformConfig, 10*time.Second, "pollIntervalMs", "poll_interval_ms") timeout := durationFromConfig(request.Candidate.PlatformConfig, 10*time.Minute, "pollTimeoutMs", "poll_timeout_ms", "timeoutMs") deadline := time.NewTimer(timeout) defer deadline.Stop() nextPoll := time.NewTimer(0) defer nextPoll.Stop() transientFailures := 0 for { if operation != nil && boolFromAny(operation["done"]) { return c.geminiVeoCompletedResponse(ctx, request, apiKey, operationName, operation, wire, lastRequestID, startedAt) } select { case <-ctx.Done(): return Response{}, &ClientError{Code: "cancelled", Message: ctx.Err().Error(), RequestID: lastRequestID, Retryable: true} case <-deadline.C: return Response{}, &ClientError{ Code: "timeout", Message: fmt.Sprintf("gemini Veo operation %s did not finish before timeout", operationName), RequestID: lastRequestID, Retryable: true, } case <-nextPoll.C: pollStartedAt := time.Now() result, requestID, pollWire, err := c.geminiVeoGetOperation(ctx, request, apiKey, operationName) if requestID != "" { lastRequestID = requestID } if err != nil { err = annotateResponseError(err, lastRequestID, pollStartedAt, time.Now()) if !IsRetryable(err) { return Response{}, err } transientFailures++ resetGeminiVeoPollTimer(nextPoll, geminiVeoRetryInterval(interval, transientFailures)) continue } transientFailures = 0 operation = result wire = pollWire if request.OnRemoteTaskPolled != nil { if err := request.OnRemoteTaskPolled(operationName, geminiVeoOperationCheckpoint(operation)); err != nil { return Response{}, err } } if boolFromAny(operation["done"]) { return c.geminiVeoCompletedResponse(ctx, request, apiKey, operationName, operation, wire, lastRequestID, startedAt) } resetGeminiVeoPollTimer(nextPoll, interval) } } } func geminiVeoBody(request Request) (map[string]any, error) { prompt := firstNonEmptyPrompt(request.Body, "") if prompt == "" { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo prompt is required", Param: "prompt", StatusCode: http.StatusBadRequest, Retryable: false} } firstFrame, lastFrame, references := geminiVeoImageInputs(request.Body) instance := map[string]any{"prompt": prompt} if firstFrame.URI != "" { image, err := geminiVeoInlineImage(firstFrame) if err != nil { return nil, err } instance["image"] = image } if lastFrame.URI != "" { image, err := geminiVeoInlineImage(lastFrame) if err != nil { return nil, err } instance["lastFrame"] = image } if len(references) > 3 { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo supports at most 3 reference images", Param: "reference_images", StatusCode: http.StatusBadRequest, Retryable: false} } if len(references) > 0 { items := make([]any, 0, len(references)) for _, reference := range references { image, err := geminiVeoInlineImage(reference) if err != nil { return nil, err } items = append(items, map[string]any{"image": image, "referenceType": "asset"}) } instance["referenceImages"] = items } parameters, err := geminiVeoParameters(request.Body, lastFrame.URI != "", len(references) > 0) if err != nil { return nil, err } body := map[string]any{"instances": []any{instance}} if len(parameters) > 0 { body["parameters"] = parameters } return body, nil } func geminiVeoParameters(body map[string]any, hasLastFrame bool, hasReferences bool) (map[string]any, error) { parameters := map[string]any{} count := intFromAny(firstPresent(body["n"], body["numberOfVideos"], body["number_of_videos"])) if count == 0 { count = 1 } if count != 1 { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo supports exactly 1 output video", Param: "n", StatusCode: http.StatusBadRequest, Retryable: false} } parameters["sampleCount"] = count duration := intFromAny(firstPresent(body["duration"], body["duration_seconds"], body["durationSeconds"])) if duration != 0 { if duration != 4 && duration != 6 && duration != 8 { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo duration must be 4, 6, or 8 seconds", Param: "duration", StatusCode: http.StatusBadRequest, Retryable: false} } parameters["durationSeconds"] = duration } aspectRatio := strings.TrimSpace(firstNonEmptyString(body["aspect_ratio"], body["aspectRatio"], body["ratio"])) if aspectRatio != "" { if aspectRatio != "16:9" && aspectRatio != "9:16" { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo aspect ratio must be 16:9 or 9:16", Param: "aspect_ratio", StatusCode: http.StatusBadRequest, Retryable: false} } parameters["aspectRatio"] = aspectRatio } resolution := geminiVeoResolution(firstNonEmptyString(body["resolution"], body["size"])) if resolution == "invalid" { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo resolution must be 720p, 1080p, or 4k", Param: "resolution", StatusCode: http.StatusBadRequest, Retryable: false} } if resolution != "" { parameters["resolution"] = resolution } if (resolution == "1080p" || resolution == "4k" || hasLastFrame || hasReferences) && duration != 0 && duration != 8 { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo requires an 8-second duration for 1080p, 4k, last-frame, or reference-image generation", Param: "duration", StatusCode: http.StatusBadRequest, Retryable: false} } for _, item := range []struct { to string from []string }{ {to: "personGeneration", from: []string{"personGeneration", "person_generation"}}, {to: "negativePrompt", from: []string{"negativePrompt", "negative_prompt"}}, } { if value := strings.TrimSpace(firstNonEmptyStringValue(body, item.from...)); value != "" { parameters[item.to] = value } } for _, key := range []string{"enhancePrompt", "enhance_prompt"} { if value, ok := body[key].(bool); ok { parameters["enhancePrompt"] = value break } } return parameters, nil } func geminiVeoResolution(value string) string { normalized := strings.ToLower(strings.TrimSpace(value)) switch normalized { case "": return "" case "720p", "1080p", "4k": return normalized case "2160p", "3840x2160", "2160x3840": return "4k" default: return "invalid" } } func geminiVeoImageInputs(body map[string]any) (geminiVeoImage, geminiVeoImage, []geminiVeoImage) { first := firstGeminiVeoImage(body["first_frame"], body["firstFrame"], body["first_frame_image"], body["firstFrameImage"]) last := firstGeminiVeoImage(body["last_frame"], body["lastFrame"], body["last_frame_image"], body["lastFrameImage"]) references := geminiVeoImagesFromValues(body["reference_images"], body["referenceImages"], body["reference_image"], body["referenceImage"]) images := geminiVeoImagesFromValues(body["image"], body["images"], body["image_url"], body["imageUrl"], body["image_urls"], body["imageUrls"]) if first.URI == "" && len(images) > 0 { first = images[0] images = images[1:] } references = append(references, images...) for _, item := range contentItems(body["content"]) { if !strings.Contains(strings.ToLower(strings.TrimSpace(stringFromAny(item["type"]))), "image") { continue } image := firstGeminiVeoImage(item["image_url"], item["imageUrl"], item["url"], item["image"]) if image.URI == "" { continue } if image.MimeType == "" { image.MimeType = firstNonEmptyString(item["mime_type"], item["mimeType"]) } switch strings.ToLower(strings.TrimSpace(stringFromAny(item["role"]))) { case "first_frame", "firstframe": if first.URI == "" { first = image } case "last_frame", "lastframe": if last.URI == "" { last = image } default: if first.URI == "" { first = image } else { references = append(references, image) } } } return first, last, deduplicateGeminiVeoImages(references, first.URI, last.URI) } func firstGeminiVeoImage(values ...any) geminiVeoImage { for _, value := range values { if images := geminiVeoImagesFromAny(value); len(images) > 0 { return images[0] } } return geminiVeoImage{} } func geminiVeoImagesFromValues(values ...any) []geminiVeoImage { out := make([]geminiVeoImage, 0) for _, value := range values { out = append(out, geminiVeoImagesFromAny(value)...) } return out } func geminiVeoImagesFromAny(value any) []geminiVeoImage { switch typed := value.(type) { case string: if uri := strings.TrimSpace(typed); uri != "" { return []geminiVeoImage{{URI: uri}} } case []any: out := make([]geminiVeoImage, 0, len(typed)) for _, item := range typed { out = append(out, geminiVeoImagesFromAny(item)...) } return out case []string: out := make([]geminiVeoImage, 0, len(typed)) for _, item := range typed { out = append(out, geminiVeoImagesFromAny(item)...) } return out case map[string]any: if nested := firstPresent(typed["image_url"], typed["imageUrl"], typed["image"]); nested != nil { if images := geminiVeoImagesFromAny(nested); len(images) > 0 { if images[0].MimeType == "" { images[0].MimeType = firstNonEmptyString(typed["mime_type"], typed["mimeType"]) } return images } } uri := firstNonEmptyString(typed["url"], typed["uri"], typed["data"], typed["bytesBase64Encoded"], typed["bytes_base64_encoded"]) if uri != "" { return []geminiVeoImage{{URI: uri, MimeType: firstNonEmptyString(typed["mime_type"], typed["mimeType"])}} } } return nil } func deduplicateGeminiVeoImages(images []geminiVeoImage, exclusions ...string) []geminiVeoImage { seen := map[string]bool{} for _, exclusion := range exclusions { if exclusion = strings.TrimSpace(exclusion); exclusion != "" { seen[exclusion] = true } } out := make([]geminiVeoImage, 0, len(images)) for _, image := range images { image.URI = strings.TrimSpace(image.URI) if image.URI == "" || seen[image.URI] { continue } seen[image.URI] = true out = append(out, image) } return out } func geminiVeoInlineImage(image geminiVeoImage) (map[string]any, error) { parsed := geminiDataURL(image.URI) mimeType := strings.TrimSpace(image.MimeType) data := "" if parsed != nil { data = parsed.data mimeType = firstNonEmptyString(mimeType, parsed.mimeType) } else if !requestLikeURL(image.URI) { data = strings.TrimSpace(image.URI) } if data == "" { return nil, &ClientError{ Code: "invalid_parameter", Message: "gemini Veo image input must be hydrated as base64 data", Param: "image", StatusCode: http.StatusBadRequest, Retryable: false, } } if _, err := base64.StdEncoding.DecodeString(data); err != nil { return nil, &ClientError{Code: "invalid_parameter", Message: "gemini Veo image input contains invalid base64 data", Param: "image", StatusCode: http.StatusBadRequest, Retryable: false} } if mimeType == "" { mimeType = "image/png" } return map[string]any{"bytesBase64Encoded": data, "mimeType": geminiMediaMime(mimeType, "image")}, nil } func requestLikeURL(value string) bool { normalized := strings.ToLower(strings.TrimSpace(value)) return strings.HasPrefix(normalized, "http://") || strings.HasPrefix(normalized, "https://") || strings.HasPrefix(normalized, "/static/") } func (c GeminiClient) geminiVeoPost(ctx context.Context, request Request, apiKey string, body map[string]any) (map[string]any, string, *WireResponse, error) { raw, _ := json.Marshal(body) endpoint := geminiVeoActionURL(request.Candidate.BaseURL, upstreamModelName(request.Candidate)) req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(raw)) if err != nil { return nil, "", nil, err } req.Header.Set("Content-Type", "application/json") req.Header.Set("x-goog-api-key", apiKey) applyUpstreamIdempotency(req, request) if err := notifySubmissionStarted(request); err != nil { return nil, "", nil, err } resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { return nil, "", nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} } if err := notifyResponseReceived(request); err != nil { resp.Body.Close() return nil, "", nil, err } requestID := requestIDFromHTTPResponse(resp) result, wire, err := decodeHTTPResponseForProtocol(resp, ProtocolGeminiVeo) if notifyErr := notifyWireResponse(request, wire); notifyErr != nil { return result, requestID, wire, notifyErr } return result, requestID, wire, err } func (c GeminiClient) geminiVeoGetOperation(ctx context.Context, request Request, apiKey string, operationName string) (map[string]any, string, *WireResponse, error) { endpoint, err := geminiVeoOperationURL(request.Candidate.BaseURL, operationName) if err != nil { return nil, "", nil, err } req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) if err != nil { return nil, "", nil, err } req.Header.Set("x-goog-api-key", apiKey) resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { return nil, "", nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} } requestID := requestIDFromHTTPResponse(resp) result, wire, err := decodeHTTPResponseForProtocol(resp, ProtocolGeminiVeo) return result, requestID, wire, err } func (c GeminiClient) geminiVeoCompletedResponse(ctx context.Context, request Request, apiKey string, operationName string, operation map[string]any, wire *WireResponse, requestID string, startedAt time.Time) (Response, error) { finishedAt := time.Now() if operationError := mapFromAny(operation["error"]); len(operationError) > 0 { code := firstNonEmptyString(operationError["status"], operationError["code"]) if code == "" { code = "gemini_veo_failed" } message := strings.TrimSpace(stringFromAny(operationError["message"])) if message == "" { message = "gemini Veo operation failed" } return Response{}, &ClientError{ Code: strings.ToLower(code), Message: message, RequestID: requestID, ResponseStartedAt: startedAt, ResponseFinishedAt: finishedAt, ResponseDurationMS: responseDurationMS(startedAt, finishedAt), Retryable: false, } } items, err := c.geminiVeoResultItems(ctx, request, apiKey, operation) finishedAt = time.Now() if err != nil { return Response{}, annotateResponseError(err, requestID, startedAt, finishedAt) } result := map[string]any{ "id": operationName, "object": "video.generation", "created": nowUnix(), "model": request.Model, "upstream_task_id": operationName, "data": items, } return Response{ Result: result, RequestID: firstNonEmpty(requestID, operationName), Progress: providerProgress(request), ResponseStartedAt: startedAt, ResponseFinishedAt: finishedAt, ResponseDurationMS: responseDurationMS(startedAt, finishedAt), UpstreamProtocol: ProtocolGeminiVeo, Wire: wire, }, nil } func (c GeminiClient) geminiVeoResultItems(ctx context.Context, request Request, apiKey string, operation map[string]any) ([]any, error) { response := mapFromAny(operation["response"]) generateResponse := mapFromAny(firstPresent(response["generateVideoResponse"], response["generate_video_response"])) samples := mapListFromAny(firstPresent(generateResponse["generatedSamples"], generateResponse["generated_samples"], response["generatedVideos"], response["generated_videos"])) if len(samples) == 0 { return nil, &ClientError{Code: "invalid_response", Message: "gemini Veo operation returned no generated video", Retryable: false} } items := make([]any, 0, len(samples)) for _, sample := range samples { video := mapFromAny(sample["video"]) if len(video) == 0 { video = sample } mimeType := firstNonEmptyString(video["mimeType"], video["mime_type"], sample["mimeType"], sample["mime_type"]) if mimeType == "" { mimeType = "video/mp4" } var payload []byte if encoded := firstNonEmptyString(video["videoBytes"], video["video_bytes"], video["bytesBase64Encoded"], video["bytes_base64_encoded"]); encoded != "" { decoded, err := base64.StdEncoding.DecodeString(encoded) if err != nil { return nil, &ClientError{Code: "invalid_response", Message: "gemini Veo returned invalid video base64 data", Retryable: false} } payload = decoded } else if uri := strings.TrimSpace(firstNonEmptyString(video["uri"], video["url"])); uri != "" { var err error payload, mimeType, err = c.geminiVeoDownload(ctx, request, apiKey, uri, mimeType) if err != nil { return nil, err } } if len(payload) == 0 { return nil, &ClientError{Code: "invalid_response", Message: "gemini Veo generated video payload is missing", Retryable: false} } items = append(items, map[string]any{"type": "video", "video_bytes": payload, "mime_type": mimeType}) } return items, nil } func (c GeminiClient) geminiVeoDownload(ctx context.Context, request Request, apiKey string, rawURI string, fallbackMimeType string) ([]byte, string, error) { downloadURL, trustedHost, err := geminiVeoDownloadURL(request.Candidate.BaseURL, rawURI) if err != nil { return nil, "", err } req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil) if err != nil { return nil, "", err } if trustedHost { req.Header.Set("x-goog-api-key", apiKey) } client := httpClient(request.HTTPClient, c.HTTPClient) redirectClient := *client originalRedirect := client.CheckRedirect redirectClient.CheckRedirect = func(next *http.Request, via []*http.Request) error { if originalRedirect != nil { if err := originalRedirect(next, via); err != nil { return err } } if len(via) > 0 && !strings.EqualFold(next.URL.Hostname(), via[0].URL.Hostname()) { next.Header.Del("x-goog-api-key") } return nil } resp, err := redirectClient.Do(req) if err != nil { return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) return nil, "", &ClientError{Code: statusCodeName(resp.StatusCode), Message: errorMessage(raw, resp.Status), StatusCode: resp.StatusCode, RequestID: requestIDFromHTTPResponse(resp), Retryable: HTTPRetryable(resp.StatusCode)} } payload, err := io.ReadAll(io.LimitReader(resp.Body, geminiVeoMaxVideoBytes+1)) if err != nil { return nil, "", &ClientError{Code: "response_read_error", Message: err.Error(), StatusCode: resp.StatusCode, Retryable: true} } if int64(len(payload)) > geminiVeoMaxVideoBytes { return nil, "", &ClientError{Code: "response_too_large", Message: fmt.Sprintf("gemini Veo video exceeds %d bytes", geminiVeoMaxVideoBytes), StatusCode: resp.StatusCode, Retryable: false} } mimeType := strings.TrimSpace(strings.Split(resp.Header.Get("Content-Type"), ";")[0]) if mimeType == "" || mimeType == "application/octet-stream" { mimeType = fallbackMimeType } if mimeType == "" { mimeType = "video/mp4" } return payload, mimeType, nil } func geminiVeoVersionedBaseURL(baseURL string) string { base := strings.TrimRight(strings.TrimSpace(baseURL), "/") if base == "" { base = "https://generativelanguage.googleapis.com" } base = strings.TrimSuffix(base, "/openai") if !strings.HasSuffix(base, "/v1") && !strings.HasSuffix(base, "/v1beta") && !strings.HasSuffix(base, "/v1alpha") { base += "/v1beta" } return base } func geminiVeoActionURL(baseURL string, model string) string { return fmt.Sprintf("%s/models/%s:predictLongRunning", geminiVeoVersionedBaseURL(baseURL), url.PathEscape(strings.TrimSpace(model))) } func geminiVeoOperationURL(baseURL string, operationName string) (string, error) { if err := validateGeminiVeoOperationName(operationName); err != nil { return "", err } segments := strings.Split(strings.Trim(operationName, "/"), "/") for index, segment := range segments { segments[index] = url.PathEscape(segment) } return geminiVeoVersionedBaseURL(baseURL) + "/" + strings.Join(segments, "/"), nil } func validateGeminiVeoOperationName(operationName string) error { name := strings.TrimSpace(operationName) if name == "" || strings.Contains(name, "..") || strings.ContainsAny(name, "?#\\") || strings.HasPrefix(name, "/") || strings.Contains(name, "://") { return &ClientError{Code: "invalid_response", Message: "gemini Veo operation name is invalid", Retryable: false} } if !strings.Contains(name, "/operations/") && !strings.HasPrefix(name, "operations/") { return &ClientError{Code: "invalid_response", Message: "gemini Veo operation name is invalid", Retryable: false} } return nil } func geminiVeoDownloadURL(baseURL string, rawURI string) (string, bool, error) { base, err := url.Parse(geminiVeoVersionedBaseURL(baseURL)) if err != nil { return "", false, &ClientError{Code: "invalid_response", Message: "gemini Veo base URL is invalid", Retryable: false} } parsed, err := url.Parse(strings.TrimSpace(rawURI)) if err != nil { return "", false, &ClientError{Code: "invalid_response", Message: "gemini Veo video URI is invalid", Retryable: false} } if !parsed.IsAbs() { parsed = base.ResolveReference(parsed) } if parsed.Scheme != "https" && parsed.Scheme != "http" { return "", false, &ClientError{Code: "invalid_response", Message: "gemini Veo video URI scheme is unsupported", Retryable: false} } hostname := strings.ToLower(strings.TrimSpace(parsed.Hostname())) trusted := strings.EqualFold(hostname, base.Hostname()) || hostname == "generativelanguage.googleapis.com" return parsed.String(), trusted, nil } func geminiVeoOperationCheckpoint(operation map[string]any) map[string]any { checkpoint := map[string]any{"done": boolFromAny(operation["done"])} if metadata := mapFromAny(operation["metadata"]); len(metadata) > 0 { checkpoint["metadata"] = metadata } if operationError := mapFromAny(operation["error"]); len(operationError) > 0 { checkpoint["error"] = operationError } return checkpoint } func resetGeminiVeoPollTimer(timer *time.Timer, duration time.Duration) { if !timer.Stop() { select { case <-timer.C: default: } } timer.Reset(duration) } func geminiVeoRetryInterval(interval time.Duration, failures int) time.Duration { if failures <= 0 { return interval } multiplier := 1 << min(failures-1, 4) result := interval * time.Duration(multiplier) if result > 30*time.Second { return 30 * time.Second } return result }