fix(video): 异步任务 - 完善全局轮询容错与调度策略

2026-09-12 21:28:41 +08:00
parent d2efd81b8e
commit 7c074e5291
14 changed files with 462 additions and 41 deletions
--- a/backend/internal/app/admin.go
+++ b/backend/internal/app/admin.go
@@ -664,10 +664,18 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error {
 		var nextPollAt *time.Time
 		if stage == "create" && log.Status == model.ApiCallStatusSucceeded && log.ProviderRequestID != "" {
 			stage = "accepted"
-			next := time.Now().Add(2 * time.Second)
+			delay := 2 * time.Second
+			if log.Capability == "video" {
+				delay = defaultVideoPollInterval
+			}
+			next := time.Now().Add(delay)
 			nextPollAt = &next
 		} else if stage == "poll" {
-			next := time.Now().Add(5 * time.Second)
+			delay := 5 * time.Second
+			if log.Capability == "video" {
+				delay = defaultVideoPollInterval
+			}
+			next := time.Now().Add(delay)
 			nextPollAt = &next
 		}
 		if err := s.repo.UpdateTaskProviderState(log.TaskID, log.ProviderRequestID, stage, nextPollAt); err != nil {
--- a/backend/internal/app/protocol_runtime_test.go
+++ b/backend/internal/app/protocol_runtime_test.go
@@ -13,6 +13,7 @@ import (
 	"regexp"
 	"strings"
 	"testing"
+	"time"

 	"infinite-canvas/backend/internal/protocol"
 )
@@ -348,6 +349,18 @@ func TestDeclarativeProtocolRuntimeExecutesCreatePollAndDownload(t *testing.T) {
 	}
 }

+func TestDeclarativeProtocolPollingPolicyOnlyChangesVideoBehavior(t *testing.T) {
+	video := declarativeProtocolPollPolicy("video")
+	if video.InitialDelay != defaultVideoPollInterval || !video.RetryTransient {
+		t.Fatalf("video polling policy = %#v, want delayed resilient polling", video)
+	}
+
+	image := declarativeProtocolPollPolicy("image")
+	if image.InitialDelay != 0 || image.Interval != 2500*time.Millisecond || image.RetryTransient {
+		t.Fatalf("image polling policy = %#v, want legacy immediate fail-fast polling", image)
+	}
+}
+
 func TestDeclarativeProtocolPollRecoversFromTransientGatewayFailure(t *testing.T) {
 	adapter, err := protocol.LoadManifest([]byte(`{
 		"apiVersion":"yingce.plugin/v1",
--- a/backend/internal/app/provider.go
+++ b/backend/internal/app/provider.go
@@ -100,7 +100,7 @@ type providerConfig struct {
 }

 const providerHTTPTimeout = 5 * time.Minute
-const videoPollTimeout = 30 * time.Minute
+const videoPollTimeout = time.Hour
 const maxProviderResponseBytes int64 = 64 << 20

 type providerMedia struct {
@@ -145,6 +145,19 @@ type providerHTTPError struct {
 	RetryAfter time.Duration
 }

+type providerResponseDecodeError struct {
+	Err error
+}
+
+func (e providerResponseDecodeError) Error() string { return e.Err.Error() }
+func (e providerResponseDecodeError) Unwrap() error { return e.Err }
+
+type providerCircuitOpenError struct{}
+
+func (providerCircuitOpenError) Error() string {
+	return "当前渠道连续失败，已暂时熔断，请稍后重试"
+}
+
 type providerStatePendingError struct {
 	TaskID string
 	Cause  error
--- a/backend/internal/app/provider_http_client.go
+++ b/backend/internal/app/provider_http_client.go
@@ -188,10 +188,10 @@ func doJSON(req *http.Request, target interface{}) error {
 		return err
 	}
 	if !strings.Contains(mimeType, "json") && !json.Valid(data) {
-		return fmt.Errorf("接口返回非 JSON 内容：%s", mimeType)
+		return providerResponseDecodeError{Err: fmt.Errorf("接口返回非 JSON 内容：%s", mimeType)}
 	}
 	if err := json.Unmarshal(data, target); err != nil {
-		return err
+		return providerResponseDecodeError{Err: err}
 	}
 	if payload, ok := target.(*imageResponse); ok {
 		if payload.Error != nil && payload.Error.Message != "" {
@@ -269,7 +269,7 @@ func doBinaryWithConsumer(req *http.Request, onChunk func(string, []byte)) ([]by
 			return nil, "", fmt.Errorf("读取渠道熔断状态失败：%w", err)
 		}
 		if open {
-			return nil, "", errors.New("当前渠道连续失败，已暂时熔断，请稍后重试")
+			return nil, "", providerCircuitOpenError{}
 		}
 		slotID := channelID
 		if slotID == "" {
--- a/backend/internal/app/provider_protocol.go
+++ b/backend/internal/app/provider_protocol.go
@@ -40,7 +40,19 @@ func runDeclarativeProtocolTask(ctx context.Context, input canvasGenerationInput
 }

 func runProtocolAdapterTask(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter) (map[string]interface{}, error) {
-	return runProtocolAdapterTaskWithPolicy(ctx, input, adapter, defaultVideoPollPolicy())
+	return runProtocolAdapterTaskWithPolicy(ctx, input, adapter, declarativeProtocolPollPolicy(input.Mode))
+}
+
+func declarativeProtocolPollPolicy(mode string) videoPollPolicy {
+	if mode == "video" {
+		return defaultVideoPollPolicy()
+	}
+	return videoPollPolicy{
+		Interval:          2500 * time.Millisecond,
+		MaxNotFoundMisses: 1,
+		MaxDownloadTries:  3,
+		Sleep:             sleepContext,
+	}
 }

 func runProtocolAdapterTaskWithPolicy(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, policy videoPollPolicy) (map[string]interface{}, error) {
--- a/backend/internal/app/provider_video_polling.go
+++ b/backend/internal/app/provider_video_polling.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"strings"
@@ -13,54 +15,111 @@ import (
 const defaultVideoPollInterval = 30 * time.Second

 type videoPollPolicy struct {
-	Interval          time.Duration
-	MaxNotFoundMisses int
-	MaxDownloadTries  int
-	Sleep             func(context.Context, time.Duration) error
+	InitialDelay          time.Duration
+	Interval              time.Duration
+	MaxNotFoundMisses     int
+	MaxMalformedResponses int
+	MaxDownloadTries      int
+	RetryTransient        bool
+	Sleep                 func(context.Context, time.Duration) error
+	Notify                func(context.Context, string, videoPollEvent, error)
 }

+type videoPollEvent string
+
+const (
+	videoPollEventRetrying  videoPollEvent = "retrying"
+	videoPollEventRecovered videoPollEvent = "recovered"
+)
+
 type videoPollOutcome struct {
 	Done   bool
 	Result map[string]interface{}
 }

+type videoDownloadError struct {
+	TaskID string
+	Cause  error
+}
+
+func (e videoDownloadError) Error() string {
+	if strings.TrimSpace(e.TaskID) == "" {
+		return fmt.Sprintf("视频结果下载失败：%v", e.Cause)
+	}
+	return fmt.Sprintf("视频结果下载失败（任务 %s）：%v", e.TaskID, e.Cause)
+}
+
+func (e videoDownloadError) Unwrap() error { return e.Cause }
+
 func defaultVideoPollPolicy() videoPollPolicy {
 	return videoPollPolicy{
-		Interval:          defaultVideoPollInterval,
-		MaxNotFoundMisses: 3,
-		MaxDownloadTries:  3,
-		Sleep:             sleepContext,
+		InitialDelay:          defaultVideoPollInterval,
+		Interval:              defaultVideoPollInterval,
+		MaxNotFoundMisses:     3,
+		MaxMalformedResponses: 3,
+		MaxDownloadTries:      3,
+		RetryTransient:        true,
+		Sleep:                 sleepContext,
+		Notify:                notifyTaskVideoPollEvent,
 	}
 }

 func runVideoPollLoop(ctx context.Context, taskID string, policy videoPollPolicy, query func(context.Context) (videoPollOutcome, error)) (map[string]interface{}, error) {
 	policy = normalizeVideoPollPolicy(policy)
 	deadline := providerPollingDeadline(ctx)
-	nextDelay := policy.Interval
+	pollContext, cancel := context.WithDeadline(ctx, deadline)
+	defer cancel()
+	ctx = pollContext
+	nextDelay := policy.InitialDelay
 	notFoundMisses := 0
+	malformedResponses := 0
+	retrying := false
 	for time.Now().Before(deadline) {
-		if err := policy.Sleep(ctx, nextDelay); err != nil {
-			return nil, err
+		if nextDelay > 0 {
+			if err := policy.Sleep(ctx, nextDelay); err != nil {
+				return nil, err
+			}
 		}
 		outcome, err := query(ctx)
 		if err != nil {
+			if !policy.RetryTransient {
+				return nil, err
+			}
 			retry, notFound := retryableVideoPollError(ctx, err)
 			if !retry {
 				return nil, err
 			}
-			if notFound {
+			malformed := isTransientResponseDecodeError(err)
+			if malformed {
+				malformedResponses++
+				notFoundMisses = 0
+				if malformedResponses >= policy.MaxMalformedResponses {
+					return nil, err
+				}
+			} else if notFound {
+				malformedResponses = 0
 				notFoundMisses++
 				if notFoundMisses >= policy.MaxNotFoundMisses {
 					return nil, err
 				}
 			} else {
+				malformedResponses = 0
 				notFoundMisses = 0
 			}
+			if !retrying {
+				retrying = true
+				policy.Notify(ctx, taskID, videoPollEventRetrying, err)
+			}
 			nextDelay = max(policy.Interval, providerRetryAfter(err))
 			continue
 		}
 		notFoundMisses = 0
+		malformedResponses = 0
 		nextDelay = policy.Interval
+		if retrying {
+			retrying = false
+			policy.Notify(ctx, taskID, videoPollEventRecovered, nil)
+		}
 		if outcome.Done {
 			return outcome.Result, nil
 		}
@@ -78,12 +137,18 @@ func normalizeVideoPollPolicy(policy videoPollPolicy) videoPollPolicy {
 	if policy.MaxNotFoundMisses <= 0 {
 		policy.MaxNotFoundMisses = 3
 	}
+	if policy.MaxMalformedResponses <= 0 {
+		policy.MaxMalformedResponses = 3
+	}
 	if policy.MaxDownloadTries <= 0 {
 		policy.MaxDownloadTries = 3
 	}
 	if policy.Sleep == nil {
 		policy.Sleep = sleepContext
 	}
+	if policy.Notify == nil {
+		policy.Notify = func(context.Context, string, videoPollEvent, error) {}
+	}
 	return policy
 }

@@ -98,23 +163,37 @@ func runVideoDownload(ctx context.Context, taskID string, policy videoPollPolicy
 		lastErr = err
 		retry, _ := retryableVideoPollError(ctx, err)
 		if !retry || attempt == policy.MaxDownloadTries {
-			return nil, "", err
+			return nil, "", videoDownloadError{TaskID: taskID, Cause: err}
 		}
 		delay := max(policy.Interval, providerRetryAfter(err))
 		if err := policy.Sleep(ctx, delay); err != nil {
 			return nil, "", err
 		}
 	}
-	return nil, "", lastErr
+	return nil, "", videoDownloadError{TaskID: taskID, Cause: 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
 	}
+	var downloadError videoDownloadError
+	if errors.As(err, &downloadError) {
+		return false, false
+	}
 	if code, _ := ChannelSlotFailureDetails(err); code != "" {
 		return true, false
 	}
+	var circuitOpen providerCircuitOpenError
+	if errors.As(err, &circuitOpen) {
+		return true, false
+	}
+	if isTransientResponseDecodeError(err) {
+		return true, false
+	}
+	if errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.ErrClosedPipe) {
+		return true, false
+	}
 	if errors.Is(err, context.DeadlineExceeded) {
 		return true, false
 	}
@@ -140,6 +219,33 @@ func retryableVideoPollError(ctx context.Context, err error) (retry bool, notFou
 	return false, false
 }

+func isTransientResponseDecodeError(err error) bool {
+	var decodeError providerResponseDecodeError
+	if errors.As(err, &decodeError) {
+		return true
+	}
+	var syntaxError *json.SyntaxError
+	return errors.As(err, &syntaxError)
+}
+
+func notifyTaskVideoPollEvent(ctx context.Context, _ string, event videoPollEvent, eventErr error) {
+	metadata, ok := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext)
+	if !ok || metadata.Service == nil || metadata.Service.repo == nil || metadata.TaskID == "" {
+		return
+	}
+	message := "上游视频查询暂时异常，将继续轮询原任务"
+	level := "warn"
+	payload := ""
+	if eventErr != nil {
+		payload = eventErr.Error()
+	}
+	if event == videoPollEventRecovered {
+		message = "上游视频查询已恢复"
+		level = "info"
+	}
+	_ = metadata.Service.log(metadata.UserID, metadata.TaskID, level, message, payload)
+}
+
 func isProviderTaskNotReadyError(httpErr providerHTTPError) bool {
 	if httpErr.StatusCode != http.StatusBadRequest && httpErr.StatusCode != http.StatusNotFound {
 		return false
--- a/backend/internal/app/provider_video_polling_test.go
+++ b/backend/internal/app/provider_video_polling_test.go
@@ -2,7 +2,9 @@ package app

 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
 	"net/http"
 	"reflect"
 	"testing"
@@ -33,6 +35,7 @@ func TestVideoPollRecoversFromTransientHTTPError(t *testing.T) {
 func TestVideoPollWaitsBeforeFirstQueryAndUsesRetryAfter(t *testing.T) {
 	var waits []time.Duration
 	policy := fastVideoPollPolicy()
+	policy.InitialDelay = 10 * time.Millisecond
 	policy.Interval = 10 * time.Millisecond
 	policy.Sleep = func(_ context.Context, delay time.Duration) error {
 		waits = append(waits, delay)
@@ -105,6 +108,68 @@ func TestVideoPollSuccessfulPendingResponseResetsNotFoundCounter(t *testing.T) {
 	}
 }

+func TestVideoPollRetriesMalformedResponseTwiceThenSucceeds(t *testing.T) {
+	attempts := 0
+	result, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) {
+		attempts++
+		if attempts <= 2 {
+			return videoPollOutcome{}, &json.SyntaxError{Offset: int64(attempts)}
+		}
+		return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil
+	})
+	if err != nil || attempts != 3 || result["mode"] != "video" {
+		t.Fatalf("result = %#v, error = %v, attempts = %d", result, err, attempts)
+	}
+}
+
+func TestVideoPollStopsAfterThreeMalformedResponses(t *testing.T) {
+	attempts := 0
+	_, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) {
+		attempts++
+		return videoPollOutcome{}, &json.SyntaxError{Offset: int64(attempts)}
+	})
+	var syntaxError *json.SyntaxError
+	if !errors.As(err, &syntaxError) || attempts != 3 {
+		t.Fatalf("error = %#v, attempts = %d, want three attempts and syntax error", err, attempts)
+	}
+}
+
+func TestVideoPollTreatsCircuitAndInterruptedReadsAsTransient(t *testing.T) {
+	for _, err := range []error{providerCircuitOpenError{}, io.ErrUnexpectedEOF, io.ErrClosedPipe} {
+		retry, notFound := retryableVideoPollError(context.Background(), err)
+		if !retry || notFound {
+			t.Fatalf("error %T classified as retry=%v notFound=%v", err, retry, notFound)
+		}
+	}
+}
+
+func TestVideoPollNotifiesOnlyWhenRetryStartsAndRecovers(t *testing.T) {
+	var events []videoPollEvent
+	policy := fastVideoPollPolicy()
+	policy.Notify = func(_ context.Context, _ string, event videoPollEvent, _ error) {
+		events = append(events, event)
+	}
+	attempts := 0
+	_, err := runVideoPollLoop(context.Background(), "provider-task-1", policy, func(context.Context) (videoPollOutcome, error) {
+		attempts++
+		switch attempts {
+		case 1, 2:
+			return videoPollOutcome{}, providerHTTPError{StatusCode: http.StatusBadGateway}
+		case 3:
+			return videoPollOutcome{}, nil
+		default:
+			return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil
+		}
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := []videoPollEvent{videoPollEventRetrying, videoPollEventRecovered}
+	if len(events) != len(want) || events[0] != want[0] || events[1] != want[1] {
+		t.Fatalf("events = %#v, want %#v", events, want)
+	}
+}
+
 func TestVideoPollContextCancellationInterruptsInitialWait(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
@@ -121,6 +186,26 @@ func TestVideoPollContextCancellationInterruptsInitialWait(t *testing.T) {
 	}
 }

+func TestVideoPollAppliesFallbackDeadlineToWaitAndQuery(t *testing.T) {
+	policy := fastVideoPollPolicy()
+	waitHadDeadline := false
+	queryHadDeadline := false
+	policy.Sleep = func(ctx context.Context, _ time.Duration) error {
+		_, waitHadDeadline = ctx.Deadline()
+		return nil
+	}
+	_, err := runVideoPollLoop(context.Background(), "provider-task-1", policy, func(ctx context.Context) (videoPollOutcome, error) {
+		_, queryHadDeadline = ctx.Deadline()
+		return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !waitHadDeadline || !queryHadDeadline {
+		t.Fatalf("wait deadline = %v, query deadline = %v", waitHadDeadline, queryHadDeadline)
+	}
+}
+
 func TestVideoDownloadRetriesTransientFailure(t *testing.T) {
 	attempts := 0
 	waits := 0
@@ -156,8 +241,21 @@ func TestVideoDownloadStopsAfterThreeTransientFailures(t *testing.T) {
 	}
 }

+func TestVideoPollDoesNotRetryExhaustedDownload(t *testing.T) {
+	attempts := 0
+	_, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) {
+		attempts++
+		return videoPollOutcome{}, videoDownloadError{TaskID: "provider-task-1", Cause: providerHTTPError{StatusCode: http.StatusBadGateway}}
+	})
+	var downloadError videoDownloadError
+	if !errors.As(err, &downloadError) || attempts != 1 {
+		t.Fatalf("error = %#v, poll attempts = %d, want one exhausted download", err, attempts)
+	}
+}
+
 func fastVideoPollPolicy() videoPollPolicy {
 	policy := defaultVideoPollPolicy()
+	policy.InitialDelay = time.Millisecond
 	policy.Interval = time.Millisecond
 	policy.Sleep = func(context.Context, time.Duration) error { return nil }
 	return policy
--- a/backend/internal/app/runtime_policy_test.go
+++ b/backend/internal/app/runtime_policy_test.go
@@ -26,6 +26,25 @@ func TestVideoTaskTimeoutHasFiveMinuteSafetyFloor(t *testing.T) {
 	}
 }

+func TestDefaultVideoTaskTimeoutIsOneHour(t *testing.T) {
+	policy := defaultRuntimePolicy().Task
+	if policy.VideoTimeoutMinutes != 60 {
+		t.Fatalf("default video timeout = %d minutes, want 60", policy.VideoTimeoutMinutes)
+	}
+	if got := taskExecutionTimeoutWithPolicy("canvas_video", policy); got != time.Hour {
+		t.Fatalf("video execution timeout = %s, want 1h", got)
+	}
+}
+
+func TestProviderPollingDeadlineDefaultsToOneHour(t *testing.T) {
+	startedAt := time.Now()
+	deadline := providerPollingDeadline(context.Background())
+	remaining := deadline.Sub(startedAt)
+	if remaining < time.Hour || remaining > time.Hour+time.Second {
+		t.Fatalf("provider polling deadline = %s, want approximately 1h", remaining)
+	}
+}
+
 func TestOnlyResumableNewAPIChannel2VideoDeadlinesStayRunning(t *testing.T) {
 	svc := &Service{}
 	input, err := json.Marshal(canvasGenerationInput{Mode: "video", Config: providerConfig{BaseURL: "https://example.com", InterfaceType: string(model.ChannelInterfaceNewAPIChannel2)}})
--- a/backend/internal/app/task_api_call_log_test.go
+++ b/backend/internal/app/task_api_call_log_test.go
@@ -4,6 +4,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"

 	"infinite-canvas/backend/internal/model"
 	"infinite-canvas/backend/internal/repository"
@@ -12,6 +13,43 @@ import (
 	"gorm.io/gorm"
 )

+func TestVideoPollLogSchedulesNextCheckThirtySecondsLater(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:provider-poll-schedule?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&model.Task{}, &model.ApiCallLog{}, &model.ModelPricing{}); err != nil {
+		t.Fatal(err)
+	}
+	task := model.Task{ID: "task-1", UserID: "user-1", Type: "canvas_video", Status: model.TaskStatusRunning}
+	if err := db.Create(&task).Error; err != nil {
+		t.Fatal(err)
+	}
+	root := model.ApiCallLog{ID: "api-log-1", UserID: task.UserID, TaskID: task.ID, Capability: "video", RequestKind: "create", ProviderRequestID: "provider-task-1", CreatedAt: time.Now()}
+	if err := db.Create(&root).Error; err != nil {
+		t.Fatal(err)
+	}
+	service := &Service{repo: repository.New(db)}
+	startedAt := time.Now()
+	if err := service.LogAPICall(model.ApiCallLog{
+		ID: "poll-log-1", UserID: task.UserID, TaskID: task.ID, Capability: "video", RequestKind: "poll",
+		ProviderRequestID: "provider-task-1", Status: model.ApiCallStatusSucceeded, StatusCode: 200, CreatedAt: startedAt,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var stored model.Task
+	if err := db.First(&stored, "id = ?", task.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if stored.NextPollAt == nil {
+		t.Fatal("next poll time is nil")
+	}
+	delay := stored.NextPollAt.Sub(startedAt)
+	if delay < 30*time.Second || delay > 31*time.Second {
+		t.Fatalf("next poll delay = %s, want approximately 30s", delay)
+	}
+}
+
 func TestEnsureFailedProviderAttemptLoggedFillsPreflightGapOnce(t *testing.T) {
 	db, err := gorm.Open(sqlite.Open("file:provider-preflight-log?mode=memory&cache=shared"), &gorm.Config{})
 	if err != nil {
--- a/backend/internal/app/workflow_provider.go
+++ b/backend/internal/app/workflow_provider.go
@@ -451,7 +451,7 @@ func (s *Service) runRunningHubWorkflow(ctx context.Context, input canvasGenerat
 	root := runningHubRootURL(input.Config.BaseURL)
 	apiKey := runningHubAPIKey(input.Config)
 	if resumed := resumedProviderRequestID(ctx); resumed != "" {
-		return s.pollRunningHubWorkflow(ctx, input.Config, root, resumed)
+		return s.pollRunningHubWorkflow(ctx, input.Config, root, resumed, input.Mode)
 	}
 	workflowID := strings.TrimSpace(input.Config.WorkflowID)
 	webappID := strings.TrimSpace(input.Config.WebappID)
@@ -535,7 +535,7 @@ func (s *Service) runRunningHubWorkflow(ctx context.Context, input canvasGenerat
 		metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext)
 		_ = s.log(metadata.UserID, metadata.TaskID, "error", "RunningHub 请求状态保存失败", taskID+"："+err.Error())
 	}
-	return s.pollRunningHubWorkflow(ctx, input.Config, root, taskID)
+	return s.pollRunningHubWorkflow(ctx, input.Config, root, taskID, input.Mode)
 }

 func runningHubWorkflowFailureMessage(response map[string]any) string {
@@ -1717,7 +1717,59 @@ func (s *Service) runningHubJSON(ctx context.Context, config providerConfig, end
 	return nil
 }

-func (s *Service) pollRunningHubWorkflow(ctx context.Context, config providerConfig, root string, taskID string) (map[string]interface{}, error) {
+func (s *Service) pollRunningHubWorkflow(ctx context.Context, config providerConfig, root string, taskID string, mode string) (map[string]interface{}, error) {
+	if mode == "video" {
+		return s.pollRunningHubVideoWorkflowWithPolicy(ctx, config, root, taskID, defaultVideoPollPolicy())
+	}
+	return s.pollRunningHubWorkflowLegacy(ctx, config, root, taskID)
+}
+
+func (s *Service) pollRunningHubVideoWorkflowWithPolicy(ctx context.Context, config providerConfig, root string, taskID string, policy videoPollPolicy) (map[string]interface{}, error) {
+	return runVideoPollLoop(ctx, taskID, policy, func(ctx context.Context) (videoPollOutcome, error) {
+		var response map[string]any
+		err := s.runningHubJSON(withProviderRequestKind(ctx, "poll"), config, root+"/task/openapi/outputs", map[string]any{"apiKey": runningHubAPIKey(config), "taskId": taskID}, &response)
+		if err != nil {
+			return videoPollOutcome{}, fmt.Errorf("RunningHub 查询任务失败：%w", err)
+		}
+		code, validCode := runningHubPayloadCode(response)
+		if !validCode {
+			validCode = len(runningHubOutputURLs(response["data"])) > 0
+			code = 0
+		}
+		if !validCode {
+			return videoPollOutcome{}, errors.New("RunningHub 查询响应缺少可识别状态")
+		}
+		if code == 0 {
+			urls := runningHubOutputURLs(response["data"])
+			if len(urls) == 0 {
+				return videoPollOutcome{}, errors.New("RunningHub 任务成功但没有返回产物")
+			}
+			for index, rawURL := range urls {
+				urls[index] = resolveRunningHubOutputURL(root, rawURL)
+			}
+			result, err := s.downloadWorkflowVideoOutputs(ctx, urls, taskID, policy)
+			if err != nil {
+				return videoPollOutcome{}, err
+			}
+			_ = s.updateWorkflowProviderState(ctx, taskID, "succeeded", nil)
+			return videoPollOutcome{Done: true, Result: result}, nil
+		}
+		if code == 805 || code == 806 {
+			return videoPollOutcome{}, fmt.Errorf("RunningHub 任务失败：%s", runningHubFailureMessage(response))
+		}
+		stage := "running"
+		if code == 813 {
+			stage = "queued"
+		} else if code != 804 {
+			stage = "pending"
+		}
+		next := time.Now().Add(policy.Interval)
+		_ = s.updateWorkflowProviderState(ctx, taskID, stage, &next)
+		return videoPollOutcome{}, nil
+	})
+}
+
+func (s *Service) pollRunningHubWorkflowLegacy(ctx context.Context, config providerConfig, root string, taskID string) (map[string]interface{}, error) {
 	for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); {
 		var response map[string]any
 		err := s.runningHubJSON(withProviderRequestKind(ctx, "poll"), config, root+"/task/openapi/outputs", map[string]any{"apiKey": runningHubAPIKey(config), "taskId": taskID}, &response)
@@ -1768,6 +1820,14 @@ func runningHubAPIKey(config providerConfig) string {
 }

 func (s *Service) downloadWorkflowOutputs(ctx context.Context, urls []string) (map[string]interface{}, error) {
+	return s.downloadWorkflowOutputsWithPolicy(ctx, urls, "", nil)
+}
+
+func (s *Service) downloadWorkflowVideoOutputs(ctx context.Context, urls []string, taskID string, policy videoPollPolicy) (map[string]interface{}, error) {
+	return s.downloadWorkflowOutputsWithPolicy(ctx, urls, taskID, &policy)
+}
+
+func (s *Service) downloadWorkflowOutputsWithPolicy(ctx context.Context, urls []string, taskID string, policy *videoPollPolicy) (map[string]interface{}, error) {
 	images := make([]map[string]interface{}, 0)
 	var video, audio map[string]interface{}
 	for _, rawURL := range urls {
@@ -1793,7 +1853,16 @@ func (s *Service) downloadWorkflowOutputs(ctx context.Context, urls []string) (m
 		if !isPublicMediaURL(rawURL) {
 			continue
 		}
-		data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), rawURL)
+		var data []byte
+		var mimeType string
+		var err error
+		if policy == nil {
+			data, mimeType, err = getExternalBinary(withProviderRequestKind(ctx, "download"), rawURL)
+		} else {
+			data, mimeType, err = runVideoDownload(ctx, taskID, *policy, func(ctx context.Context) ([]byte, string, error) {
+				return getExternalBinary(withProviderRequestKind(ctx, "download"), rawURL)
+			})
+		}
 		if err != nil {
 			return nil, fmt.Errorf("下载 RunningHub 产物失败：%w", err)
 		}
--- a/backend/internal/app/workflow_video_polling_test.go
+++ b/backend/internal/app/workflow_video_polling_test.go
@@ -0,0 +1,41 @@
+package app
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestRunningHubVideoPollRecoversFromTransientGatewayFailure(t *testing.T) {
+	pollCalls := 0
+	var server *httptest.Server
+	server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/task/openapi/outputs":
+			pollCalls++
+			w.Header().Set("Content-Type", "application/json")
+			if pollCalls == 1 {
+				w.WriteHeader(http.StatusBadGateway)
+				return
+			}
+			_, _ = w.Write([]byte(`{"code":0,"data":[{"fileUrl":"` + server.URL + `/result.mp4"}]}`))
+		case "/result.mp4":
+			w.Header().Set("Content-Type", "video/mp4")
+			_, _ = w.Write([]byte("video"))
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	defer server.Close()
+
+	config := providerConfig{BaseURL: server.URL, APIKey: "test-key", AllowLocalChannel: true}
+	ctx := withProviderOutboundPolicy(context.Background(), config)
+	result, err := (&Service{}).pollRunningHubVideoWorkflowWithPolicy(ctx, config, server.URL, "provider-task-1", fastVideoPollPolicy())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result["mode"] != "video" || pollCalls != 2 {
+		t.Fatalf("result = %#v, poll calls = %d", result, pollCalls)
+	}
+}
--- a/backend/internal/platform/runtime_policy.go
+++ b/backend/internal/platform/runtime_policy.go
@@ -127,7 +127,7 @@ func DefaultRuntimePolicy() RuntimePolicySetting {
 			ImageTimeoutMinutes:      8,
 			TextTimeoutMinutes:       8,
 			AudioTimeoutMinutes:      8,
-			VideoTimeoutMinutes:      30,
+			VideoTimeoutMinutes:      60,
 			StoryboardTimeoutMinutes: 20,
 			DefaultTimeoutMinutes:    10,
 		},
--- a/docs/content/docs/backend/newapi-channel-2.mdx
+++ b/docs/content/docs/backend/newapi-channel-2.mdx
@@ -26,26 +26,18 @@ description: Channel 2 视频任务创建、轮询与上游任务同步延迟处

 宿主只在同时满足以下条件时把该响应视为“任务状态尚未同步”：

-1. 渠道接口类型精确为 `newapi-channel-2`；
-2. 已经保存非空 Provider Task ID；
-3. HTTP 状态为 `400`；
-4. JSON 的 `code` 或 `message` 精确等于 `task_not_exist`。
+1. 已经保存非空 Provider Task ID；
+2. HTTP 状态为 `400` 或 `404`；
+3. JSON 的 `code` 或 `message` 精确等于 `task_not_exist`、`task_not_found` 或对应的空格拼写。

-不符合条件的 HTTP 错误、非 JSON 正文或其他错误码仍立即按失败语义处理。
+该判断属于宿主统一的视频轮询语义，不再按渠道名分支。不符合条件的 400/401/403/422 等参数、鉴权或权限错误仍立即按失败语义处理。

 ## 轮询与延期

-新任务首次查询前等待 `2` 秒；同步错误在同一次执行中按 `2` 秒、`3` 秒、`5` 秒、`5` 秒退避，最多短期查询 5 次。短期窗口耗尽后，Worker 不发送新的创建请求，也不切换渠道或退款，而是：
+新任务和恢复任务都在首次查询前等待 30 秒，后续统一每 30 秒查询。连续三次 `task_not_exist` 或 404 才结束；中间任何一次正常的 pending/running 响应都会清零连续计数。HTTP 408/409/425/429/5xx、临时网络错误、断流、渠道槽位和短时熔断继续查询原 Provider Task ID，不发送第二次创建请求。

- 保持本地任务状态为 `running`；
- 保留原 `provider_request_id` 和计费状态；
- 释放当前租约并把 `next_poll_at` 延后 15 秒；
- 下一轮从已保存的 Provider Task ID 恢复，只发送查询请求。
-
-`task_not_exist` 的跨轮次同步等待从本地任务首次开始执行起最多持续 5 分钟。超过该上限仍未得到有效任务状态时，Worker 将任务置为失败并释放活跃任务与计费预留，避免无效 Provider Task ID 永久占用用户任务名额。该限制只针对持续的 `task_not_exist`；查询接口已经返回正常 pending/running 状态后的长视频生成仍按视频任务超时和延期策略恢复。
-
-因此，上游生成耗时 500 至 600 秒时，只要查询接口返回有效的处理中状态，本地会持续以低频查询等待完成，不会把正常排队误报为“模型服务拒绝了请求”。视频真正返回失败状态时，仍按供应商失败处理。
+默认视频执行窗口为 60 分钟；管理员已经保存的运行时策略仍按保存值执行。视频真正返回 failed/cancelled/expired 等终态时，仍按供应商失败处理。

 ## 兼容边界

-该容错只适用于 `newapi-channel-2` 视频任务。其他 NewAPI 接口、其他 Provider、缺少 Provider Task ID 的任务，以及真实参数错误不进入延期路径。
+该容错适用于所有异步视频声明式插件和宿主内置协议。缺少 Provider Task ID 的创建失败、真实参数错误、鉴权错误和非法插件映射不进入重试路径。
--- a/docs/content/docs/progress/pending-test.mdx
+++ b/docs/content/docs/progress/pending-test.mdx
@@ -5,6 +5,18 @@ description: 当前项目已实现功能的手工验证清单与未确认边界

 # 待测试

+## 异步视频统一轮询容错
+
+- 声明式视频插件、遗留内置视频协议和 RunningHub 视频工作流统一在创建成功后等待 30 秒首次查询，后续每 30 秒查询；图片、音频和文本协议不受影响。
+- 查询阶段的 HTTP 408/409/425/429/5xx、临时网络错误、断流、渠道槽位或熔断，以及最多连续两次非法 JSON 不再直接结束任务；HTTP 404 和结构化 `task_not_exist` 连续第三次才结束。
+- 结果下载复用同一暂态错误分类，最多尝试 3 次且不重新创建或轮询上游任务。默认视频任务执行窗口由 30 分钟改为 60 分钟；已保存的管理员策略不会被覆盖。
+- [x] `cd backend && go test ./internal/app ./internal/platform -count=1`：通过。
+- [x] `cd backend && go test ./... -count=1`、`go vet ./...`：通过。
+- [ ] 在测试渠道制造一次查询 502，确认任务保持运行、供应商任务 ID 不变，后续查询恢复后正常完成且不重复扣费。
+- [ ] 检查任务日志只出现一次「查询暂时异常」与一次「查询已恢复」，请求明细仍保留每次上游调用。
+- [ ] 让视频下载首次返回 502、第二次成功，确认宿主不重新查询或创建任务；连续三次下载失败时应返回包含供应商任务 ID 的错误。
+- [ ] 将管理员视频超时仍设为旧值的升级环境启动一次，确认保存值保持不变；新安装环境默认显示 60 分钟。
+
 ## 画布图片资源缓存链路优化

 消除首屏满屏转圈与"一直在下载"：后端图片资源改 30 天磁盘强缓存（`Cache-Control: max-age=2592000`），前端内存防抖 IndexedDB 元数据写入、O(1) 淘汰预检、并发下载 4→16，视口内远程图片直链首帧上屏（走原生磁盘缓存秒开），Blob 缓存后台平滑升级。已编译通过，需浏览器最终确认：