From 44d5cf2b9d3c5d4347a63732937209fe83bd182e Mon Sep 17 00:00:00 2001 From: wangbo Date: Tue, 4 Aug 2026 16:26:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(gemini):=20=E6=8E=A5=E5=85=A5=20Veo=20?= =?UTF-8?q?=E5=AE=98=E6=96=B9=E5=BC=82=E6=AD=A5=E7=94=9F=E6=88=90=E5=8D=8F?= =?UTF-8?q?=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将 Gemini Veo 视频请求改为 predictLongRunning 提交与 Operation 轮询，支持任务恢复、首尾帧和参考图的官方 Base64 字段映射。 完成后由网关携带 API Key 从受信任的 Gemini 文件域下载视频，再进入现有对象存储转存链路，避免暴露鉴权地址。 验证：env -u AI_GATEWAY_TEST_DATABASE_URL go test ./... -count=1；go vet ./...；真实 Veo 3.1 任务生成并下载 1633544 字节视频。 --- apps/api/internal/clients/gemini.go | 3 + apps/api/internal/clients/gemini_veo.go | 650 ++++++++++++++++++ apps/api/internal/clients/gemini_veo_test.go | 267 +++++++ apps/api/internal/clients/responses_compat.go | 1 + apps/api/internal/runner/request_assets.go | 10 +- .../internal/runner/request_assets_test.go | 18 + apps/api/internal/runner/service.go | 3 + 7 files changed, 951 insertions(+), 1 deletion(-) create mode 100644 apps/api/internal/clients/gemini_veo.go create mode 100644 apps/api/internal/clients/gemini_veo_test.go diff --git a/apps/api/internal/clients/gemini.go b/apps/api/internal/clients/gemini.go index a342fb1..ce2505b 100644 --- a/apps/api/internal/clients/gemini.go +++ b/apps/api/internal/clients/gemini.go @@ -23,6 +23,9 @@ func (c GeminiClient) Run(ctx context.Context, request Request) (Response, error if apiKey == "" { return Response{}, &ClientError{Code: "missing_credentials", Message: "gemini api key is required", Retryable: false} } + if geminiVeoRequest(request) { + return c.runVeo(ctx, request, apiKey) + } body := geminiBody(request) raw, _ := json.Marshal(body) endpoint := geminiURL(request.Candidate.BaseURL, upstreamModelName(request.Candidate), apiKey) diff --git a/apps/api/internal/clients/gemini_veo.go b/apps/api/internal/clients/gemini_veo.go new file mode 100644 index 0000000..74d4b23 --- /dev/null +++ b/apps/api/internal/clients/gemini_veo.go @@ -0,0 +1,650 @@ +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 +} diff --git a/apps/api/internal/clients/gemini_veo_test.go b/apps/api/internal/clients/gemini_veo_test.go new file mode 100644 index 0000000..9387d14 --- /dev/null +++ b/apps/api/internal/clients/gemini_veo_test.go @@ -0,0 +1,267 @@ +package clients + +import ( + "context" + "encoding/base64" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestGeminiVeoRunUsesOfficialLongRunningProtocol(t *testing.T) { + const operationName = "models/veo-3.1-generate-preview/operations/test-operation" + videoPayload := []byte("test mp4 payload") + var server *httptest.Server + var pollCount atomic.Int32 + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("x-goog-api-key") != "test-gemini-key" { + t.Errorf("missing Gemini API key header for %s", r.URL.Path) + } + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1beta/models/veo-3.1-generate-preview:predictLongRunning": + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatalf("decode Veo body: %v", err) + } + instances := body["instances"].([]any) + if prompt := instances[0].(map[string]any)["prompt"]; prompt != "a paper boat on a lake" { + t.Errorf("unexpected prompt: %v", prompt) + } + parameters := body["parameters"].(map[string]any) + if parameters["durationSeconds"] != float64(4) || parameters["resolution"] != "720p" || parameters["aspectRatio"] != "16:9" { + t.Errorf("unexpected parameters: %+v", parameters) + } + _ = json.NewEncoder(w).Encode(map[string]any{"name": operationName}) + case r.Method == http.MethodGet && r.URL.Path == "/v1beta/"+operationName: + if pollCount.Add(1) == 1 { + _ = json.NewEncoder(w).Encode(map[string]any{"name": operationName, "done": false}) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "name": operationName, + "done": true, + "response": map[string]any{"generateVideoResponse": map[string]any{"generatedSamples": []any{ + map[string]any{"video": map[string]any{"uri": server.URL + "/v1beta/files/generated-video:download", "mimeType": "video/mp4"}}, + }}}, + }) + case r.Method == http.MethodGet && r.URL.Path == "/v1beta/files/generated-video:download": + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write(videoPayload) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + var submitted string + polled := 0 + response, err := (GeminiClient{}).Run(context.Background(), Request{ + Kind: "videos.generations", + ModelType: "video_generate", + Model: "veo-3.1", + Body: map[string]any{ + "prompt": "a paper boat on a lake", + "duration": 4, + "resolution": "720p", + "aspect_ratio": "16:9", + }, + Candidate: testGeminiVeoCandidate(server.URL), + OnRemoteTaskSubmitted: func(remoteTaskID string, _ map[string]any) error { + submitted = remoteTaskID + return nil + }, + OnRemoteTaskPolled: func(remoteTaskID string, _ map[string]any) error { + if remoteTaskID != operationName { + t.Errorf("unexpected polled operation: %s", remoteTaskID) + } + polled++ + return nil + }, + }) + if err != nil { + t.Fatalf("run Gemini Veo: %v", err) + } + if submitted != operationName || polled != 2 || pollCount.Load() != 2 { + t.Fatalf("unexpected async callbacks: submitted=%q polled=%d requests=%d", submitted, polled, pollCount.Load()) + } + if response.UpstreamProtocol != ProtocolGeminiVeo || response.RequestID != operationName { + t.Fatalf("unexpected response metadata: %+v", response) + } + data := response.Result["data"].([]any) + item := data[0].(map[string]any) + if got := item["video_bytes"].([]byte); !reflect.DeepEqual(got, videoPayload) { + t.Fatalf("unexpected video payload: %q", got) + } + if _, exposed := item["uri"]; exposed { + t.Fatalf("upstream authenticated URI must not be exposed: %+v", item) + } +} + +func TestGeminiVeoRunResumesOperationWithoutSubmittingAgain(t *testing.T) { + const operationName = "models/veo-3.1-generate-preview/operations/resumed" + var postCount atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + postCount.Add(1) + http.Error(w, "unexpected submit", http.StatusInternalServerError) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "name": operationName, + "done": true, + "response": map[string]any{"generateVideoResponse": map[string]any{"generatedSamples": []any{ + map[string]any{"video": map[string]any{"videoBytes": base64.StdEncoding.EncodeToString([]byte("resumed video")), "mimeType": "video/mp4"}}, + }}}, + }) + })) + defer server.Close() + + response, err := (GeminiClient{}).Run(context.Background(), Request{ + Kind: "videos.generations", + ModelType: "video_generate", + Model: "veo-3.1", + Body: map[string]any{"prompt": "ignored on resume"}, + Candidate: testGeminiVeoCandidate(server.URL), + RemoteTaskID: operationName, + }) + if err != nil { + t.Fatalf("resume Gemini Veo: %v", err) + } + if postCount.Load() != 0 || response.Result["id"] != operationName { + t.Fatalf("resume submitted a new operation or returned wrong ID: posts=%d result=%+v", postCount.Load(), response.Result) + } +} + +func TestGeminiVeoBodyMapsImageAndParameters(t *testing.T) { + first := "data:image/png;base64," + base64.StdEncoding.EncodeToString([]byte("first")) + last := "data:image/jpeg;base64," + base64.StdEncoding.EncodeToString([]byte("last")) + reference := "data:image/webp;base64," + base64.StdEncoding.EncodeToString([]byte("reference")) + body, err := geminiVeoBody(Request{Body: map[string]any{ + "prompt": "animate", + "first_frame": first, + "last_frame": last, + "reference_images": []any{reference}, + "duration": 8, + "resolution": "2160p", + "aspect_ratio": "9:16", + "negative_prompt": "text", + "person_generation": "allow_adult", + "enhance_prompt": false, + }}) + if err != nil { + t.Fatalf("build Gemini Veo body: %v", err) + } + instance := body["instances"].([]any)[0].(map[string]any) + image := instance["image"].(map[string]any) + lastFrame := instance["lastFrame"].(map[string]any) + references := instance["referenceImages"].([]any) + if image["bytesBase64Encoded"] != base64.StdEncoding.EncodeToString([]byte("first")) || image["mimeType"] != "image/png" { + t.Fatalf("unexpected first frame: %+v", image) + } + if lastFrame["mimeType"] != "image/jpeg" || len(references) != 1 { + t.Fatalf("unexpected last/reference images: last=%+v references=%+v", lastFrame, references) + } + parameters := body["parameters"].(map[string]any) + if parameters["resolution"] != "4k" || parameters["durationSeconds"] != 8 || parameters["enhancePrompt"] != false { + t.Fatalf("unexpected Gemini Veo parameters: %+v", parameters) + } +} + +func TestGeminiVeoRejectsInvalidDurationBeforeSubmission(t *testing.T) { + _, err := geminiVeoBody(Request{Body: map[string]any{"prompt": "test", "duration": 5}}) + if err == nil || ErrorCode(err) != "invalid_parameter" || ErrorParam(err) != "duration" || !strings.Contains(err.Error(), "4, 6, or 8") { + t.Fatalf("unexpected invalid duration error: %v", err) + } +} + +func TestGeminiVeoDownloadURLOnlyTrustsProviderAndGoogleAPIs(t *testing.T) { + for _, test := range []struct { + uri string + trusted bool + }{ + {uri: "https://gemini-proxy.example.com/v1beta/files/video", trusted: true}, + {uri: "https://generativelanguage.googleapis.com/v1beta/files/video", trusted: true}, + {uri: "https://storage.googleapis.com/signed/video", trusted: false}, + {uri: "https://attacker.example.com/video", trusted: false}, + } { + _, trusted, err := geminiVeoDownloadURL("https://gemini-proxy.example.com/v1beta", test.uri) + if err != nil { + t.Fatalf("parse %s: %v", test.uri, err) + } + if trusted != test.trusted { + t.Fatalf("unexpected trust for %s: got %t want %t", test.uri, trusted, test.trusted) + } + } +} + +func TestGeminiVeoLive(t *testing.T) { + apiKey := strings.TrimSpace(os.Getenv("GEMINI_VEO_LIVE_API_KEY")) + if apiKey == "" { + t.Skip("GEMINI_VEO_LIVE_API_KEY is not configured") + } + baseURL := strings.TrimSpace(os.Getenv("GEMINI_VEO_LIVE_BASE_URL")) + model := strings.TrimSpace(os.Getenv("GEMINI_VEO_LIVE_MODEL")) + if model == "" { + model = "veo-3.1-generate-preview" + } + ctx, cancel := context.WithTimeout(context.Background(), 12*time.Minute) + defer cancel() + operationName := strings.TrimSpace(os.Getenv("GEMINI_VEO_LIVE_OPERATION")) + response, err := (GeminiClient{}).Run(ctx, Request{ + Kind: "videos.generations", + ModelType: "video_generate", + Model: model, + Body: map[string]any{ + "prompt": "A small red paper boat gently floating on calm water, static camera, natural daylight.", + "duration": 4, + "resolution": "720p", + "aspect_ratio": "16:9", + }, + Candidate: store.RuntimeModelCandidate{ + Provider: "gemini", + BaseURL: baseURL, + ProviderModelName: model, + Credentials: map[string]any{"apiKey": apiKey}, + PlatformConfig: map[string]any{"pollIntervalMs": 10000, "pollTimeoutMs": 660000}, + }, + RemoteTaskID: operationName, + OnRemoteTaskSubmitted: func(remoteTaskID string, _ map[string]any) error { + t.Logf("live Gemini Veo submitted operation %s", remoteTaskID) + return nil + }, + }) + if err != nil { + t.Fatalf("live Gemini Veo request failed: %v", err) + } + data, _ := response.Result["data"].([]any) + if len(data) != 1 { + t.Fatalf("live Gemini Veo returned %d videos", len(data)) + } + item, _ := data[0].(map[string]any) + payload, _ := item["video_bytes"].([]byte) + if len(payload) < 1024 { + t.Fatalf("live Gemini Veo video is unexpectedly small: %d bytes", len(payload)) + } + if mimeType := strings.TrimSpace(stringFromAny(item["mime_type"])); !strings.HasPrefix(mimeType, "video/") { + t.Fatalf("live Gemini Veo returned unexpected MIME type: %q", mimeType) + } + t.Logf("live Gemini Veo accepted operation %s and downloaded %d video bytes", response.RequestID, len(payload)) +} + +func testGeminiVeoCandidate(baseURL string) store.RuntimeModelCandidate { + return store.RuntimeModelCandidate{ + Provider: "gemini", + BaseURL: baseURL, + ProviderModelName: "veo-3.1-generate-preview", + Credentials: map[string]any{"apiKey": "test-gemini-key"}, + PlatformConfig: map[string]any{"pollIntervalMs": 1, "pollTimeoutMs": 1000}, + } +} diff --git a/apps/api/internal/clients/responses_compat.go b/apps/api/internal/clients/responses_compat.go index 63c0783..0527677 100644 --- a/apps/api/internal/clients/responses_compat.go +++ b/apps/api/internal/clients/responses_compat.go @@ -18,6 +18,7 @@ const ( ProtocolOpenAIEmbeddings = "openai_embeddings" ProtocolOpenAIImages = "openai_images" ProtocolGeminiGenerateContent = "gemini_generate_content" + ProtocolGeminiVeo = "gemini_veo_predict_long_running" ProtocolVolcesContents = "volces_contents_generations_v3" ProtocolKlingV1Omni = "kling_v1_omni_video" ProtocolKlingV2Omni = "kling_v2_omni_video" diff --git a/apps/api/internal/runner/request_assets.go b/apps/api/internal/runner/request_assets.go index b65fcfc..40c728c 100644 --- a/apps/api/internal/runner/request_assets.go +++ b/apps/api/internal/runner/request_assets.go @@ -164,7 +164,7 @@ func (s *Service) hydrateProviderRequestAssetString(ctx context.Context, value s return value, nil } style, ok := requestAssetHydrateUnsupported, false - if openAIImageEditRequiresMultipartBytes(candidate) { + if geminiVeoRequiresInlineImage(candidate) || openAIImageEditRequiresMultipartBytes(candidate) { style, ok = requestAssetHydrateDataURL, true } else { style, ok = requestAssetCapabilityHydrationForMedia("image", candidate, raw, "") @@ -395,6 +395,9 @@ func requestAssetHydrationForField(path []string, asset store.RequestAsset, cand return requestAssetHydrateDataURL } if requestAssetMediaKindForHydration(path, asset) == "image" { + if geminiVeoRequiresInlineImage(candidate) { + return requestAssetHydrateDataURL + } if openAIImageEditRequiresMultipartBytes(candidate) { return requestAssetHydrateDataURL } @@ -418,6 +421,11 @@ func openAIImageEditRequiresMultipartBytes(candidate store.RuntimeModelCandidate strings.TrimSpace(candidate.ModelType) == "image_edit" } +func geminiVeoRequiresInlineImage(candidate store.RuntimeModelCandidate) bool { + return normalizeProviderKey(candidate.Provider) == "gemini" && + strings.Contains(strings.ToLower(strings.TrimSpace(firstNonEmptyString(candidate.ProviderModelName, candidate.ModelName))), "veo") +} + func requestAssetMediaKindForHydration(path []string, asset store.RequestAsset) string { if mediaURLFieldNeedsHydration(path) { return requestAssetMediaURLKind(path) diff --git a/apps/api/internal/runner/request_assets_test.go b/apps/api/internal/runner/request_assets_test.go index 1d05812..3888428 100644 --- a/apps/api/internal/runner/request_assets_test.go +++ b/apps/api/internal/runner/request_assets_test.go @@ -172,6 +172,24 @@ func TestHydrateProviderRequestAssetsConvertsGeminiInlineDataAssetToRawBase64(t } } +func TestGeminiVeoForcesOfficialInlineImageFormat(t *testing.T) { + candidate := store.RuntimeModelCandidate{ + Provider: "gemini", + ProviderModelName: "veo-3.1-generate-preview", + ModelType: "image_to_video", + Capabilities: map[string]any{ + "image_to_video": map[string]any{ + "support_url_input": true, + "support_base64_input": true, + }, + }, + } + asset := store.RequestAsset{URL: "https://cdn.example.com/input.png", StorageProvider: "remote", ContentType: "image/png"} + if got := requestAssetHydrationForField([]string{"image"}, asset, candidate); got != requestAssetHydrateDataURL { + t.Fatalf("Gemini Veo must hydrate public image URLs as data URLs, got %q", got) + } +} + func TestHydrateProviderRequestAssetsConvertsVolcesImageURLAssetToDataURL(t *testing.T) { storageDir := t.TempDir() fileName := "gateway-request-asset-chat-image.png" diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index de069f5..45d30ba 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -1919,6 +1919,9 @@ func compatibilitySourceProtocol(kind string, candidate store.RuntimeModelCandid return clients.ProtocolKlingV1Omni } case "gemini": + if kind == "videos.generations" && strings.Contains(strings.ToLower(strings.TrimSpace(firstNonEmptyString(candidate.ProviderModelName, candidate.ModelName))), "veo") { + return clients.ProtocolGeminiVeo + } return clients.ProtocolGeminiGenerateContent case "openai": switch kind {