diff --git a/backend/internal/app/task_queued_channel_test.go b/backend/internal/app/task_queued_channel_test.go new file mode 100644 index 000000000..e58de47f3 --- /dev/null +++ b/backend/internal/app/task_queued_channel_test.go @@ -0,0 +1,126 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "infinite-canvas/backend/internal/model" +) + +type queuedChannelExecutionProbe struct { + *Service + calls int +} + +func (p *queuedChannelExecutionProbe) processTask(context.Context, model.Task) (map[string]interface{}, []map[string]interface{}, error) { + p.calls++ + return nil, nil, errors.New("execution probe reached") +} + +func TestQueuedCustomChannelRecheck(t *testing.T) { + for _, tc := range []struct { + name string + retry, enabled, system, uncertain bool + dispatch string + }{ + {name: "new"}, {name: "retry", retry: true}, + {name: "enabled", enabled: true}, {name: "system", system: true}, + {name: "accepted", dispatch: "accepted", uncertain: true}, + {name: "submission_unknown", dispatch: "submission_unknown", uncertain: true}, + {name: "billing_review", uncertain: true}, + } { + t.Run(tc.name, func(t *testing.T) { + s, db := newMediaRecoveryTestService(t) + actor := &model.User{ID: "admin", Role: "admin"} + features := FeatureAvailability{CustomChannelsEnabled: true, TaskCenterEnabled: true, CreditsEnabled: true} + if _, err := s.UpdateFeatureAvailability(actor, features); err != nil { + t.Fatal(err) + } + input := map[string]any{"mode": "text", "config": map[string]any{"baseUrl": "https://provider.example", "apiKey": "test-key", "model": "test"}} + if tc.system { + input["config"].(map[string]any)["channelId"] = "system-channel" + } + if err := s.requireCustomChannelsForTaskInput(input); err != nil { + t.Fatal(err) + } + if err := s.protectTaskSecrets(input); err != nil { + t.Fatal(err) + } + encoded, _ := json.Marshal(input) + task := &model.Task{ID: newID(), UserID: "user", Type: "canvas_text", Status: model.TaskStatusQueued, InputJSON: string(encoded), RouteRun: 1} + if tc.retry { + task.Status = model.TaskStatusFailed + task.Attempts = 1 + } + if err := db.Create(task).Error; err != nil { + t.Fatal(err) + } + if tc.retry { + var err error + task, err = s.repo.RetryTaskWithBilling(task.UserID, task, nil, 10) + if err != nil { + t.Fatal(err) + } + } + order := model.BillingOrder{ID: newID(), UserID: task.UserID, TaskID: task.ID, IdempotencyKey: newID(), Status: model.BillingStatusReserved, AmountMicrocredits: 100, ReservedAmountMicrocredits: 100} + if tc.uncertain && tc.dispatch == "" { + order.Status = model.BillingStatusUncertain + } + if tc.dispatch != "" { + attempt := &model.RouteAttempt{ID: newID(), TaskID: task.ID, RouteRun: task.RouteRun, DispatchState: tc.dispatch, Status: "selected", AttemptNumber: 1} + if err := db.Create(attempt).Error; err != nil { + t.Fatal(err) + } + } + for _, value := range []any{&order, &model.CreditAccount{UserID: task.UserID, AvailableMicrocredits: 900, ReservedMicrocredits: 100}} { + if err := db.Create(value).Error; err != nil { + t.Fatal(err) + } + } + if err := db.Model(task).Update("billing_order_id", order.ID).Error; err != nil { + t.Fatal(err) + } + features.CustomChannelsEnabled = tc.enabled + if _, err := s.UpdateFeatureAvailability(actor, features); err != nil { + t.Fatal(err) + } + claimed, err := s.repo.ClaimNextTask("worker", time.Minute) + if err != nil || claimed == nil { + t.Fatalf("claim: %v %v", claimed, err) + } + probe := &queuedChannelExecutionProbe{Service: s} + s.taskRouteExecutor = &taskRouteExecutor{port: probe} + err = s.taskWorker().processClaimedTask(claimed, nil) + stored, readErr := s.repo.Task(task.ID) + if readErr != nil { + t.Fatal(readErr) + } + wantCalls, wantError := 0, "自定义渠道" + if tc.enabled || tc.system { + wantCalls, wantError = 1, "execution probe reached" + } + if err == nil || !strings.Contains(stored.Error, wantError) || stored.Status != model.TaskStatusFailed || probe.calls != wantCalls || stored.CompletedAt == nil || stored.LeaseOwner != "" { + t.Errorf("queued channel rejection: err=%v status=%s error=%q calls=%d", err, stored.Status, stored.Error, probe.calls) + } + got, err := s.repo.BillingOrder(order.ID) + if err != nil { + t.Fatal(err) + } + var account model.CreditAccount + if err := db.First(&account, "user_id = ?", task.UserID).Error; err != nil { + t.Fatal(err) + } + wantStatus, available, reserved := model.BillingStatusRefunded, int64(1000), int64(0) + if tc.uncertain { + wantStatus, available, reserved = model.BillingStatusUncertain, 900, 100 + } + if got.Status != wantStatus || account.AvailableMicrocredits != available || account.ReservedMicrocredits != reserved { + t.Errorf("reservation not refunded: status=%s available=%d reserved=%d", got.Status, account.AvailableMicrocredits, account.ReservedMicrocredits) + } + }) + } +} diff --git a/backend/internal/app/task_worker.go b/backend/internal/app/task_worker.go index 5a1465bc7..cfd176463 100644 --- a/backend/internal/app/task_worker.go +++ b/backend/internal/app/task_worker.go @@ -191,6 +191,34 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot return s.finishTaskMediaRecovery(task, result, recoveryErr) } + // Creation and manual retry admission can precede execution by a long time. + // Re-read the feature policy before preparing or dispatching a model route. + decryptedInput, err := s.decryptTaskInputJSON(task.InputJSON) + if err != nil { + return err + } + var admissionInput map[string]any + if err := json.Unmarshal([]byte(decryptedInput), &admissionInput); err != nil { + return err + } + if err := s.requireCustomChannelsForTaskInput(admissionInput); err != nil { + // A reclaimed task may already have submitted upstream. Keep the existing + // review path for that evidence; only an unsent reservation is refundable. + attempts, readErr := s.repo.RouteAttempts(task.ID, task.RouteRun) + uncertain := readErr != nil || task.ProviderRequestID != "" || s.taskBilling().BillingFailureRequiresReview(task.BillingOrderID, task.ID, err) + for _, attempt := range attempts { + if attempt.DispatchState == "accepted" || attempt.DispatchState == "submission_unknown" { + uncertain = true + } + } + // Legacy tasks without an explicit new retry run retain their conservative + // submission handling, as in createDirectTaskAttempt. + if len(attempts) == 0 && task.RouteRun <= 1 && task.Attempts > 1 { + uncertain = true + } + return terminal.markPreparationFailure(task, "渠道不可用", errors.Join(err, readErr), uncertain, "自定义渠道已关闭,上游请求未发出") + } + s.markAgentMemoryCompactRunning(*task) task.Stage = "调用生成模型" task.Progress = 35 diff --git a/docs/content/docs/backend/task-runtime-stability.mdx b/docs/content/docs/backend/task-runtime-stability.mdx index 087dfd697..0c41c139f 100644 --- a/docs/content/docs/backend/task-runtime-stability.mdx +++ b/docs/content/docs/backend/task-runtime-stability.mdx @@ -5,6 +5,10 @@ description: 热点回源保护、并发租约、故障边界和上线验收 # 任务缓存与队列稳定性 +## 排队后的自定义渠道检查 + +worker 在准备模型路由前重新读取自定义渠道开关,覆盖首次排队和失败后重新入队。排队期间关闭开关的任务以“自定义渠道”不可用的明确错误进入失败终态,不调用渠道;未发出的预扣沿用已有失败退款路径。已接受、提交不明或原本待核对的订单仍进入既有待核对路径,不将可能已发生的费用自动退回。已生成作品的媒体保存恢复不重新生成,不受此入口阻断。 + ## 详情读取和日志筛选边界 任务中心打开详情只读取任务和日志,不在该路径调用画布结果同步。后台同步遇到短剧领域项目任务(`domainProjectId` 等于任务 `projectId`)或没有生成节点关联时跳过画布写回,避免把短剧项目 ID 当作画布 ID 查询。 diff --git a/docs/content/docs/progress/pending-test.mdx b/docs/content/docs/progress/pending-test.mdx index 96b1fc4cb..108a49a87 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -23,6 +23,10 @@ description: 当前项目已实现功能的手工验证清单与未确认边界 - [ ] 运行 `TestUploadContentImportSocket` 验证 URL 导入与重试;在可解析外部存储域名的环境复测 `TestResolveAccessPolicyMatrix`。 - [ ] PostgreSQL 上传重试、真实 OSS 附件代理与浏览器附件下载验收。 +## 排队后的自定义渠道检查 + +- [ ] PostgreSQL 环境验证排队期间关闭自定义渠道:首次与重试任务均停止调用,未发出订单退款,已提交或待核对订单保持待核对。 + ## Agent 失败消息合并与展示 - 已修复:同一运行的 `run_failed` / `error` 与失败状态快照共用一条错误消息;以公开失败原因补全内容,重放、乱序和空原因不覆盖已有详细说明。不同运行即使错误文本相同也分别保留,工具失败与连接错误不合并。