Skip to content

Commit 9e71cfa

Browse files
Bowl42Bowl42
committed
fix(proxy): align stream errors and video API paths
1 parent 3b88bf8 commit 9e71cfa

13 files changed

Lines changed: 241 additions & 33 deletions

File tree

‎internal/adapter/client/adapter.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -257,19 +257,22 @@ func isOpenAIImagesPath(path string) bool {
257257
}
258258

259259
// IsVideoGenerationsPath matches the async video-generation surface: the exact
260-
// submit path (POST /v1/video/generations) and the poll subpath
261-
// (GET /v1/video/generations/{task_id}). The poll carries no body/model, so it
262-
// must be classified by path. Exported so the ingress gate can allow GET polls.
260+
// submit path (POST /v1/video/generations or the OpenAI-compatible
261+
// POST /v1/videos alias) and the poll subpath (GET /.../{task_id}). The poll
262+
// carries no body/model, so it must be classified by path. Exported so the
263+
// ingress gate can allow GET polls.
263264
func IsVideoGenerationsPath(path string) bool {
264265
return path == "/v1/video/generations" || strings.HasPrefix(path, "/v1/video/generations/") ||
265-
path == "/video/generations" || strings.HasPrefix(path, "/video/generations/")
266+
path == "/video/generations" || strings.HasPrefix(path, "/video/generations/") ||
267+
path == "/v1/videos" || strings.HasPrefix(path, "/v1/videos/") ||
268+
path == "/videos" || strings.HasPrefix(path, "/videos/")
266269
}
267270

268271
// IsVideoPollPath reports whether path is a video-generation POLL — the
269272
// collection path plus a non-empty task id (e.g. /v1/video/generations/task_abc).
270273
// Only the poll is a GET; the bare collection path is submit-only (POST).
271274
func IsVideoPollPath(path string) bool {
272-
for _, base := range []string{"/v1/video/generations/", "/video/generations/"} {
275+
for _, base := range []string{"/v1/video/generations/", "/video/generations/", "/v1/videos/", "/videos/"} {
273276
if strings.HasPrefix(path, base) && strings.TrimPrefix(path, base) != "" {
274277
return true
275278
}

‎internal/adapter/client/adapter_test.go‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -86,22 +86,22 @@ func TestDetectClientTypeRecognizesVideoGenerationsPath(t *testing.T) {
8686
adapter := NewAdapter()
8787

8888
// Submit: POST with a JSON body carrying the model.
89-
submitBody := []byte(`{"model":"doubao-seedance-2-0-260128","prompt":"a cat runs"}`)
90-
for _, path := range []string{"/v1/video/generations", "/video/generations"} {
89+
submitBody := []byte(`{"model":"video-test-model","prompt":"a cat runs"}`)
90+
for _, path := range []string{"/v1/video/generations", "/video/generations", "/v1/videos", "/videos"} {
9191
req := httptest.NewRequest("POST", path, strings.NewReader(string(submitBody)))
9292
if got := adapter.DetectClientType(req, submitBody); got != domain.ClientTypeVideo {
9393
t.Fatalf("DetectClientType(POST %s) = %s, want %s", path, got, domain.ClientTypeVideo)
9494
}
9595
if got, ok := adapter.Match(req); !ok || got != domain.ClientTypeVideo {
9696
t.Fatalf("Match(POST %s) = (%s, %v), want (%s, true)", path, got, ok, domain.ClientTypeVideo)
9797
}
98-
if got := adapter.ExtractModel(req, submitBody, domain.ClientTypeVideo); got != "doubao-seedance-2-0-260128" {
99-
t.Fatalf("ExtractModel(%s) = %q, want the seedance model", path, got)
98+
if got := adapter.ExtractModel(req, submitBody, domain.ClientTypeVideo); got != "video-test-model" {
99+
t.Fatalf("ExtractModel(%s) = %q, want video-test-model", path, got)
100100
}
101101
}
102102

103103
// Poll: GET /{task_id} with no body — classified by path, model is empty.
104-
for _, path := range []string{"/v1/video/generations/task_abc123", "/video/generations/task_abc123"} {
104+
for _, path := range []string{"/v1/video/generations/task_abc123", "/video/generations/task_abc123", "/v1/videos/task_abc123", "/videos/task_abc123"} {
105105
req := httptest.NewRequest("GET", path, nil)
106106
if got := adapter.DetectClientType(req, nil); got != domain.ClientTypeVideo {
107107
t.Fatalf("DetectClientType(GET %s) = %s, want %s", path, got, domain.ClientTypeVideo)
@@ -118,7 +118,7 @@ func TestDetectClientTypeRecognizesVideoGenerationsPath(t *testing.T) {
118118
// Only the poll (collection path + a non-empty task id) may be a GET; the bare
119119
// submit endpoints stay POST-only, so the method gate must not treat them as polls.
120120
func TestIsVideoPollPath(t *testing.T) {
121-
polls := []string{"/v1/video/generations/task_abc123", "/video/generations/task_abc123"}
121+
polls := []string{"/v1/video/generations/task_abc123", "/video/generations/task_abc123", "/v1/videos/task_abc123", "/videos/task_abc123"}
122122
for _, p := range polls {
123123
if !IsVideoPollPath(p) {
124124
t.Fatalf("IsVideoPollPath(%s) = false, want true", p)
@@ -127,6 +127,8 @@ func TestIsVideoPollPath(t *testing.T) {
127127
notPolls := []string{
128128
"/v1/video/generations", "/video/generations",
129129
"/v1/video/generations/", "/video/generations/", // trailing slash, no task id
130+
"/v1/videos", "/videos",
131+
"/v1/videos/", "/videos/", // trailing slash, no task id
130132
"/v1/chat/completions",
131133
}
132134
for _, p := range notPolls {

‎internal/adapter/provider/custom/adapter.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1068,6 +1068,8 @@ func normalizeOpenAIUpstreamRequestPath(requestPath string) string {
10681068
return "/v1" + requestPath
10691069
case requestPath == "/video/generations" || strings.HasPrefix(requestPath, "/video/generations/"):
10701070
return "/v1" + requestPath
1071+
case requestPath == "/videos" || strings.HasPrefix(requestPath, "/videos/"):
1072+
return "/v1" + requestPath
10711073
case requestPath == "/models" || strings.HasPrefix(requestPath, "/models?"):
10721074
return "/v1" + requestPath
10731075
default:

‎internal/adapter/provider/custom/adapter_url_test.go‎

Lines changed: 24 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -94,21 +94,39 @@ func TestBuildUpstreamURLNormalizesOpenAIBaseRoots(t *testing.T) {
9494
},
9595
{
9696
name: "root-style video submit path gains v1",
97-
baseURL: "https://code0.ai",
97+
baseURL: "https://video.example.test",
9898
requestPath: "/video/generations",
99-
want: "https://code0.ai/v1/video/generations",
99+
want: "https://video.example.test/v1/video/generations",
100100
},
101101
{
102102
name: "root-style video poll path gains v1",
103-
baseURL: "https://code0.ai",
103+
baseURL: "https://video.example.test",
104104
requestPath: "/video/generations/task_abc",
105-
want: "https://code0.ai/v1/video/generations/task_abc",
105+
want: "https://video.example.test/v1/video/generations/task_abc",
106106
},
107107
{
108108
name: "canonical v1 video poll path unchanged",
109-
baseURL: "https://code0.ai",
109+
baseURL: "https://video.example.test",
110110
requestPath: "/v1/video/generations/task_abc",
111-
want: "https://code0.ai/v1/video/generations/task_abc",
111+
want: "https://video.example.test/v1/video/generations/task_abc",
112+
},
113+
{
114+
name: "root-style videos submit path gains v1",
115+
baseURL: "https://video.example.test",
116+
requestPath: "/videos",
117+
want: "https://video.example.test/v1/videos",
118+
},
119+
{
120+
name: "root-style videos poll path gains v1",
121+
baseURL: "https://video.example.test",
122+
requestPath: "/videos/task_abc",
123+
want: "https://video.example.test/v1/videos/task_abc",
124+
},
125+
{
126+
name: "canonical v1 videos poll path unchanged",
127+
baseURL: "https://video.example.test",
128+
requestPath: "/v1/videos/task_abc",
129+
want: "https://video.example.test/v1/videos/task_abc",
112130
},
113131
}
114132

‎internal/adapter/provider/custom/adapter_video_test.go‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -24,19 +24,33 @@ func TestCustomAdapterExecuteVideoSubmitAndPoll(t *testing.T) {
2424
wantPath string
2525
}{
2626
{
27-
name: "submit",
27+
name: "legacy submit",
2828
method: http.MethodPost,
2929
requestURI: "/v1/video/generations",
30-
body: []byte(`{"model":"doubao-seedance-2-0-260128","prompt":"a cat runs"}`),
30+
body: []byte(`{"model":"video-test-model","prompt":"a cat runs"}`),
3131
wantPath: "/v1/video/generations",
3232
},
3333
{
34-
name: "poll",
34+
name: "legacy poll",
3535
method: http.MethodGet,
3636
requestURI: "/v1/video/generations/task_abc123",
3737
body: nil,
3838
wantPath: "/v1/video/generations/task_abc123",
3939
},
40+
{
41+
name: "videos submit",
42+
method: http.MethodPost,
43+
requestURI: "/v1/videos",
44+
body: []byte(`{"model":"video-test-model","prompt":"a cat runs"}`),
45+
wantPath: "/v1/videos",
46+
},
47+
{
48+
name: "videos poll",
49+
method: http.MethodGet,
50+
requestURI: "/v1/videos/task_abc123",
51+
body: nil,
52+
wantPath: "/v1/videos/task_abc123",
53+
},
4054
}
4155

4256
for _, tc := range cases {
@@ -55,13 +69,13 @@ func TestCustomAdapterExecuteVideoSubmitAndPoll(t *testing.T) {
5569
defer server.Close()
5670

5771
adapter, err := NewAdapter(&domain.Provider{
58-
Name: "code0-seedance",
72+
Name: "video-provider",
5973
Type: "custom",
6074
SupportedClientTypes: []domain.ClientType{domain.ClientTypeVideo},
6175
Config: &domain.ProviderConfig{
6276
Custom: &domain.ProviderConfigCustom{
6377
BaseURL: server.URL,
64-
APIKey: "sk-seedance",
78+
APIKey: "sk-video-provider",
6579
},
6680
},
6781
})
@@ -95,8 +109,8 @@ func TestCustomAdapterExecuteVideoSubmitAndPoll(t *testing.T) {
95109
if gotPath != tc.wantPath {
96110
t.Fatalf("upstream path = %q, want %q", gotPath, tc.wantPath)
97111
}
98-
if gotAuth != "Bearer sk-seedance" {
99-
t.Fatalf("upstream Authorization = %q, want %q", gotAuth, "Bearer sk-seedance")
112+
if gotAuth != "Bearer sk-video-provider" {
113+
t.Fatalf("upstream Authorization = %q, want %q", gotAuth, "Bearer sk-video-provider")
100114
}
101115
// The client's token must not leak through any alternate auth header.
102116
if gotAPIKey != "" || gotGoog != "" || gotProxyAuth != "" {

‎internal/core/proxy_routes.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,10 @@ func RegisterProxyRoutes(mux *http.ServeMux, handlers ProxyRouteHandlers) {
4848
mux.Handle("/v1/video/generations/", handlers.ProxyHandler)
4949
mux.Handle("/video/generations", handlers.ProxyHandler)
5050
mux.Handle("/video/generations/", handlers.ProxyHandler)
51+
mux.Handle("/v1/videos", handlers.ProxyHandler)
52+
mux.Handle("/v1/videos/", handlers.ProxyHandler)
53+
mux.Handle("/videos", handlers.ProxyHandler)
54+
mux.Handle("/videos/", handlers.ProxyHandler)
5155
// Codex API
5256
mux.Handle("/responses", responsesHandler)
5357
mux.Handle("/responses/", responsesHandler)

‎internal/core/proxy_routes_test.go‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,37 @@ func TestRegisterProxyRoutes_RoutesGeminiGenerationToProxy(t *testing.T) {
8888
}
8989
}
9090

91+
func TestRegisterProxyRoutes_RoutesVideosToProxy(t *testing.T) {
92+
mux := http.NewServeMux()
93+
calledPaths := make([]string, 0, 2)
94+
95+
RegisterProxyRoutes(mux, ProxyRouteHandlers{
96+
ProxyHandler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
97+
calledPaths = append(calledPaths, r.URL.Path)
98+
w.WriteHeader(http.StatusNoContent)
99+
}),
100+
})
101+
102+
for _, path := range []string{"/v1/videos", "/v1/videos/task_abc123"} {
103+
rec := httptest.NewRecorder()
104+
mux.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, path, nil))
105+
if rec.Code != http.StatusNoContent {
106+
t.Fatalf("%s status = %d, want %d", path, rec.Code, http.StatusNoContent)
107+
}
108+
}
109+
wantPaths := map[string]bool{"/v1/videos": false, "/v1/videos/task_abc123": false}
110+
for _, path := range calledPaths {
111+
if _, ok := wantPaths[path]; ok {
112+
wantPaths[path] = true
113+
}
114+
}
115+
for path, called := range wantPaths {
116+
if !called {
117+
t.Fatalf("proxy handler calls = %v, missing %s", calledPaths, path)
118+
}
119+
}
120+
}
121+
91122
func TestRegisterProxyRoutes_GeminiGenerationEnabledByDefault(t *testing.T) {
92123
mux := http.NewServeMux()
93124
proxyCalled := false

‎internal/handler/provider_proxy.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99

1010
maxxctx "github.com/awsl-project/maxx/internal/context"
1111
"github.com/awsl-project/maxx/internal/domain"
12+
"github.com/awsl-project/maxx/internal/executor"
1213
"github.com/awsl-project/maxx/internal/flow"
1314
"github.com/awsl-project/maxx/internal/repository"
1415
"github.com/awsl-project/maxx/internal/requestmeta"
@@ -86,7 +87,7 @@ func (h *ProviderProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
8687
log.Printf("[ProviderProxy] Direct forwarding through provider: %s (ID: %d)", provider.Name, provider.ID)
8788
r.URL.Path = apiPath
8889

89-
ctx := flow.NewCtx(w, r)
90+
ctx := flow.NewCtx(executor.NewResponseCapture(w), r)
9091
handlers := append([]flow.HandlerFunc{}, h.proxyHandler.extra...)
9192
handlers = append(handlers, h.directDispatch(provider))
9293
h.proxyHandler.engine.HandleWith(ctx, handlers...)
@@ -145,7 +146,7 @@ func (h *ProviderProxyHandler) directDispatch(provider *domain.Provider) flow.Ha
145146

146147
if proxyErr, ok := asHandlerProxyError(err); ok {
147148
if isStream {
148-
writeStreamError(c.Writer, proxyErr)
149+
writeProxyStreamError(c.Writer, proxyErr)
149150
} else {
150151
writeProxyError(c.Writer, proxyErr)
151152
}

‎internal/handler/proxy.go‎

Lines changed: 43 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -158,7 +158,7 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
158158
return
159159
}
160160

161-
ctx := flow.NewCtx(w, r)
161+
ctx := flow.NewCtx(executor.NewResponseCapture(w), r)
162162
h.engine.HandleWith(ctx, h.proxyHandlers()...)
163163
}
164164

@@ -354,11 +354,7 @@ func (h *ProxyHandler) ingress(c *flow.Ctx) {
354354
if err := h.tokenAuth.AcquireConcurrency(apiToken); err != nil {
355355
log.Printf("[Proxy] Token concurrency limit hit: tokenID=%d err=%v", apiToken.ID, err)
356356
h.executor.RecordRejectedProxyRequest(c, apiToken, http.StatusTooManyRequests, err.Error())
357-
if stream {
358-
writeStreamRateLimitError(w, err.Error(), 1)
359-
} else {
360-
writeRateLimitError(w, err.Error(), 1)
361-
}
357+
writeRateLimitError(w, err.Error(), 1)
362358
c.Abort()
363359
return
364360
}
@@ -438,7 +434,7 @@ func (h *ProxyHandler) dispatch(c *flow.Ctx) {
438434
proxyErr, ok := asHandlerProxyError(err)
439435
if ok {
440436
if stream {
441-
writeStreamError(c.Writer, proxyErr)
437+
writeProxyStreamError(c.Writer, proxyErr)
442438
} else {
443439
writeProxyError(c.Writer, proxyErr)
444440
}
@@ -630,6 +626,46 @@ func writeStreamError(w http.ResponseWriter, err *domain.ProxyError) {
630626
}
631627
}
632628

629+
func writeProxyStreamError(w http.ResponseWriter, err *domain.ProxyError) {
630+
if responseStarted(w) {
631+
writeStreamErrorEvent(w, err)
632+
return
633+
}
634+
writeProxyError(w, err)
635+
}
636+
637+
type responseStartTracker interface {
638+
WroteToClient() bool
639+
}
640+
641+
func responseStarted(w http.ResponseWriter) bool {
642+
tracker, ok := w.(responseStartTracker)
643+
return ok && tracker.WroteToClient()
644+
}
645+
646+
func writeStreamErrorEvent(w http.ResponseWriter, err *domain.ProxyError) {
647+
payload := map[string]interface{}{
648+
"message": err.Error(),
649+
"type": "upstream_error",
650+
"retryable": err.Retryable,
651+
}
652+
if err.Code != "" {
653+
payload["code"] = err.Code
654+
}
655+
errorEvent := map[string]interface{}{
656+
"type": "error",
657+
"error": payload,
658+
}
659+
data, _ := json.Marshal(errorEvent)
660+
w.Write([]byte("data: "))
661+
w.Write(data)
662+
w.Write([]byte("\n\n"))
663+
664+
if f, ok := w.(http.Flusher); ok {
665+
f.Flush()
666+
}
667+
}
668+
633669
func isClaudeCountTokensRequest(req *http.Request, clientType domain.ClientType) bool {
634670
return clientType == domain.ClientTypeClaude && req != nil && req.URL.Path == "/v1/messages/count_tokens"
635671
}

‎internal/handler/proxy_api_paths.go‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,8 @@ var proxyAPIEndpoints = []proxyAPIEndpoint{
3939
{path: "/v1/images/generations", subtree: false}, // OpenAI Images API
4040
{path: "/v1/images/edits", subtree: false}, // OpenAI Images API
4141
{path: "/v1/images", subtree: false}, // OpenRouter unified image API
42+
{path: "/v1/video/generations", subtree: true}, // Async video generation
43+
{path: "/v1/videos", subtree: true}, // OpenAI-compatible video API
4244
{path: "/responses", subtree: true}, // Codex API
4345
{path: "/v1/responses", subtree: true}, // Codex API
4446
{path: "/v1/models", subtree: true}, // Model list API
@@ -93,3 +95,25 @@ func proxyRouteExposureEnabled(settings repository.SystemSettingRepository, path
9395
func proxyRouteExposureEnabledByDefault(key string) bool {
9496
return true
9597
}
98+
99+
func isStaticAPIPath(path string) bool {
100+
if path == "/api" || strings.HasPrefix(path, "/api/") ||
101+
path == "/provider" || strings.HasPrefix(path, "/provider/") ||
102+
path == "/responses" || strings.HasPrefix(path, "/responses/") ||
103+
path == "/chat" || strings.HasPrefix(path, "/chat/") ||
104+
path == "/images" || strings.HasPrefix(path, "/images/") ||
105+
path == "/video" || strings.HasPrefix(path, "/video/") ||
106+
path == "/videos" || strings.HasPrefix(path, "/videos/") ||
107+
path == "/models" || strings.HasPrefix(path, "/models/") ||
108+
path == "/v1" || strings.HasPrefix(path, "/v1/") ||
109+
path == "/v1beta" || strings.HasPrefix(path, "/v1beta/") ||
110+
path == "/v1internal" || strings.HasPrefix(path, "/v1internal/") {
111+
return true
112+
}
113+
for _, e := range proxyAPIEndpoints {
114+
if path == e.path || strings.HasPrefix(path, e.path+"/") {
115+
return true
116+
}
117+
}
118+
return false
119+
}

0 commit comments

Comments
 (0)