diff --git a/apps/api/internal/clients/gemini.go b/apps/api/internal/clients/gemini.go index 102c6b2..7bf8152 100644 --- a/apps/api/internal/clients/gemini.go +++ b/apps/api/internal/clients/gemini.go @@ -61,7 +61,7 @@ func (c GeminiClient) Run(ctx context.Context, request Request) (Response, error } resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return Response{}, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return Response{}, transportClientError(err) } if err := notifyResponseReceived(request); err != nil { return Response{}, err @@ -294,7 +294,7 @@ func uploadOfficialGeminiImageFile(ctx context.Context, client *http.Client, can startRequest.Header.Set("Content-Type", "application/json") startResponse, err := client.Do(startRequest) if err != nil { - return nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, transportClientError(err) } if startResponse.StatusCode < http.StatusOK || startResponse.StatusCode >= http.StatusMultipleChoices { _, responseErr := decodeHTTPResponse(startResponse) @@ -314,7 +314,7 @@ func uploadOfficialGeminiImageFile(ctx context.Context, client *http.Client, can uploadRequest.Header.Set("Content-Type", mimeType) uploadResponse, err := client.Do(uploadRequest) if err != nil { - return nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, transportClientError(err) } result, err := decodeHTTPResponse(uploadResponse) if err != nil { @@ -392,7 +392,9 @@ func decodeGeminiStreamResponse(resp *http.Response, onDelta StreamDelta, wire * } } if err := scanner.Err(); err != nil { - return nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true, Wire: wire} + clientErr := transportClientError(err) + clientErr.Wire = wire + return nil, clientErr } if len(result) == 0 { return nil, &ClientError{Code: "invalid_response", Message: "gemini stream returned no events", StatusCode: resp.StatusCode, Retryable: false, Wire: wire} diff --git a/apps/api/internal/clients/gemini_veo.go b/apps/api/internal/clients/gemini_veo.go index 1df36ec..320d595 100644 --- a/apps/api/internal/clients/gemini_veo.go +++ b/apps/api/internal/clients/gemini_veo.go @@ -55,7 +55,7 @@ func (c GeminiClient) runVeo(ctx context.Context, request Request, apiKey string } 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") + timeout := durationFromConfig(request.Candidate.PlatformConfig, ProviderRequestTimeout(request.Kind), "pollTimeoutMs", "poll_timeout_ms", "timeoutMs") deadline := time.NewTimer(timeout) defer deadline.Stop() nextPoll := time.NewTimer(0) @@ -400,7 +400,7 @@ func (c GeminiClient) geminiVeoPost(ctx context.Context, request Request, apiKey } resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return nil, "", nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", nil, transportClientError(err) } if err := notifyResponseReceived(request); err != nil { resp.Body.Close() @@ -426,7 +426,7 @@ func (c GeminiClient) geminiVeoGetOperation(ctx context.Context, request Request 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} + return nil, "", nil, transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, wire, err := decodeHTTPResponseForProtocol(resp, ProtocolGeminiVeo) @@ -546,7 +546,7 @@ func (c GeminiClient) geminiVeoDownload(ctx context.Context, request Request, ap } resp, err := redirectClient.Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { diff --git a/apps/api/internal/clients/keling.go b/apps/api/internal/clients/keling.go index a353162..833db66 100644 --- a/apps/api/internal/clients/keling.go +++ b/apps/api/internal/clients/keling.go @@ -613,7 +613,7 @@ func (c KelingClient) postJSONAt(ctx context.Context, request Request, baseURL s req.Header.Set("Authorization", "Bearer "+token) resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return nil, "", nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", nil, transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, wire, err := decodeHTTPResponseForProtocol(resp, protocol) @@ -638,7 +638,7 @@ func (c KelingClient) getJSONAt(ctx context.Context, request Request, baseURL st req.Header.Set("Authorization", "Bearer "+token) resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, err := decodeHTTPResponse(resp) @@ -661,7 +661,7 @@ func (c KelingClient) createKelingElement(ctx context.Context, request Request, req.Header.Set("Authorization", "Bearer "+token) resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return "", transportClientError(err) } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 16*1024*1024)) @@ -745,7 +745,7 @@ func kelingImageToBase64(ctx context.Context, request Request, value string) (st } resp, err := httpClient(request.HTTPClient).Do(req) if err != nil { - return "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return "", transportClientError(err) } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { @@ -754,7 +754,7 @@ func kelingImageToBase64(ctx context.Context, request Request, value string) (st } raw, err := io.ReadAll(io.LimitReader(resp.Body, 16*1024*1024)) if err != nil { - return "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return "", transportClientError(err) } return base64.StdEncoding.EncodeToString(raw), nil } @@ -1373,9 +1373,10 @@ func kelingPollInterval(request Request) time.Duration { } func kelingPollTimeout(request Request) time.Duration { - seconds := numericValue(firstPresent(request.Candidate.PlatformConfig["kelingPollTimeoutSeconds"], request.Candidate.PlatformConfig["klingPollTimeoutSeconds"], request.Body["pollTimeoutSeconds"], request.Body["poll_timeout_seconds"]), 600) + fallbackSeconds := ProviderRequestTimeout(request.Kind).Seconds() + seconds := numericValue(firstPresent(request.Candidate.PlatformConfig["kelingPollTimeoutSeconds"], request.Candidate.PlatformConfig["klingPollTimeoutSeconds"], request.Body["pollTimeoutSeconds"], request.Body["poll_timeout_seconds"]), fallbackSeconds) if seconds < 1 { - seconds = 600 + seconds = fallbackSeconds } return time.Duration(seconds) * time.Second } diff --git a/apps/api/internal/clients/media_clients.go b/apps/api/internal/clients/media_clients.go index 148935c..f5ad9f3 100644 --- a/apps/api/internal/clients/media_clients.go +++ b/apps/api/internal/clients/media_clients.go @@ -646,7 +646,7 @@ func providerPostMultipartFile(ctx context.Context, client *http.Client, url str applyProviderAuth(req, credentials, auth) resp, err := client.Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, err := decodeHTTPResponse(resp) diff --git a/apps/api/internal/clients/media_timeout_test.go b/apps/api/internal/clients/media_timeout_test.go new file mode 100644 index 0000000..08ee646 --- /dev/null +++ b/apps/api/internal/clients/media_timeout_test.go @@ -0,0 +1,47 @@ +package clients + +import ( + "context" + "testing" + "time" +) + +func TestProviderRequestTimeoutUsesMediaDefaults(t *testing.T) { + for _, test := range []struct { + kind string + want time.Duration + }{ + {kind: "images.generations", want: 20 * time.Minute}, + {kind: "images.edits", want: 20 * time.Minute}, + {kind: "videos.generations", want: 30 * time.Minute}, + {kind: "chat.completions", want: 10 * time.Minute}, + } { + if got := ProviderRequestTimeout(test.kind); got != test.want { + t.Fatalf("timeout for %s: got %s want %s", test.kind, got, test.want) + } + } +} + +func TestTransportTimeoutUsesTerminalTimeoutCode(t *testing.T) { + err := transportClientError(context.DeadlineExceeded) + if err.Code != "timeout" || err.Retryable { + t.Fatalf("transport timeout classification = %+v, want terminal timeout", err) + } +} + +func TestMediaPollTimeoutsUseTaskDefaultsWithoutPlatformOverride(t *testing.T) { + image := Request{Kind: "images.generations", Candidate: storeCandidateWithConfig("", "", nil, nil)} + video := Request{Kind: "videos.generations", Candidate: storeCandidateWithConfig("", "", nil, nil)} + if got := providerPollTimeout(image); got != 20*time.Minute { + t.Fatalf("image provider poll timeout: got %s want %s", got, 20*time.Minute) + } + if got := providerPollTimeout(video); got != 30*time.Minute { + t.Fatalf("video provider poll timeout: got %s want %s", got, 30*time.Minute) + } + if got := kelingPollTimeout(video); got != 30*time.Minute { + t.Fatalf("Keling video poll timeout: got %s want %s", got, 30*time.Minute) + } + if got := volcesPollTimeout(video); got != 30*time.Minute { + t.Fatalf("Volces video poll timeout: got %s want %s", got, 30*time.Minute) + } +} diff --git a/apps/api/internal/clients/openai.go b/apps/api/internal/clients/openai.go index a67ed24..4dbb86c 100644 --- a/apps/api/internal/clients/openai.go +++ b/apps/api/internal/clients/openai.go @@ -82,7 +82,8 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error upstreamEndpoint := joinURL(openAIBaseURL(endpointKind, request.Candidate), endpoint) responseStartedAt := time.Now() correctionScope := newParameterCorrectionScope(request, endpointKind) - correctionEnabled := endpointKind == "chat.completions" || endpointKind == "responses" + correctionEnabled := endpointKind == "chat.completions" || endpointKind == "responses" || + endpointKind == "images.generations" || endpointKind == "images.edits" protectedCorrections := callerProtectedCorrectionParameters(request, endpointKind) if correctionEnabled { c.Corrections.apply(correctionScope, body, protectedCorrections) @@ -108,7 +109,7 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error } resp, requestErr = requestClient.Do(req) if requestErr != nil { - return Response{}, &ClientError{Code: "network", Message: requestErr.Error(), Retryable: true} + return Response{}, transportClientError(requestErr) } if err := notifyResponseReceived(request); err != nil { _ = resp.Body.Close() diff --git a/apps/api/internal/clients/openai_image_correction_test.go b/apps/api/internal/clients/openai_image_correction_test.go new file mode 100644 index 0000000..a656e2f --- /dev/null +++ b/apps/api/internal/clients/openai_image_correction_test.go @@ -0,0 +1,92 @@ +package clients + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestOpenAIImageEndpointsRemoveRejectedResponseFormatAndCacheCorrection(t *testing.T) { + for _, test := range []struct { + name string + kind string + body map[string]any + }{ + { + name: "generation JSON", + kind: "images.generations", + body: map[string]any{"prompt": "test", "response_format": "url"}, + }, + { + name: "edit multipart", + kind: "images.edits", + body: map[string]any{"prompt": "test", "image": "aW1hZ2U=", "response_format": "url"}, + }, + } { + t.Run(test.name, func(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestNo := requests.Add(1) + responseFormat := "" + if test.kind == "images.edits" { + if err := r.ParseMultipartForm(1 << 20); err != nil { + t.Fatalf("parse multipart image request: %v", err) + } + responseFormat = r.FormValue("response_format") + } else { + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatalf("decode image request: %v", err) + } + responseFormat, _ = body["response_format"].(string) + } + if requestNo == 1 { + if responseFormat != "url" { + t.Fatalf("first request lost caller response_format: %q", responseFormat) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{"error": map[string]any{ + "message": "Unknown parameter: 'response_format'.", + "type": "invalid_request_error", "param": "response_format", "code": "unknown_parameter", + }}) + return + } + if responseFormat != "" { + t.Fatalf("corrected request still contains response_format: %q", responseFormat) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"data": []any{map[string]any{"b64_json": "aW1hZ2U="}}}) + })) + defer server.Close() + + cache := NewParameterCorrectionCache() + request := Request{ + Kind: test.kind, Model: "gpt-image-2", Body: test.body, OriginalBody: test.body, + UpstreamIdempotencyKey: "gateway-task-id", + Candidate: store.RuntimeModelCandidate{ + Provider: "openai", BaseURL: server.URL, ProviderModelName: "gpt-image-2", + Credentials: map[string]any{"apiKey": "test-key"}, + }, + } + client := OpenAIClient{HTTPClient: server.Client(), Corrections: cache} + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("correct rejected response_format: %v", err) + } + if requests.Load() != 2 { + t.Fatalf("first corrected call requests=%d, want 2", requests.Load()) + } + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("reuse cached response_format correction: %v", err) + } + if requests.Load() != 3 { + t.Fatalf("cached correction should avoid another 400, requests=%d", requests.Load()) + } + }) + } +} diff --git a/apps/api/internal/clients/parameter_correction.go b/apps/api/internal/clients/parameter_correction.go index 5282f45..b3415e6 100644 --- a/apps/api/internal/clients/parameter_correction.go +++ b/apps/api/internal/clients/parameter_correction.go @@ -22,6 +22,7 @@ var safeUpstreamCorrectionParameters = stringSet( "temperature", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "n", "max_tokens", "max_completion_tokens", "max_output_tokens", "logprobs", "top_logprobs", "service_tier", "verbosity", + "response_format", ) var ( @@ -286,6 +287,12 @@ func callerProtectedCorrectionParameters(request Request, endpointKind string) m } } } + if endpointKind == "images.generations" || endpointKind == "images.edits" { + // The Gateway canonicalizes image output independently of the provider's + // URL/Base64 preference. Allow an explicitly supplied response_format to + // be removed only after the selected upstream rejects it as unsupported. + delete(protected, "response_format") + } return protected } diff --git a/apps/api/internal/clients/provider_task.go b/apps/api/internal/clients/provider_task.go index 7c8aed8..4f3e8c3 100644 --- a/apps/api/internal/clients/provider_task.go +++ b/apps/api/internal/clients/provider_task.go @@ -190,7 +190,7 @@ func providerPostJSON(ctx context.Context, client *http.Client, url string, body applyProviderAuth(req, credentials, auth) resp, err := client.Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, err := decodeHTTPResponse(resp) @@ -205,7 +205,7 @@ func providerGetJSON(ctx context.Context, client *http.Client, url string, crede applyProviderAuth(req, credentials, auth) resp, err := client.Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, err := decodeHTTPResponse(resp) @@ -460,7 +460,7 @@ func providerPollInterval(request Request) time.Duration { } func providerPollTimeout(request Request) time.Duration { - return durationFromConfig(request.Candidate.PlatformConfig, 10*time.Minute, "pollTimeoutMs", "poll_timeout_ms", "timeoutMs") + return durationFromConfig(request.Candidate.PlatformConfig, ProviderRequestTimeout(request.Kind), "pollTimeoutMs", "poll_timeout_ms", "timeoutMs") } func durationFromConfig(config map[string]any, fallback time.Duration, keys ...string) time.Duration { diff --git a/apps/api/internal/clients/topaz.go b/apps/api/internal/clients/topaz.go index 6df4085..e74e739 100644 --- a/apps/api/internal/clients/topaz.go +++ b/apps/api/internal/clients/topaz.go @@ -163,7 +163,7 @@ func (c TopazClient) prepareSource(ctx context.Context, request Request, rawURL downloadClient := topazSourceHTTPClient(httpClient(request.HTTPClient, c.HTTPClient), boolishDefault(request.Candidate.PlatformConfig["allowPrivateSourceDownloads"], false)) resp, err := downloadClient.Do(req) if err != nil { - return topazSource{}, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return topazSource{}, transportClientError(err) } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { @@ -287,7 +287,7 @@ func (c TopazClient) uploadSource(ctx context.Context, request Request, apiKey, req.Header.Set("Content-Type", topazContainerContentType(source.Container)) resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return transportClientError(err) } _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<20)) resp.Body.Close() @@ -353,7 +353,7 @@ func (c TopazClient) topazJSON(ctx context.Context, request Request, apiKey, met } resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, transportClientError(err) } result, decodeErr := decodeHTTPResponse(resp) if decodeErr != nil { diff --git a/apps/api/internal/clients/types.go b/apps/api/internal/clients/types.go index 344d2f3..51bc88e 100644 --- a/apps/api/internal/clients/types.go +++ b/apps/api/internal/clients/types.go @@ -3,6 +3,7 @@ package clients import ( "context" "errors" + "net" "net/http" "strings" "time" @@ -10,6 +11,41 @@ import ( "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) +const ( + defaultProviderRequestTimeout = 10 * time.Minute + imageProviderRequestTimeout = 20 * time.Minute + videoProviderRequestTimeout = 30 * time.Minute +) + +// ProviderRequestTimeout returns the default end-to-end provider request or +// polling budget for a task kind. Explicit platform polling configuration may +// still override these defaults for providers with a documented requirement. +func ProviderRequestTimeout(kind string) time.Duration { + switch { + case strings.HasPrefix(strings.TrimSpace(kind), "images."): + return imageProviderRequestTimeout + case strings.HasPrefix(strings.TrimSpace(kind), "videos."): + return videoProviderRequestTimeout + default: + return defaultProviderRequestTimeout + } +} + +func transportClientError(err error) *ClientError { + if err == nil { + return nil + } + var netErr net.Error + if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &netErr) && netErr.Timeout()) { + return &ClientError{ + Code: "timeout", + Message: "upstream request timed out: " + err.Error(), + Retryable: false, + } + } + return &ClientError{Code: "network", Message: err.Error(), Retryable: true} +} + type Request struct { Kind string ModelType string diff --git a/apps/api/internal/clients/universal.go b/apps/api/internal/clients/universal.go index bf528bf..95bd023 100644 --- a/apps/api/internal/clients/universal.go +++ b/apps/api/internal/clients/universal.go @@ -153,7 +153,7 @@ func (c UniversalClient) universalSubmit(ctx context.Context, executor *scripten func (c UniversalClient) universalPollUntilDone(ctx context.Context, executor *scriptengine.Executor, request Request, modelType string, upstreamTaskID string, payload map[string]any, requestID string, startedAt time.Time) (map[string]any, string, error) { interval := universalDurationConfig(request.Candidate.PlatformConfig, 2*time.Second, "pollIntervalMs", "poll_interval_ms") - timeout := universalDurationConfig(request.Candidate.PlatformConfig, 10*time.Minute, "pollTimeoutMs", "poll_timeout_ms", "timeoutMs") + timeout := universalDurationConfig(request.Candidate.PlatformConfig, ProviderRequestTimeout(request.Kind), "pollTimeoutMs", "poll_timeout_ms", "timeoutMs") deadline := time.NewTimer(timeout) defer deadline.Stop() ticker := time.NewTicker(interval) @@ -234,7 +234,7 @@ func universalScriptContext(request Request, modelType string, payload map[strin "platformModelId": request.Candidate.PlatformModelID, "canonicalModelKey": request.Candidate.CanonicalModelKey, "modelType": modelType, - "timeout": universalDurationConfig(request.Candidate.PlatformConfig, 10*time.Minute, "pollTimeoutMs", "poll_timeout_ms").Milliseconds(), + "timeout": universalDurationConfig(request.Candidate.PlatformConfig, ProviderRequestTimeout(request.Kind), "pollTimeoutMs", "poll_timeout_ms").Milliseconds(), }, "env": cloneMapAny(request.Candidate.PlatformConfig), "candidate": universalCandidateSnapshot(request), @@ -318,7 +318,7 @@ func universalPostJSON(ctx context.Context, client *http.Client, baseURL string, } resp, err := client.Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, err := decodeHTTPResponse(resp) @@ -335,7 +335,7 @@ func universalGetJSON(ctx context.Context, client *http.Client, url string, cred } resp, err := client.Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, err := decodeHTTPResponse(resp) diff --git a/apps/api/internal/clients/vectorizer.go b/apps/api/internal/clients/vectorizer.go index cbde679..b8a3f51 100644 --- a/apps/api/internal/clients/vectorizer.go +++ b/apps/api/internal/clients/vectorizer.go @@ -78,7 +78,7 @@ func (c VectorizerClient) Run(ctx context.Context, request Request) (Response, e client := httpClient(request.HTTPClient, c.HTTPClient) resp, err := client.Do(req) if err != nil { - return Response{}, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return Response{}, transportClientError(err) } defer resp.Body.Close() requestID := requestIDFromHTTPResponse(resp) diff --git a/apps/api/internal/clients/volces.go b/apps/api/internal/clients/volces.go index e68e3c3..92f05a3 100644 --- a/apps/api/internal/clients/volces.go +++ b/apps/api/internal/clients/volces.go @@ -52,7 +52,7 @@ func (c VolcesClient) runImage(ctx context.Context, request Request, apiKey stri } resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return Response{}, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return Response{}, transportClientError(err) } if err := notifyResponseReceived(request); err != nil { return Response{}, err @@ -199,7 +199,7 @@ func (c VolcesClient) DeleteVideoTask(ctx context.Context, request Request) (map req.Header.Set("Authorization", "Bearer "+apiKey) response, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return nil, "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", transportClientError(err) } requestID := requestIDFromHTTPResponse(response) result, err := decodeHTTPResponse(response) @@ -239,7 +239,7 @@ func (c VolcesClient) postJSON(ctx context.Context, request Request, baseURL str } resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return nil, "", nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", nil, transportClientError(err) } if err := notifyResponseReceived(request); err != nil { return nil, "", nil, err @@ -271,7 +271,7 @@ func (c VolcesClient) getJSON(ctx context.Context, request Request, baseURL stri req.Header.Set("Authorization", "Bearer "+apiKey) resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) if err != nil { - return nil, "", nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return nil, "", nil, transportClientError(err) } requestID := requestIDFromHTTPResponse(resp) result, wire, err := decodeHTTPResponseForProtocol(resp, ProtocolVolcesContents) @@ -1184,9 +1184,10 @@ func volcesPollInterval(request Request) time.Duration { } func volcesPollTimeout(request Request) time.Duration { - seconds := numericValue(firstPresent(request.Candidate.PlatformConfig["volcesPollTimeoutSeconds"], request.Body["pollTimeoutSeconds"], request.Body["poll_timeout_seconds"]), 600) + fallbackSeconds := ProviderRequestTimeout(request.Kind).Seconds() + seconds := numericValue(firstPresent(request.Candidate.PlatformConfig["volcesPollTimeoutSeconds"], request.Body["pollTimeoutSeconds"], request.Body["poll_timeout_seconds"]), fallbackSeconds) if seconds < 1 { - seconds = 600 + seconds = fallbackSeconds } return time.Duration(seconds) * time.Second } diff --git a/apps/api/internal/clients/volces_assets.go b/apps/api/internal/clients/volces_assets.go index 3adb11b..454bcce 100644 --- a/apps/api/internal/clients/volces_assets.go +++ b/apps/api/internal/clients/volces_assets.go @@ -106,7 +106,7 @@ func (c VolcesAssetClient) call(ctx context.Context, credentials VolcesAssetCred response, err := httpClient(nil, c.HTTPClient).Do(req) if err != nil { - return "", &ClientError{Code: "network", Message: err.Error(), Retryable: true} + return "", transportClientError(err) } defer response.Body.Close() var envelope struct { diff --git a/apps/api/internal/runner/proxy.go b/apps/api/internal/runner/proxy.go index 8a8bb7a..2e4f96f 100644 --- a/apps/api/internal/runner/proxy.go +++ b/apps/api/internal/runner/proxy.go @@ -5,7 +5,6 @@ import ( "net/url" "strings" "sync" - "time" "github.com/easyai/easyai-ai-gateway/apps/api/internal/clients" "github.com/easyai/easyai-ai-gateway/apps/api/internal/netproxy" @@ -19,8 +18,6 @@ type httpClientCache struct { custom map[string]*http.Client } -const providerHTTPClientTimeout = 10 * time.Minute - const ( providerHTTPMaxIdleConnections = 2048 providerHTTPMaxIdleConnectionsPerHost = 1024 @@ -34,16 +31,19 @@ func newHTTPClientCache() *httpClientCache { } } -func (s *Service) httpClientForCandidate(candidate store.RuntimeModelCandidate, simulated bool) (*http.Client, error) { +func (s *Service) httpClientForCandidate(candidate store.RuntimeModelCandidate, simulated bool, kind string) (*http.Client, error) { + var client *http.Client if simulated { - return s.httpClients.none, nil + client = s.httpClients.none + return providerHTTPClientForKind(client, kind), nil } // Some Worker sites have direct provider egress but are not authorized to // use a platform's shared proxy. Keep this override explicitly scoped to // platform UUIDs so one site's routing exception cannot bypass proxies for // unrelated providers or platforms. if platformIDListed(s.cfg.PlatformProxyBypassIDs, candidate.PlatformID) { - return s.httpClients.none, nil + client = s.httpClients.none + return providerHTTPClientForKind(client, kind), nil } config, err := netproxy.Normalize(netproxy.FromPlatformConfig(candidate.PlatformConfig)) if err != nil { @@ -52,14 +52,22 @@ func (s *Service) httpClientForCandidate(candidate store.RuntimeModelCandidate, switch config.Mode { case netproxy.ModeGlobal: if strings.TrimSpace(s.cfg.GlobalHTTPProxy) != "" { - return s.httpClients.customClient(s.cfg.GlobalHTTPProxy) + client, err = s.httpClients.customClient(s.cfg.GlobalHTTPProxy) + if err != nil { + return nil, err + } + break } - return s.httpClients.global, nil + client = s.httpClients.global case netproxy.ModeCustom: - return s.httpClients.customClient(config.HTTPProxy) + client, err = s.httpClients.customClient(config.HTTPProxy) + if err != nil { + return nil, err + } default: - return s.httpClients.none, nil + client = s.httpClients.none } + return providerHTTPClientForKind(client, kind), nil } func platformIDListed(raw string, platformID string) bool { @@ -100,7 +108,16 @@ func newHTTPClient(proxy func(*http.Request) (*url.URL, error)) *http.Client { transport.MaxIdleConns = providerHTTPMaxIdleConnections transport.MaxIdleConnsPerHost = providerHTTPMaxIdleConnectionsPerHost return &http.Client{ - Timeout: providerHTTPClientTimeout, + Timeout: clients.ProviderRequestTimeout(""), Transport: transport, } } + +func providerHTTPClientForKind(base *http.Client, kind string) *http.Client { + if base == nil || base.Timeout == clients.ProviderRequestTimeout(kind) { + return base + } + client := *base + client.Timeout = clients.ProviderRequestTimeout(kind) + return &client +} diff --git a/apps/api/internal/runner/proxy_test.go b/apps/api/internal/runner/proxy_test.go index c699ad3..d313a45 100644 --- a/apps/api/internal/runner/proxy_test.go +++ b/apps/api/internal/runner/proxy_test.go @@ -16,6 +16,19 @@ func TestProviderHTTPClientTimeoutAllowsLongRunningMediaRequests(t *testing.T) { if client.Timeout != 10*time.Minute { t.Fatalf("unexpected provider HTTP timeout: got %s want %s", client.Timeout, 10*time.Minute) } + for _, test := range []struct { + kind string + want time.Duration + }{ + {kind: "images.generations", want: 20 * time.Minute}, + {kind: "images.edits", want: 20 * time.Minute}, + {kind: "videos.generations", want: 30 * time.Minute}, + {kind: "chat.completions", want: 10 * time.Minute}, + } { + if got := providerHTTPClientForKind(client, test.kind).Timeout; got != test.want { + t.Fatalf("provider HTTP timeout for %s: got %s want %s", test.kind, got, test.want) + } + } transport, ok := client.Transport.(*http.Transport) if !ok { t.Fatalf("provider transport type = %T, want *http.Transport", client.Transport) @@ -51,7 +64,7 @@ func TestPlatformProxyModeNoneIgnoresEnvironmentProxy(t *testing.T) { client, err := testProxyService(config.Config{}).httpClientForCandidate(store.RuntimeModelCandidate{ PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "none"}}, - }, false) + }, false, "chat.completions") if err != nil { t.Fatalf("build http client: %v", err) } @@ -86,7 +99,7 @@ func TestPlatformProxyModeCustomUsesConfiguredHTTPProxy(t *testing.T) { client, err := testProxyService(config.Config{}).httpClientForCandidate(store.RuntimeModelCandidate{ PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "custom", "httpProxy": proxy.URL}}, - }, false) + }, false, "chat.completions") if err != nil { t.Fatalf("build http client: %v", err) } @@ -121,7 +134,7 @@ func TestPlatformProxyBypassIDUsesDirectConnection(t *testing.T) { }).httpClientForCandidate(store.RuntimeModelCandidate{ PlatformID: "official-gemini-platform", PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "custom", "httpProxy": proxy.URL}}, - }, false) + }, false, "chat.completions") if err != nil { t.Fatalf("build bypassed http client: %v", err) } @@ -153,7 +166,7 @@ func TestPlatformProxyModeGlobalUsesConfiguredGlobalHTTPProxy(t *testing.T) { client, err := testProxyService(config.Config{GlobalHTTPProxy: proxy.URL}).httpClientForCandidate(store.RuntimeModelCandidate{ PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "global"}}, - }, false) + }, false, "chat.completions") if err != nil { t.Fatalf("build http client: %v", err) } diff --git a/apps/api/internal/runner/retry_decision.go b/apps/api/internal/runner/retry_decision.go index 8b55cb2..40c5093 100644 --- a/apps/api/internal/runner/retry_decision.go +++ b/apps/api/internal/runner/retry_decision.go @@ -63,6 +63,7 @@ type resolveCandidateFailureInput struct { FailoverExpired bool Async bool DownstreamStarted bool + TaskKind string } type priorityDemoteDecision struct { @@ -150,6 +151,20 @@ func failoverDecisionForCandidate(runnerPolicy store.RunnerPolicy, candidate sto func resolveCandidateFailure(input resolveCandidateFailureInput) failureDecision { info := failureInfoFromError(input.Err) + if terminalMediaTimeout(input.TaskKind, input.Err) { + return failureDecision{ + Route: "stop", + Effect: "none", + Reason: "media_timeout_terminal", + Match: policyRuleMatch{ + Source: "gateway_media_timeout", + Policy: "terminalTimeout", + Rule: "taskKind", + Value: strings.TrimSpace(input.TaskKind), + }, + Info: info, + } + } if isResultPersistenceFailure(input.Err) { return failureDecision{ Route: "stop", @@ -260,6 +275,12 @@ func resolveCandidateFailure(input resolveCandidateFailureInput) failureDecision } } +func terminalMediaTimeout(kind string, err error) bool { + kind = strings.TrimSpace(kind) + return (strings.HasPrefix(kind, "images.") || strings.HasPrefix(kind, "videos.")) && + strings.EqualFold(strings.TrimSpace(clients.ErrorCode(err)), "timeout") +} + func failoverEffect(action string) string { switch action { case "cooldown_and_next": diff --git a/apps/api/internal/runner/retry_decision_test.go b/apps/api/internal/runner/retry_decision_test.go index 25113cb..a82816e 100644 --- a/apps/api/internal/runner/retry_decision_test.go +++ b/apps/api/internal/runner/retry_decision_test.go @@ -442,6 +442,25 @@ func TestResolveCandidateFailureStopsAfterDownstreamStarts(t *testing.T) { } } +func TestResolveCandidateFailureStopsTerminalMediaTimeout(t *testing.T) { + decision := resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{ + Status: "active", + FailoverPolicy: map[string]any{"enabled": true}, + }, + Candidate: store.RuntimeModelCandidate{ModelRetryPolicy: map[string]any{"enabled": true, "maxAttempts": 3}}, + Err: &clients.ClientError{Code: "timeout", Message: "upstream request timed out", Retryable: true}, + ClientAttempt: 1, + MaxClientAttempts: 3, + HasNextCandidate: true, + TaskKind: "images.edits", + }) + + if decision.Route != "stop" || decision.Effect != "none" || decision.Reason != "media_timeout_terminal" { + t.Fatalf("media timeout must stop without retry or rotation, got %+v", decision) + } +} + func TestResolveCandidateFailureLegacyPolicyPrecedence(t *testing.T) { candidate := store.RuntimeModelCandidate{ DegradePolicy: map[string]any{ diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index d577ceb..f322872 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -81,6 +81,10 @@ type upstreamSubmissionUnknownError struct { Cause error } +func shouldClassifyUpstreamSubmissionUnknown(simulated bool, submissionStatus string, err error) bool { + return !simulated && submissionStatus == "submitting" && clients.ErrorCode(err) != "timeout" +} + func (e *upstreamSubmissionUnknownError) Error() string { return "upstream submission result is unknown" } @@ -1105,6 +1109,7 @@ candidatesLoop: FailoverExpired: failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration), Async: task.AsyncMode, DownstreamStarted: downstreamStarted.Load(), + TaskKind: task.Kind, }) if candidateDecision.Route == "requeue" { if _, platformLimited := platformModelRateLimitError(err); platformLimited { @@ -1162,6 +1167,7 @@ candidatesLoop: FailoverExpired: failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration), Async: task.AsyncMode, DownstreamStarted: downstreamStarted.Load(), + TaskKind: task.Kind, }) candidateDecisionAttempt = attemptNo candidateClientAttempt = clientAttempt @@ -1431,7 +1437,7 @@ func (s *Service) runCandidate( } defer s.store.RecordClientRelease(context.WithoutCancel(ctx), candidate.ClientID, "") - requestHTTPClient, err := s.httpClientForCandidate(candidate, simulated) + requestHTTPClient, err := s.httpClientForCandidate(candidate, simulated, task.Kind) if err != nil { _ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{ AttemptID: attemptID, @@ -1685,7 +1691,7 @@ func (s *Service) runCandidate( ErrorMessage: err.Error(), }) _ = s.emit(ctx, task.ID, "task.attempt.failed", "running", "attempt_failed", 0.45, err.Error(), map[string]any{"attempt": attemptNo, "retryable": retryable, "requestId": requestID, "statusCode": clients.ErrorResponseMetadata(err).StatusCode, "metrics": metrics}, simulated) - if !simulated && submissionStatus == "submitting" { + if shouldClassifyUpstreamSubmissionUnknown(simulated, submissionStatus, err) { return clients.Response{}, &upstreamSubmissionUnknownError{AttemptID: attemptID, Cause: err} } return clients.Response{}, err diff --git a/apps/api/internal/runner/submission_timeout_test.go b/apps/api/internal/runner/submission_timeout_test.go new file mode 100644 index 0000000..5a025e3 --- /dev/null +++ b/apps/api/internal/runner/submission_timeout_test.go @@ -0,0 +1,18 @@ +package runner + +import ( + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/clients" +) + +func TestTimeoutDoesNotBecomeUpstreamSubmissionUnknown(t *testing.T) { + timeoutErr := &clients.ClientError{Code: "timeout", Message: "upstream request timed out", Retryable: false} + if shouldClassifyUpstreamSubmissionUnknown(false, "submitting", timeoutErr) { + t.Fatal("a definitive provider timeout must remain timeout instead of manual-review unknown") + } + networkErr := &clients.ClientError{Code: "network", Message: "connection reset", Retryable: true} + if !shouldClassifyUpstreamSubmissionUnknown(false, "submitting", networkErr) { + t.Fatal("an ambiguous non-timeout disconnect must retain manual-review protection") + } +} diff --git a/apps/api/internal/runner/task_cancel.go b/apps/api/internal/runner/task_cancel.go index 5a304dd..69fc3f7 100644 --- a/apps/api/internal/runner/task_cancel.go +++ b/apps/api/internal/runner/task_cancel.go @@ -129,7 +129,7 @@ func (s *Service) CancelVolcesVideoTask(ctx context.Context, task store.GatewayT if !found || !isVolcesCancellationCandidate(candidate) { return local, nil } - httpClient, err := s.httpClientForCandidate(candidate, false) + httpClient, err := s.httpClientForCandidate(candidate, false, "videos.generations") if err != nil { return TaskCancelResult{}, err } diff --git a/apps/api/internal/runner/voice_clone.go b/apps/api/internal/runner/voice_clone.go index cb59c96..177f19a 100644 --- a/apps/api/internal/runner/voice_clone.go +++ b/apps/api/internal/runner/voice_clone.go @@ -210,7 +210,7 @@ func (s *Service) DeleteClonedVoice(ctx context.Context, user *auth.User, rawID if err != nil { return DeletedClonedVoiceResult{}, err } - requestHTTPClient, err := s.httpClientForCandidate(candidate, false) + requestHTTPClient, err := s.httpClientForCandidate(candidate, false, "voice.clone") if err != nil { return DeletedClonedVoiceResult{}, err }