diff --git a/backend/internal/app/protocol_runtime_test.go b/backend/internal/app/protocol_runtime_test.go index 6e823a283..d21fd9bde 100644 --- a/backend/internal/app/protocol_runtime_test.go +++ b/backend/internal/app/protocol_runtime_test.go @@ -395,6 +395,52 @@ func TestDeclarativeProtocolPollRecoversFromTransientGatewayFailure(t *testing.T } } +func TestDeclarativeProtocolRetriesResultDownloadWithoutRepolling(t *testing.T) { + adapter, err := protocol.LoadManifest([]byte(`{ + "apiVersion":"yingce.plugin/v1", + "id":"test-declarative-download-retry","version":"1.0.0","name":"Test Declarative Download Retry","author":"Test","documentation":"# Test", + "contributes":{"providers":[{"id":"test-declarative-download-retry","label":"Test Declarative Download Retry","capabilities":["video"],"scopes":["canvas"],"create":{"method":"POST","path":"/tasks","fields":{"model":"request.model"}},"poll":{"method":"GET","path":"/tasks/{{taskId}}"},"response":{"taskIdPaths":["id"],"statusPaths":["status"],"resultPaths":["video_url"],"resultKind":"video"}}]} + }`)) + if err != nil { + t.Fatal(err) + } + pollCalls := 0 + downloadCalls := 0 + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/tasks": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"provider-task-1","status":"pending"}`)) + case "/v1/tasks/provider-task-1": + pollCalls++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"provider-task-1","status":"succeeded","video_url":"` + server.URL + `/media.mp4"}`)) + case "/media.mp4": + downloadCalls++ + if downloadCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "video-model", InterfaceType: "test-declarative-download-retry", AllowLocalChannel: true} + ctx := withProviderOutboundPolicy(context.Background(), config) + result, err := runProtocolAdapterTaskWithPolicy(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, fastVideoPollPolicy()) + if err != nil { + t.Fatal(err) + } + if result["mode"] != "video" || pollCalls != 1 || downloadCalls != 2 { + t.Fatalf("result = %#v, poll calls = %d, download calls = %d", result, pollCalls, downloadCalls) + } +} + func TestDeclarativeProtocolRuntimeGeneratesPerCreateIdempotencyKey(t *testing.T) { manifest := []byte(`{ "apiVersion":"yingce.plugin/v1", diff --git a/backend/internal/app/provider_protocol.go b/backend/internal/app/provider_protocol.go index bf4914092..ab48f51a5 100644 --- a/backend/internal/app/provider_protocol.go +++ b/backend/internal/app/provider_protocol.go @@ -77,7 +77,7 @@ func runProtocolAdapterTaskWithPolicy(ctx context.Context, input canvasGeneratio return nil, protocolResultError(created.Message, taskID) } if created.Status == protocol.StatusSucceeded { - return finishProtocolAdapterResult(ctx, input, adapter, request, taskID, created.Result) + return finishProtocolAdapterResult(ctx, input, adapter, request, taskID, created.Result, policy) } if taskID == "" { return nil, errors.New("声明式协议创建请求没有返回任务 ID") @@ -102,7 +102,7 @@ func runProtocolAdapterTaskWithPolicy(ctx context.Context, input canvasGeneratio } switch state.Status { case protocol.StatusSucceeded: - result, err := finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result) + result, err := finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result, policy) return videoPollOutcome{Done: err == nil, Result: result}, err case protocol.StatusFailed, protocol.StatusCancelled: return videoPollOutcome{}, protocolResultError(state.Message, taskID) @@ -152,7 +152,7 @@ func queryProtocolAdapterVideoTask(ctx context.Context, input canvasGenerationIn providerStatus := string(state.Status) switch state.Status { case protocol.StatusSucceeded: - result, err := finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result) + result, err := finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result, defaultVideoPollPolicy()) return result, providerStatus, err case protocol.StatusFailed, protocol.StatusCancelled: return nil, providerStatus, protocolResultError(state.Message, taskID) @@ -731,7 +731,7 @@ func appendProtocolQuery(rawURL string, values map[string][]string) (string, err return parsed.String(), nil } -func finishProtocolResult(ctx context.Context, config providerConfig, mode string, result *protocol.Result) (map[string]interface{}, error) { +func finishProtocolResult(ctx context.Context, config providerConfig, mode string, taskID string, result *protocol.Result, pollPolicy videoPollPolicy) (map[string]interface{}, error) { if result == nil { return nil, errors.New("声明式协议已完成但没有返回结果") } @@ -758,7 +758,16 @@ func finishProtocolResult(ctx context.Context, config providerConfig, mode strin } items := make([]interface{}, 0, len(references)) for _, reference := range references { - data, mimeType, err := protocolMediaBytes(ctx, config, reference) + var data []byte + var mimeType string + var err error + if mode == "video" { + data, mimeType, err = runVideoDownload(ctx, taskID, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return protocolMediaBytesOnce(ctx, config, reference) + }) + } else { + data, mimeType, err = protocolMediaBytes(ctx, config, reference) + } if err != nil { return nil, err } @@ -778,20 +787,29 @@ func finishProtocolResult(ctx context.Context, config providerConfig, mode strin // finishProtocolAdapterResult 优先消费 create/poll 已返回的内联或 URL 结果，只有插件明确声明独立结果端点时才下载。 // 空响应或下载失败必须向上失败，不能生成伪素材；application/octet-stream 仅表示传输层未知类型， // 不会把未知内容伪装成具体图片、视频或音频 MIME。 -func finishProtocolAdapterResult(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, request protocol.GenerationRequest, taskID string, result *protocol.Result) (map[string]interface{}, error) { +func finishProtocolAdapterResult(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, request protocol.GenerationRequest, taskID string, result *protocol.Result, pollPolicy videoPollPolicy) (map[string]interface{}, error) { if protocolResultHasOutput(input.Mode, result) { - return finishProtocolResult(ctx, input.Config, input.Mode, result) + return finishProtocolResult(ctx, input.Config, input.Mode, taskID, result, pollPolicy) } resultAdapter, ok := adapter.(protocol.ResultAdapter) capability, hasCapability := adapter.(protocol.ResultCapability) if !ok || !hasCapability || !capability.ResultAvailable() { - return finishProtocolResult(ctx, input.Config, input.Mode, result) + return finishProtocolResult(ctx, input.Config, input.Mode, taskID, result, pollPolicy) } spec, err := resultAdapter.BuildResult(ctx, protocol.PollContext{BaseURL: input.Config.BaseURL, Model: request.Model, Request: request, TaskID: taskID}) if err != nil { return nil, err } - data, mimeType, err := executeProtocolBinaryRequest(withProviderRequestKind(ctx, "download"), input.Config, spec) + download := func(ctx context.Context) ([]byte, string, error) { + return executeProtocolBinaryRequest(withProviderRequestKind(ctx, "download"), input.Config, spec) + } + var data []byte + var mimeType string + if input.Mode == "video" { + data, mimeType, err = runVideoDownload(ctx, taskID, pollPolicy, download) + } else { + data, mimeType, err = download(ctx) + } if err != nil { return nil, fmt.Errorf("声明式协议结果下载失败：%w", err) } @@ -814,7 +832,7 @@ func finishProtocolAdapterResult(ctx context.Context, input canvasGenerationInpu default: return nil, fmt.Errorf("声明式协议结果下载不支持生成模式 %s", input.Mode) } - return finishProtocolResult(ctx, input.Config, input.Mode, downloaded) + return finishProtocolResult(ctx, input.Config, input.Mode, taskID, downloaded, pollPolicy) } func protocolResultHasOutput(mode string, result *protocol.Result) bool { @@ -836,21 +854,13 @@ func protocolResultHasOutput(mode string, result *protocol.Result) bool { } func protocolMediaBytes(ctx context.Context, config providerConfig, reference protocol.MediaReference) ([]byte, string, error) { - if strings.TrimSpace(reference.DataURL) != "" { - mimeType, data, err := decodeProviderDataURL(reference.DataURL) - return data, mimeType, err - } - value := strings.TrimSpace(reference.URL) - if value == "" { - return nil, "", errors.New("声明式协议媒体结果地址为空") - } var data []byte var mimeType string var err error for attempt := 0; attempt < 3; attempt++ { - data, mimeType, err = getProviderExternalBinary(withProviderRequestKind(ctx, "download"), config, value) + data, mimeType, err = protocolMediaBytesOnce(ctx, config, reference) if err == nil { - return data, normalizedMediaMimeType(mimeType, data), nil + return data, mimeType, nil } if attempt == 2 || !retryableProtocolMediaDownload(err) { break @@ -862,6 +872,19 @@ func protocolMediaBytes(ctx context.Context, config providerConfig, reference pr return nil, "", fmt.Errorf("声明式协议媒体结果下载失败：%w", err) } +func protocolMediaBytesOnce(ctx context.Context, config providerConfig, reference protocol.MediaReference) ([]byte, string, error) { + if strings.TrimSpace(reference.DataURL) != "" { + mimeType, data, err := decodeProviderDataURL(reference.DataURL) + return data, mimeType, err + } + value := strings.TrimSpace(reference.URL) + if value == "" { + return nil, "", errors.New("声明式协议媒体结果地址为空") + } + data, mimeType, err := getProviderExternalBinary(withProviderRequestKind(ctx, "download"), config, value) + return data, normalizedMediaMimeType(mimeType, data), err +} + func retryableProtocolMediaDownload(err error) bool { if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return false diff --git a/backend/internal/app/provider_test.go b/backend/internal/app/provider_test.go index 66a227f76..73b36e3a1 100644 --- a/backend/internal/app/provider_test.go +++ b/backend/internal/app/provider_test.go @@ -1625,6 +1625,45 @@ func TestRunVideoTaskRetriesTransientPollFailureWithoutRecreating(t *testing.T) } } +func TestRunVideoTaskRetriesDownloadWithoutRepolling(t *testing.T) { + t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true") + pollCalls := 0 + downloadCalls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method + " " + r.URL.Path { + case "POST /v1/videos": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"video-1","status":"queued"}`)) + case "GET /v1/videos/video-1": + pollCalls++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"video-1","status":"completed"}`)) + case "GET /v1/videos/video-1/content": + downloadCalls++ + if downloadCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + result, err := runVideoTaskWithPolicy(context.Background(), canvasGenerationInput{ + Prompt: "make it move", + Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", Model: "custom-video-v1"}, + }, fastVideoPollPolicy()) + if err != nil { + t.Fatal(err) + } + if result["mode"] != "video" || pollCalls != 1 || downloadCalls != 2 { + t.Fatalf("result = %#v, poll calls = %d, download calls = %d", result, pollCalls, downloadCalls) + } +} + func TestRunVideoTaskSendsOnlyDeclaredResolutionName(t *testing.T) { tests := []struct { name string diff --git a/backend/internal/app/provider_video.go b/backend/internal/app/provider_video.go index 7d73edc3f..51884a221 100644 --- a/backend/internal/app/provider_video.go +++ b/backend/internal/app/provider_video.go @@ -145,14 +145,18 @@ func runVideoTaskWithPolicy(ctx context.Context, input canvasGenerationInput, po status := strings.ToLower(stringField(state, "status")) if status == "completed" || status == "succeeded" || status == "success" || status == "done" { if videoURL := newAPIVideoResultURL(state); videoURL != "" { - data, mimeType, err := getProviderExternalBinary(withProviderRequestKind(ctx, "download"), input.Config, videoURL) + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getProviderExternalBinary(withProviderRequestKind(ctx, "download"), input.Config, videoURL) + }) if err != nil { return videoPollOutcome{}, fmt.Errorf("视频结果下载失败（任务 %s）：%w", id, err) } mimeType = normalizedMediaMimeType(mimeType, data) return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}}, nil } - data, mimeType, err := getBinary(ctx, input.Config, "/videos/"+id+"/content") + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getBinary(withProviderRequestKind(ctx, "download"), input.Config, "/videos/"+id+"/content") + }) if err != nil { return videoPollOutcome{}, err } @@ -258,13 +262,17 @@ func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput, pol if status == "completed" || status == "succeeded" { videoURL := stringField(state, "video_url") if videoURL != "" { - data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + }) if err != nil { return videoPollOutcome{}, fmt.Errorf("视频结果下载失败：%w", err) } return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}}, nil } - data, mimeType, err := getBinary(ctx, input.Config, "/videos/"+id+"/content") + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getBinary(withProviderRequestKind(ctx, "download"), input.Config, "/videos/"+id+"/content") + }) if err != nil { return videoPollOutcome{}, fmt.Errorf("Seedance 任务成功但未返回视频 URL，备用内容下载失败：%w", err) } @@ -343,7 +351,9 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn if videoURL == "" { return videoPollOutcome{}, fmt.Errorf("%s任务成功但没有返回视频 URL", providerName) } - data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + }) if err != nil { return videoPollOutcome{}, fmt.Errorf("视频结果下载失败：%w", err) } diff --git a/backend/internal/app/provider_video_polling.go b/backend/internal/app/provider_video_polling.go index 6f72c646f..e542bd859 100644 --- a/backend/internal/app/provider_video_polling.go +++ b/backend/internal/app/provider_video_polling.go @@ -15,6 +15,7 @@ const defaultVideoPollInterval = 30 * time.Second type videoPollPolicy struct { Interval time.Duration MaxNotFoundMisses int + MaxDownloadTries int Sleep func(context.Context, time.Duration) error } @@ -27,6 +28,7 @@ func defaultVideoPollPolicy() videoPollPolicy { return videoPollPolicy{ Interval: defaultVideoPollInterval, MaxNotFoundMisses: 3, + MaxDownloadTries: 3, Sleep: sleepContext, } } @@ -76,12 +78,36 @@ func normalizeVideoPollPolicy(policy videoPollPolicy) videoPollPolicy { if policy.MaxNotFoundMisses <= 0 { policy.MaxNotFoundMisses = 3 } + if policy.MaxDownloadTries <= 0 { + policy.MaxDownloadTries = 3 + } if policy.Sleep == nil { policy.Sleep = sleepContext } return policy } +func runVideoDownload(ctx context.Context, taskID string, policy videoPollPolicy, download func(context.Context) ([]byte, string, error)) ([]byte, string, error) { + policy = normalizeVideoPollPolicy(policy) + var lastErr error + for attempt := 1; attempt <= policy.MaxDownloadTries; attempt++ { + data, mimeType, err := download(ctx) + if err == nil { + return data, mimeType, nil + } + lastErr = err + retry, _ := retryableVideoPollError(ctx, err) + if !retry || attempt == policy.MaxDownloadTries { + return nil, "", err + } + delay := max(policy.Interval, providerRetryAfter(err)) + if err := policy.Sleep(ctx, delay); err != nil { + return nil, "", err + } + } + return nil, "", lastErr +} + func retryableVideoPollError(ctx context.Context, err error) (retry bool, notFound bool) { if err == nil || ctx.Err() != nil || errors.Is(err, context.Canceled) { return false, false diff --git a/backend/internal/app/provider_video_polling_test.go b/backend/internal/app/provider_video_polling_test.go index 1f778e41e..7027237fb 100644 --- a/backend/internal/app/provider_video_polling_test.go +++ b/backend/internal/app/provider_video_polling_test.go @@ -121,6 +121,41 @@ func TestVideoPollContextCancellationInterruptsInitialWait(t *testing.T) { } } +func TestVideoDownloadRetriesTransientFailure(t *testing.T) { + attempts := 0 + waits := 0 + policy := fastVideoPollPolicy() + policy.Sleep = func(context.Context, time.Duration) error { + waits++ + return nil + } + data, mimeType, err := runVideoDownload(context.Background(), "provider-task-1", policy, func(context.Context) ([]byte, string, error) { + attempts++ + if attempts == 1 { + return nil, "", providerHTTPError{StatusCode: http.StatusBadGateway, Status: "502 Bad Gateway"} + } + return []byte("video"), "video/mp4", nil + }) + if err != nil { + t.Fatal(err) + } + if string(data) != "video" || mimeType != "video/mp4" || attempts != 2 || waits != 1 { + t.Fatalf("data = %q, mime = %q, attempts = %d, waits = %d", data, mimeType, attempts, waits) + } +} + +func TestVideoDownloadStopsAfterThreeTransientFailures(t *testing.T) { + attempts := 0 + _, _, err := runVideoDownload(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) ([]byte, string, error) { + attempts++ + return nil, "", providerHTTPError{StatusCode: http.StatusServiceUnavailable, Status: "503 Service Unavailable"} + }) + var httpErr providerHTTPError + if !errors.As(err, &httpErr) || attempts != 3 { + t.Fatalf("error = %#v, attempts = %d, want three attempts and provider error", err, attempts) + } +} + func fastVideoPollPolicy() videoPollPolicy { policy := defaultVideoPollPolicy() policy.Interval = time.Millisecond