From 98c6127ef701302251ab3710f41742ac55bf5da9 Mon Sep 17 00:00:00 2001 From: andy Date: Thu, 3 Sep 2026 17:23:00 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=87=E6=A1=88=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/jobs/provider.go | 18 ++++++++- backend/internal/jobs/provider_test.go | 48 +++++++++++++++++++++++- backend/internal/providers/provider.go | 3 +- backend/internal/providers/volcengine.go | 4 +- 4 files changed, 67 insertions(+), 6 deletions(-) diff --git a/backend/internal/jobs/provider.go b/backend/internal/jobs/provider.go index 98db5d4..9e74d3d 100644 --- a/backend/internal/jobs/provider.go +++ b/backend/internal/jobs/provider.go @@ -74,7 +74,7 @@ func (p ProviderProcessor) Advance(ctx context.Context, job Job) (Job, error) { ErrorClass: "unknown_outcome", }) failed := StatusFailed - failure := &JobError{Message: "provider submission outcome is unknown; refusing duplicate submission", Retryable: false} + failure := &JobError{Message: unknownProviderSubmissionMessage, Retryable: false} if p.Store == nil { job.Status, job.Error = failed, failure return job, nil @@ -103,7 +103,7 @@ func (p ProviderProcessor) Advance(ctx context.Context, job Job) (Job, error) { } if err != nil && job.ProviderTaskID == "" && p.Store != nil { failed := StatusFailed - failure := &JobError{Message: "provider submission outcome is unknown; refusing duplicate submission", Retryable: false} + failure := providerSubmissionFailure(err) return p.Store.UpdateJob(ctx, job.ID, workerPatch(job, Patch{Status: &failed, Error: failure})) } if err != nil { @@ -152,6 +152,20 @@ func (p ProviderProcessor) Advance(ctx context.Context, job Job) (Job, error) { return job, nil } +const ( + unknownProviderSubmissionMessage = "provider submission outcome is unknown; refusing duplicate submission" + outputImageSafetyMessage = "生成结果触发内容安全审核，请更换素材或调整内容后重试。" +) + +func providerSubmissionFailure(err error) *JobError { + message := unknownProviderSubmissionMessage + var providerError *providers.ProviderError + if errors.As(err, &providerError) && (providerError.Code == "OutputImageSensitiveContentDetected" || strings.HasPrefix(providerError.Code, "OutputImageSensitiveContentDetected.")) { + message = outputImageSafetyMessage + } + return &JobError{Message: message, Retryable: false} +} + type providerFailureDiagnostic struct { JobID string Provider string diff --git a/backend/internal/jobs/provider_test.go b/backend/internal/jobs/provider_test.go index 7933bcd..0c17d6c 100644 --- a/backend/internal/jobs/provider_test.go +++ b/backend/internal/jobs/provider_test.go @@ -415,7 +415,7 @@ func TestProviderProcessorLogsSafeJobCorrelationWhenSubmitFails(t *testing.T) { processor := ProviderProcessor{Providers: ProviderRegistry{"volcengine-visual": adapter}, Store: store} got, err := processor.Advance(context.Background(), job) - if err != nil || got.Status != StatusFailed { + if err != nil || got.Status != StatusFailed || got.Error == nil || got.Error.Message != unknownProviderSubmissionMessage { t.Fatalf("got=%#v err=%v", got, err) } logged := output.String() @@ -431,6 +431,52 @@ func TestProviderProcessorLogsSafeJobCorrelationWhenSubmitFails(t *testing.T) { } } +func TestProviderProcessorPersistsFriendlySeedreamOutputSafetyMessage(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/api/v3/images/generations" { + t.Fatalf("request path = %q", request.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{ + "error": { + "code": "OutputImageSensitiveContentDetected.PolicyViolation", + "type": "BadRequest", + "message": "private upstream moderation detail" + } + }`)) + })) + defer server.Close() + + store := newMemoryJobStore() + job := Job{ + ID: "job-seedream-output-safety", OwnerID: "owner", Provider: "seedream", ReqKey: providers.Seedream50ProModel, + Capability: "image.generate", Status: StatusRunning, LockedBy: "worker", + RequestPayload: json.RawMessage(`{ + "capability":"image.generate", + "model":"doubao-seedream-5-0-pro-260628", + "prompt":"拆分元素", + "inputUrls":["https://assets.test/source.jpg"], + "settings":{"layerDecomposition":true,"size":"2K","outputFormat":"png"} + }`), + } + store.jobs[job.ID] = job + adapter := providers.NewSeedream(providers.Config{ + BaseURL: server.URL + "/api/v3", APIKey: "private-api-key", Model: providers.Seedream50ProModel, + }, server.Client()) + processor := ProviderProcessor{Providers: ProviderRegistry{"seedream": adapter}, Store: store} + + got, err := processor.Advance(context.Background(), job) + const want = "生成结果触发内容安全审核，请更换素材或调整内容后重试。" + if err != nil || got.Status != StatusFailed || got.Error == nil || got.Error.Message != want || got.Error.Retryable { + t.Fatalf("got=%#v err=%v", got, err) + } + persisted := store.jobs[job.ID] + if persisted.Error == nil || persisted.Error.Message != want { + t.Fatalf("persisted error = %#v", persisted.Error) + } +} + func TestProviderProcessorClearsTransientErrorAfterSuccessfulPoll(t *testing.T) { store := newMemoryJobStore() job := Job{ID: "job-recovered", OwnerID: "owner", Provider: "fixture", ReqKey: "model-a", Capability: "image.generate", Status: StatusQueued, LockedBy: "worker", ProviderTaskID: "provider-task", Error: &JobError{Message: "temporary timeout", Retryable: true}, RequestPayload: json.RawMessage(`{"capability":"image.generate","model":"model-a","prompt":"hello"}`)} diff --git a/backend/internal/providers/provider.go b/backend/internal/providers/provider.go index ff31cdb..ef243b9 100644 --- a/backend/internal/providers/provider.go +++ b/backend/internal/providers/provider.go @@ -111,6 +111,7 @@ type Config struct { type ProviderError struct { Operation string Status int + Code string Cause error } @@ -215,7 +216,7 @@ func (a *httpAdapter) call(ctx context.Context, method, path string, body []byte ErrorClass: "service", ElapsedMS: elapsedMilliseconds(startedAt), }) - return Result{}, &ProviderError{Operation: a.name + " " + operation, Status: resp.StatusCode} + return Result{}, &ProviderError{Operation: a.name + " " + operation, Status: resp.StatusCode, Code: code} } if !json.Valid(raw) { logHTTPProviderFailure(httpProviderDiagnostic{ diff --git a/backend/internal/providers/volcengine.go b/backend/internal/providers/volcengine.go index 6a51a1c..d4f3923 100644 --- a/backend/internal/providers/volcengine.go +++ b/backend/internal/providers/volcengine.go @@ -169,7 +169,7 @@ func (v *Volcengine) call(ctx context.Context, action, version string, payload a if resp.StatusCode < 200 || resp.StatusCode >= 300 { diagnostic.ErrorClass = "service" logVolcengineFailure(diagnostic) - return Result{}, &ProviderError{Operation: "volcengine request", Status: resp.StatusCode} + return Result{}, &ProviderError{Operation: "volcengine request", Status: resp.StatusCode, Code: diagnostic.Code} } if !validJSON { diagnostic.ErrorClass = "invalid_json" @@ -179,7 +179,7 @@ func (v *Volcengine) call(ctx context.Context, action, version string, payload a if response.code != nil && !volcengineRequestSucceeded(response.code) { diagnostic.ErrorClass = "service" logVolcengineFailure(diagnostic) - return Result{}, &ProviderError{Operation: "volcengine request"} + return Result{}, &ProviderError{Operation: "volcengine request", Code: diagnostic.Code} } d := object(first(response.business["data"], response.business["Data"])) out := []string{}