diff --git a/backend/internal/app/cloud_agent_empty_output.go b/backend/internal/app/cloud_agent_empty_output.go index 93450eecd..2eefcedf6 100644 --- a/backend/internal/app/cloud_agent_empty_output.go +++ b/backend/internal/app/cloud_agent_empty_output.go @@ -9,6 +9,9 @@ import ( const cloudAgentMaxEmptyOutputNudges = 2 +// cloudAgentMaxEmptyOutputEscalations 是"空输出升级重试"次数：催办没用时改为关思考 + 放大预算。 +const cloudAgentMaxEmptyOutputEscalations = 1 + // cloudAgentEmptyModelOutput 判断模型任务是不是「上游成功、但正文与工具调用都为空」。 // 开思考时偶发只流 reasoning、可见正文和工具调用都空，解析层会归为「没有返回内容」。 func cloudAgentEmptyModelOutput(task *model.Task) bool { @@ -28,3 +31,21 @@ func (s *Service) correctCloudAgentEmptyOutput(run *model.CloudAgentExecution, s return cloudAgentSave(current, state) }) } + +// correctCloudAgentEmptyOutputEscalation 是空输出的兜底：催办无效时关掉上游思考并放大输出预算， +// 重新发同一步（不追加催办消息，避免模型误以为出现了新任务）。 +func (s *Service) correctCloudAgentEmptyOutputEscalation(run *model.CloudAgentExecution, state *cloudAgentRuntime) error { + return s.repo.MutateCloudAgent(run.UserID, run.ID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error { + state.EmptyOutputEscalated++ + state.ForceThinkingOff = true + state.BoostStepOutputBudget = true + state.ActiveTaskID = "" + state.Calls = nil + state.CallIndex = 0 + state.event(run.ID, "model_failure_recovered", map[string]any{ + "reason": "empty_output_escalated", + "text": "上游连续返回空内容；已关闭思考并放大输出预算重试同一步", + }) + return cloudAgentSave(current, state) + }) +} diff --git a/backend/internal/app/cloud_agent_runtime.go b/backend/internal/app/cloud_agent_runtime.go index 4ac672348..2e1971380 100644 --- a/backend/internal/app/cloud_agent_runtime.go +++ b/backend/internal/app/cloud_agent_runtime.go @@ -82,6 +82,18 @@ type cloudAgentRuntime struct { TransientReferences map[string]cloudAgentTransientReference `json:"transientReferences,omitempty"` InterjectionIDs []string `json:"interjectionIds,omitempty"` Events []CloudAgentEvent `json:"events"` + // EmptyOutputEscalated 记录"空输出已经升级重试过几次"（关思考 + 放大输出预算）。 + EmptyOutputEscalated int `json:"emptyOutputEscalated,omitempty"` + // StepTimeoutEscalated 记录"单步墙钟到点后已经关思考重试过几次"。 + StepTimeoutEscalated int `json:"stepTimeoutEscalated,omitempty"` + // ForceThinkingOff 让本步请求强制关闭上游思考：思考模型偶发把整个输出预算花在推理上， + // 结果正文与工具调用皆空（实测 output_tokens 正好等于 maxOutputTokens）。 + ForceThinkingOff bool `json:"forceThinkingOff,omitempty"` + // BoostStepOutputBudget 让本步请求使用放大后的输出预算（配合关思考重试）。 + BoostStepOutputBudget bool `json:"boostStepOutputBudget,omitempty"` + // StepLimits 是本步实际生效的执行边界（管理员策略解析结果），只用于构造请求： + // 不进状态 JSON——每次推进都按当时的策略重新解析，改配置无需重发本轮。 + StepLimits cloudAgentStepLimits `json:"-"` } type cloudAgentTransientReference struct { @@ -105,7 +117,7 @@ func (s *Service) ensureCloudAgentExecution(task *model.Task, initial cloudAgent canonical := input.Requests.Canonical canonical.SystemPrompt = stripCloudAgentPlanBlock(canonical.SystemPrompt) canonical.Messages = stripCloudAgentRuntimeContext(canonical.Messages) - state := cloudAgentRuntime{Request: initial.Request, Policy: initial.Policy, ParentID: initial.ParentID, Fingerprint: initial.Fingerprint, CreativeAnchor: initial.CreativeAnchor, TextHistory: input.TextHistory, Skills: initial.Skills, Profile: initial.Profile, Canonical: canonical, ActiveTaskID: task.ID, TaskIDs: []string{task.ID}, Step: 1, Decisions: map[string]string{}, Plan: initial.Plan, Events: []CloudAgentEvent{}} + state := cloudAgentRuntime{Request: initial.Request, Policy: initial.Policy, ParentID: initial.ParentID, Fingerprint: initial.Fingerprint, CreativeAnchor: initial.CreativeAnchor, TextHistory: input.TextHistory, Skills: initial.Skills, Profile: initial.Profile, Canonical: canonical, ActiveTaskID: task.ID, TaskIDs: []string{task.ID}, Step: 1, Decisions: map[string]string{}, Plan: initial.Plan, Events: []CloudAgentEvent{}, StepLimits: s.cloudAgentStepLimits()} if len(initial.Skills) > 0 { state.event(task.ID, "tool_completed", map[string]any{"toolName": "skills_load", "text": fmt.Sprintf("已启用 %d 个技能，正文将按需读取", len(initial.Skills))}) } @@ -595,6 +607,8 @@ func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error) } return s.terminateCloudAgent(run, "Agent 运行状态损坏，本轮已停止") } + // 单步边界每次推进都重新解析：管理员改配置后，正在跑的这一轮下一步就用新值。 + state.StepLimits = s.cloudAgentStepLimits() if state.ActiveTaskID != "" { task, err := s.repo.TaskForUser(run.UserID, state.ActiveTaskID) if errors.Is(err, gorm.ErrRecordNotFound) { @@ -652,8 +666,18 @@ func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error) if task.Status != model.TaskStatusSucceeded && cloudAgentTruncatedToolArguments(task) { return s.correctCloudAgentTruncatedCalls(run, &state) } - if cloudAgentEmptyModelOutput(task) && state.EmptyOutputNudged < cloudAgentMaxEmptyOutputNudges { - return s.correctCloudAgentEmptyOutput(run, &state) + if cloudAgentEmptyModelOutput(task) { + if state.EmptyOutputNudged < cloudAgentMaxEmptyOutputNudges { + return s.correctCloudAgentEmptyOutput(run, &state) + } + // 催过仍然空：改为"关思考 + 放大输出预算"重试同一步，而不是把整轮判死。 + if state.EmptyOutputEscalated < cloudAgentMaxEmptyOutputEscalations { + return s.correctCloudAgentEmptyOutputEscalation(run, &state) + } + } + // 单步墙钟到点同样是可恢复失败：关思考重试一次，而不是把整轮判死。 + if cloudAgentStepTimedOut(task) && state.StepTimeoutEscalated < cloudAgentMaxStepTimeoutEscalations { + return s.correctCloudAgentStepTimeout(run, &state) } return s.repo.MutateCloudAgent(run.UserID, run.ID, run.Revision, func(current *model.CloudAgentExecution, repo *repository.Repository) error { if task.Status != model.TaskStatusSucceeded { @@ -726,7 +750,11 @@ func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error) return contextErr } s.attachCloudAgentLessons(&canonical, run.UserID, cloudAgentLessonTaskText(&state)) - input := map[string]any{"mode": "text", "prompt": state.Request.Prompt, "agentRequests": map[string]any{"canonical": canonical}, "config": map[string]any{"channelId": state.Request.ChannelID, "channelModelKey": state.Request.ChannelModelKey, "model": firstNonEmpty(state.Request.ChannelModelKey, state.Request.Model)}, "textOptions": map[string]any{"stream": true, "thinking": cloudAgentReasoningEnabled(state.Policy.ReasoningMode)}} + // 单步输出上限与思考开关：默认按策略给每一步带上界（不带上界时上游按剩余上下文放行， + // 思考模型可以把单步拖到几分钟）；空输出升级重试时改为关思考 + 放大预算。 + stepThinking := cloudAgentReasoningEnabled(state.Policy.ReasoningMode) && !state.ForceThinkingOff + stepOutputTokens := cloudAgentStepOutputBudget(state.StepLimits, state.BoostStepOutputBudget) + input := map[string]any{"mode": "text", "prompt": state.Request.Prompt, "agentRequests": map[string]any{"canonical": canonical}, "config": map[string]any{"channelId": state.Request.ChannelID, "channelModelKey": state.Request.ChannelModelKey, "model": firstNonEmpty(state.Request.ChannelModelKey, state.Request.Model)}, "textOptions": map[string]any{"stream": true, "thinking": stepThinking, "maxOutputTokens": stepOutputTokens}} tokens, tokenErr := cloudAgentRequestEstimatedTokens(&canonical) if tokenErr != nil { return s.failCloudAgent(run, &state, "模型上下文估算失败，请稍后重试") @@ -861,6 +889,11 @@ func cloudAgentModelFailure(task *model.Task) (string, string) { } raw := strings.ToLower(task.Error) switch { + case strings.Contains(task.Error, "没有返回内容"): + // 思考模型的典型失败：整个输出预算被推理吃掉，正文与工具调用皆空。 + detail, reason = "上游连续返回空内容（通常是思考占满输出预算）；已自动关思考并放大预算重试仍失败，建议换用非思考模型或调小上下文", "model_empty_output" + case strings.Contains(task.Error, cloudAgentStepTimeoutError): + detail, reason = "单步模型调用超过执行时限仍未返回（长思考或上下文过大时常见）；已自动关思考重试仍超时，可在管理端调大 Agent 单步超时", "model_step_timeout" case strings.Contains(raw, "connection reset by peer"): detail, reason = "模型连接被对端或中间网络设备重置", "model_connection_reset" case strings.Contains(raw, "timeout"), strings.Contains(raw, "deadline exceeded"): diff --git a/backend/internal/app/cloud_agent_step_limits.go b/backend/internal/app/cloud_agent_step_limits.go new file mode 100644 index 000000000..d0d90b4c6 --- /dev/null +++ b/backend/internal/app/cloud_agent_step_limits.go @@ -0,0 +1,55 @@ +package app + +import ( + "time" + + "infinite-canvas/backend/internal/platform" +) + +// 画布 Agent 单步的执行边界：一次模型调用最多输出多少 token、最多等多久。 +// 两个值都来自运行时策略（管理端可改、环境变量可给部署级默认值），"0"是各自的关闭档位。 + +// cloudAgentStepOperation 是一次"模型调用"任务的操作名。它同时是任务中心里区分 +// "这一条任务是 Agent 的某一步"的依据。 +const cloudAgentStepOperation = "cloud_agent_step" + +// cloudAgentStepBoostFallbackTokens 是"不限制输出"（策略值为 0）时放大重试用的预算： +// 不限制的本意是"让模型写完"，重试却必须有个上界，否则一次卡住的调用会一直占着单步墙钟。 +const cloudAgentStepBoostFallbackTokens = 32_768 + +// cloudAgentStepLimits 是一次模型调用实际生效的执行边界。 +type cloudAgentStepLimits struct { + // OutputTokens 是本次调用的输出上限；0 表示不限制（只有 Timeout 兜底）。 + OutputTokens int + // Timeout 是单步墙钟：策略给了秒级值时用它，否则沿用文本任务超时。 + Timeout time.Duration +} + +// cloudAgentStepLimits 解析当前生效的单步边界。策略读取失败时退回出厂默认值： +// 拿到一个确定的上界，好过让一次调用无限跑下去。 +func (s *Service) cloudAgentStepLimits() cloudAgentStepLimits { + policy, err := s.runtimeConcurrencySetting() + if err != nil { + policy = defaultRuntimePolicy().Task + } + limits := cloudAgentStepLimits{OutputTokens: policy.AgentStepMaxOutputTokens} + if policy.AgentStepTimeoutSeconds > 0 { + limits.Timeout = time.Duration(policy.AgentStepTimeoutSeconds) * time.Second + } else { + limits.Timeout = time.Duration(policy.TextTimeoutMinutes) * time.Minute + } + return limits +} + +// cloudAgentStepOutputBudget 把生效上限折算成本步请求要带的 maxOutputTokens： +// 放大档（空输出升级重试）在原值上翻倍并以硬上限封顶；原值为 0（不限制）时用兜底值， +// 让重试仍然有界。 +func cloudAgentStepOutputBudget(limits cloudAgentStepLimits, boosted bool) int { + if !boosted { + return limits.OutputTokens + } + if limits.OutputTokens <= 0 { + return cloudAgentStepBoostFallbackTokens + } + return min(limits.OutputTokens*2, platform.MaxRuntimeAgentStepOutputTokens) +} diff --git a/backend/internal/app/cloud_agent_step_timeout.go b/backend/internal/app/cloud_agent_step_timeout.go new file mode 100644 index 000000000..1bf1967f1 --- /dev/null +++ b/backend/internal/app/cloud_agent_step_timeout.go @@ -0,0 +1,57 @@ +package app + +import ( + "strings" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" +) + +// cloudAgentStepTimeoutError 是一次画布 Agent 模型调用被单步墙钟掐断时写进任务错误的标记。 +// 秒级超时（策略 AgentStepTimeoutSeconds）与文本任务超时都会走到这里，运行期按这个标记 +// 把它当成"可恢复失败"处理，而不是整轮判死。 +const cloudAgentStepTimeoutError = "画布 Agent 单步模型调用超时" + +// cloudAgentMaxStepTimeoutEscalations 是单步超时的重试次数：超时几乎总是思考阶段在长时间 +// 无输出，因此重试的手段是关掉思考（更省时的路径），只给一次机会。 +const cloudAgentMaxStepTimeoutEscalations = 1 + +// cloudAgentModelOperation 判断任务是不是"画布 Agent 的一次模型调用"：根任务（第一步）与 +// 后续每一步都属于同一类，都要按单步口径计时。 +func cloudAgentModelOperation(task *model.Task) bool { + if task == nil { + return false + } + switch task.Operation { + case cloudAgentOperation, cloudAgentStepOperation: + return true + default: + return false + } +} + +// cloudAgentStepTimedOut 判断这一步是不是被单步墙钟中止的。 +func cloudAgentStepTimedOut(task *model.Task) bool { + if task == nil || task.Status == model.TaskStatusSucceeded { + return false + } + return strings.Contains(task.Error, cloudAgentStepTimeoutError) +} + +// correctCloudAgentStepTimeout 是单步超时的兜底：关掉上游思考并重发同一步。 +// 超时不改输出预算——墙钟是这一档的约束，放大预算只会让重试更容易再次撞上墙钟。 +func (s *Service) correctCloudAgentStepTimeout(run *model.CloudAgentExecution, state *cloudAgentRuntime) error { + return s.repo.MutateCloudAgent(run.UserID, run.ID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error { + state.StepTimeoutEscalated++ + state.ForceThinkingOff = true + state.BoostStepOutputBudget = false + state.ActiveTaskID = "" + state.Calls = nil + state.CallIndex = 0 + state.event(run.ID, "model_failure_recovered", map[string]any{ + "reason": "step_timeout_retried", + "text": "上一步模型调用超时；已关闭思考重试同一步", + }) + return cloudAgentSave(current, state) + }) +} diff --git a/backend/internal/app/cloud_agent_step_timeout_test.go b/backend/internal/app/cloud_agent_step_timeout_test.go new file mode 100644 index 000000000..9745fbfd3 --- /dev/null +++ b/backend/internal/app/cloud_agent_step_timeout_test.go @@ -0,0 +1,294 @@ +package app + +// 自研：画布 Agent 单步边界（输出上限 + 秒级墙钟）的可配置化与超时可恢复。 +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/platform" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" + "infinite-canvas/backend/internal/repository" +) + +func TestCloudAgentStepOutputBudgetFollowsPolicy(t *testing.T) { + cases := []struct { + name string + outputTokens int + boosted bool + expectedValue int + }{ + {name: "默认上限", outputTokens: platform.DefaultRuntimeAgentStepOutputTokens, expectedValue: 16_384}, + {name: "空输出放大档翻倍", outputTokens: platform.DefaultRuntimeAgentStepOutputTokens, boosted: true, expectedValue: 32_768}, + {name: "自定义上限", outputTokens: 32_000, expectedValue: 32_000}, + {name: "自定义上限放大后封顶", outputTokens: 100_000, boosted: true, expectedValue: platform.MaxRuntimeAgentStepOutputTokens}, + {name: "不限制时原值即 0", outputTokens: 0, expectedValue: 0}, + {name: "不限制时放大档必须有界", outputTokens: 0, boosted: true, expectedValue: cloudAgentStepBoostFallbackTokens}, + } + for _, item := range cases { + t.Run(item.name, func(t *testing.T) { + got := cloudAgentStepOutputBudget(cloudAgentStepLimits{OutputTokens: item.outputTokens}, item.boosted) + if got != item.expectedValue { + t.Fatalf("budget = %d, want %d", got, item.expectedValue) + } + }) + } +} + +func TestTaskExecutionTimeoutPrefersAgentStepSeconds(t *testing.T) { + policy := defaultRuntimePolicy().Task + policy.TextTimeoutMinutes = 8 + policy.AgentStepTimeoutSeconds = 90 + step := &model.Task{Type: "canvas_text", Operation: cloudAgentStepOperation} + if got := taskExecutionTimeout(step, policy); got != 90*time.Second { + t.Fatalf("agent step timeout = %s, want 90s", got) + } + // 没配秒级值时沿用文本任务超时（旧行为不变）。 + policy.AgentStepTimeoutSeconds = 0 + if got := taskExecutionTimeout(step, policy); got != 8*time.Minute { + t.Fatalf("fallback timeout = %s, want 8m", got) + } + // 非 Agent 的文本任务不受这个开关影响。 + policy.AgentStepTimeoutSeconds = 90 + plain := &model.Task{Type: "canvas_text", Operation: "canvas_text_generate"} + if got := taskExecutionTimeout(plain, policy); got != 8*time.Minute { + t.Fatalf("plain text timeout = %s, want 8m", got) + } +} + +// 单步边界必须真的来自策略：管理端改完，运行期下一步就用新值，不需要重发本轮。 +func TestCloudAgentStepLimitsFollowAdminPolicy(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.SystemSetting{}, &model.AdminAuditEvent{}); err != nil { + t.Fatal(err) + } + svc := New(repository.New(db), t.TempDir()) + actor := &model.User{ID: "admin", Role: model.UserRoleAdmin} + + defaults := svc.cloudAgentStepLimits() + if defaults.OutputTokens != platform.DefaultRuntimeAgentStepOutputTokens || defaults.Timeout != 8*time.Minute { + t.Fatalf("default step limits = %+v", defaults) + } + + policy := defaultRuntimePolicy() + policy.Task.AgentStepMaxOutputTokens = 32_768 + policy.Task.AgentStepTimeoutSeconds = 150 + if _, err := svc.UpdateRuntimePolicySetting(actor, policy); err != nil { + t.Fatal(err) + } + updated := svc.cloudAgentStepLimits() + if updated.OutputTokens != 32_768 || updated.Timeout != 150*time.Second { + t.Fatalf("updated step limits = %+v", updated) + } + + // 0 的两个含义：输出不限制、超时沿用文本任务超时。 + policy.Task.AgentStepMaxOutputTokens = 0 + policy.Task.AgentStepTimeoutSeconds = 0 + policy.Task.TextTimeoutMinutes = 12 + if _, err := svc.UpdateRuntimePolicySetting(actor, policy); err != nil { + t.Fatal(err) + } + unbounded := svc.cloudAgentStepLimits() + if unbounded.OutputTokens != 0 || unbounded.Timeout != 12*time.Minute { + t.Fatalf("unbounded step limits = %+v", unbounded) + } +} + +func TestRuntimePolicyRejectsOutOfRangeAgentStepLimits(t *testing.T) { + policy := defaultRuntimePolicy() + policy.Task.AgentStepMaxOutputTokens = platform.MinRuntimeAgentStepOutputTokens - 1 + if err := validateRuntimePolicy(policy); err == nil { + t.Fatal("过小的单步输出上限应被拒绝") + } + policy = defaultRuntimePolicy() + policy.Task.AgentStepMaxOutputTokens = platform.MaxRuntimeAgentStepOutputTokens + 1 + if err := validateRuntimePolicy(policy); err == nil { + t.Fatal("过大的单步输出上限应被拒绝") + } + policy = defaultRuntimePolicy() + policy.Task.AgentStepMaxOutputTokens = 0 + if err := validateRuntimePolicy(policy); err != nil { + t.Fatalf("0（不限制）应当合法: %v", err) + } + policy = defaultRuntimePolicy() + policy.Task.AgentStepTimeoutSeconds = platform.MinRuntimeAgentStepTimeoutSeconds - 1 + if err := validateRuntimePolicy(policy); err == nil { + t.Fatal("过小的单步超时（非 0）应被拒绝") + } + policy = defaultRuntimePolicy() + policy.Task.AgentStepTimeoutSeconds = platform.MaxRuntimeAgentStepTimeoutSeconds + 1 + if err := validateRuntimePolicy(policy); err == nil { + t.Fatal("过大的单步超时应被拒绝") + } +} + +// 旧配置 JSON 没有这两个字段时按默认值回填：不需要数据迁移。 +func TestRuntimePolicyBackfillsAgentStepLimitsForLegacyJSON(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.SystemSetting{}); err != nil { + t.Fatal(err) + } + legacy := defaultRuntimePolicy() + encoded, err := json.Marshal(legacy) + if err != nil { + t.Fatal(err) + } + var value map[string]any + if err := json.Unmarshal(encoded, &value); err != nil { + t.Fatal(err) + } + task := value["task"].(map[string]any) + delete(task, "agentStepMaxOutputTokens") + delete(task, "agentStepTimeoutSeconds") + trimmed, err := json.Marshal(value) + if err != nil { + t.Fatal(err) + } + if err := db.Create(&model.SystemSetting{Key: runtimePolicySettingKey, ValueJSON: string(trimmed)}).Error; err != nil { + t.Fatal(err) + } + svc := New(repository.New(db), t.TempDir()) + effective, err := svc.RuntimePolicy() + if err != nil { + t.Fatal(err) + } + if effective.Task.AgentStepMaxOutputTokens != platform.DefaultRuntimeAgentStepOutputTokens { + t.Fatalf("legacy json 未回填单步输出上限: %d", effective.Task.AgentStepMaxOutputTokens) + } + if effective.Task.AgentStepTimeoutSeconds != platform.DefaultRuntimeAgentStepTimeout { + t.Fatalf("legacy json 未回填单步超时: %d", effective.Task.AgentStepTimeoutSeconds) + } +} + +// 单步墙钟到点不再把整轮判死：关思考重试一次；重试仍超时才失败，且失败原因可读。 +func TestCloudAgentStepTimeoutRetriesThenFailsWithReadableReason(t *testing.T) { + s, db, root := reliableAgentRoot(t) + failStepTask(t, db, root.ID) + + if err := s.advanceCloudAgentByID("user", root.ID); err != nil { + t.Fatal(err) + } + run, state := agentInterjectionState(t, s, root.ID) + if state.StepTimeoutEscalated != 1 || !state.ForceThinkingOff { + t.Fatalf("超时后应关思考重试: escalated=%d forceThinkingOff=%v", state.StepTimeoutEscalated, state.ForceThinkingOff) + } + if run.Status != "running" { + t.Fatalf("超时应先重试而不是判死，实际 status=%s（%s）", run.Status, run.FailureMessage) + } + if !agentHasEventWithReason(state, "model_failure_recovered", "step_timeout_retried") { + t.Fatal("缺少可读的自动重试事件") + } + + // 重试那一步：必须真的关掉思考，并且带着生效的输出上限。 + if err := s.advanceCloudAgentByID("user", root.ID); err != nil { + t.Fatal(err) + } + _, retried := agentInterjectionState(t, s, root.ID) + if retried.ActiveTaskID == "" { + t.Fatal("重试没有重新发起模型调用") + } + task, err := s.repo.TaskForUser("user", retried.ActiveTaskID) + if err != nil { + t.Fatal(err) + } + options := stepTaskTextOptions(t, s, task) + if options["thinking"] != false { + t.Fatalf("重试请求没有关思考: %+v", options) + } + if options["maxOutputTokens"] != float64(platform.DefaultRuntimeAgentStepOutputTokens) { + t.Fatalf("重试请求的输出上限 = %v", options["maxOutputTokens"]) + } + + // 重试也超时 → 整轮失败，原因指向可配置的那条线。 + failStepTask(t, db, retried.ActiveTaskID) + if err := s.advanceCloudAgentByID("user", root.ID); err != nil { + t.Fatal(err) + } + run, state = agentInterjectionState(t, s, root.ID) + if run.Status != "failed" { + t.Fatalf("重试仍超时应结束本轮，实际 status=%s", run.Status) + } + if !strings.Contains(run.FailureMessage, "单步模型调用超过执行时限") { + t.Fatalf("失败文案没有解释单步超时: %s", run.FailureMessage) + } + if !agentHasEventWithReason(state, "run_failed", "model_step_timeout") { + t.Fatal("失败事件缺少 model_step_timeout 原因") + } +} + +func failStepTask(t *testing.T, db *gorm.DB, taskID string) { + t.Helper() + if err := db.Model(&model.Task{}).Where("id = ?", taskID).Updates(map[string]any{ + "status": model.TaskStatusFailed, + "error": cloudAgentStepTimeoutError + "，已中止这一步", + }).Error; err != nil { + t.Fatal(err) + } +} + +func stepTaskTextOptions(t *testing.T, s *Service, task *model.Task) map[string]any { + t.Helper() + decrypted, err := s.decryptTaskInputJSON(task.InputJSON) + if err != nil { + t.Fatal(err) + } + var input struct { + TextOptions map[string]any `json:"textOptions"` + } + if err := json.Unmarshal([]byte(decrypted), &input); err != nil { + t.Fatal(err) + } + if len(input.TextOptions) == 0 { + t.Fatalf("任务缺少 textOptions: %s", decrypted) + } + return input.TextOptions +} + +func agentHasEventWithReason(state cloudAgentRuntime, eventType, reason string) bool { + for _, event := range state.Events { + if event.Type == eventType && event.Payload["reason"] == reason { + return true + } + } + return false +} + +// agentEventValue 取最近一条指定类型事件里的某个字段值。 +func agentEventValue(state cloudAgentRuntime, eventType, key string) (any, bool) { + for i := len(state.Events) - 1; i >= 0; i-- { + if state.Events[i].Type != eventType { + continue + } + value, ok := state.Events[i].Payload[key] + return value, ok + } + return nil, false +} + +// 执行时限到点必须按失败收尾（而不是被"租约丢失"提前返回）。 +func TestAgentExecutionDeadlineCountsAsTaskFailure(t *testing.T) { + policy := defaultRuntimePolicy().Task + policy.AgentStepTimeoutSeconds = 30 + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + <-ctx.Done() + task := &model.Task{Type: "canvas_text", Operation: cloudAgentStepOperation} + if got := taskExecutionTimeout(task, policy); got != 30*time.Second { + t.Fatalf("timeout = %s", got) + } + if !errors.Is(ctx.Err(), context.DeadlineExceeded) { + t.Fatal("到期后 ctx 必须报告 DeadlineExceeded（worker 据此区分租约丢失）") + } +} diff --git a/backend/internal/app/provider.go b/backend/internal/app/provider.go index 54e104e07..cb83e8674 100644 --- a/backend/internal/app/provider.go +++ b/backend/internal/app/provider.go @@ -47,6 +47,10 @@ type canvasGenerationInput struct { type canvasTextOptions struct { Stream *bool `json:"stream"` Thinking bool `json:"thinking"` + // MaxOutputTokens 是本次调用的输出上限（思考 + 正文 + 工具参数）。 + // 画布 Agent 的每一步都带上限：不设时上游按"剩余上下文"放行，思考模型可以把单步 + // 拖到几分钟（实测 output_tokens 正好吃满可用预算、正文与工具调用皆空）；0 表示不限制。 + MaxOutputTokens int `json:"maxOutputTokens,omitempty"` } type agentToolRequests struct { diff --git a/backend/internal/app/provider_text.go b/backend/internal/app/provider_text.go index cb2235efb..28a4816a4 100644 --- a/backend/internal/app/provider_text.go +++ b/backend/internal/app/provider_text.go @@ -58,6 +58,7 @@ func runAgentToolTask(ctx context.Context, input canvasGenerationInput) (map[str } body["model"] = input.Config.Model applyTextThinking(body, input, protocol) + applyAgentOutputLimit(body, agentStepOutputLimit(input), protocol) normalizeAgentToolChoice(body, input, protocol) result, err := postAgentRequest(ctx, input, path, body, protocol) if protocol == "chat-completion" && isAgentToolChoiceCompatibilityError(err) { @@ -90,6 +91,39 @@ func postAgentRequest(ctx context.Context, input canvasGenerationInput, path str return parseAgentToolPayload(payload, protocol) } +// agentStepOutputLimit 取本次 Agent 调用的输出上限。 +// +// 两个来源语义相同但通道不同：textOptions.maxOutputTokens 是任务信封里下发的策略值 +// （画布 Agent 每步按运行时策略给，persist 在任务输入里，重启后仍在）；input.MaxOutputTokens +// 是进程内直传的旧入口（渠道模型能力声明）。都为 0 时不写上限字段，交给上游按剩余上下文放行。 +// 都非零时取较小值：策略上限不该超过模型自己声明的物理上限。 +func agentStepOutputLimit(input canvasGenerationInput) int { + limits := []int{input.TextOptions.MaxOutputTokens, input.MaxOutputTokens} + limit := 0 + for _, candidate := range limits { + if candidate <= 0 { + continue + } + if limit == 0 || candidate < limit { + limit = candidate + } + } + return limit +} + +// applyAgentOutputLimit 按协议写入输出上限字段名：Claude 与 Chat Completions 用 max_tokens， +// Responses 用 max_output_tokens。 +func applyAgentOutputLimit(body map[string]interface{}, limit int, protocol string) { + if limit <= 0 { + return + } + field := "max_tokens" + if protocol == "responses" { + field = "max_output_tokens" + } + applyTextOutputLimit(body, limit, field) +} + func runDeclarativeAgentTask(ctx context.Context, input canvasGenerationInput, adapter protocol.AgentAdapter) (map[string]interface{}, error) { wire := input.Config.InterfaceType if wire == string(model.ChannelInterfaceOpenAIResponse) { @@ -118,6 +152,7 @@ func runDeclarativeAgentTask(ctx context.Context, input canvasGenerationInput, a return nil, errors.New("声明式 Agent 请求体必须是 JSON 对象") } applyTextThinking(body, input, wire) + applyAgentOutputLimit(body, agentStepOutputLimit(input), wire) normalizeAgentToolChoice(body, input, wire) spec.Body = body if input.StreamText { diff --git a/backend/internal/app/task_worker.go b/backend/internal/app/task_worker.go index 528c03dc1..fde699684 100644 --- a/backend/internal/app/task_worker.go +++ b/backend/internal/app/task_worker.go @@ -137,7 +137,7 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot if err != nil { return err } - ctx, cancel := context.WithTimeout(context.Background(), taskExecutionTimeoutWithPolicy(task.Type, policy.Task)) + ctx, cancel := context.WithTimeout(context.Background(), taskExecutionTimeout(task, policy.Task)) defer cancel() leaseDone := make(chan struct{}) leaseLost := make(chan error, 1) @@ -227,11 +227,17 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot if code, _ := ChannelSlotFailureDetails(err); code != "" { channelSlotFailedBeforeRequest = true } - // 续租使用独立 context；即使执行同时超时，真实租约失效仍必须阻止写入。 + // 续租使用独立 context，真实租约失效必须阻止写入；但执行时限到点 + // （单步墙钟或任务超时）本身不是"租约丢失"：续租可能正好被同一个时限打断， + // 此时必须继续走失败收尾，否则任务停在 running 被反复重跑。 + deadlineExpired := errors.Is(ctx.Err(), context.DeadlineExceeded) select { case leaseErr := <-leaseLost: - _ = s.log(task.UserID, task.ID, "warn", "任务租约失效，等待其他 worker 恢复", leaseErr.Error()) - return leaseErr + if !deadlineExpired { + _ = s.log(task.UserID, task.ID, "warn", "任务租约失效，等待其他 worker 恢复", leaseErr.Error()) + return leaseErr + } + _ = s.log(task.UserID, task.ID, "warn", "任务执行时限到点，按失败收尾（不视为租约丢失）", leaseErr.Error()) default: } decryptedInput, decryptErr := s.decryptTaskInputJSON(task.InputJSON) @@ -252,8 +258,14 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot if newAPIChannel2TaskSyncExpired(*task, err, time.Now()) { err = errors.New("上游任务长时间未同步，已停止自动查询，请确认渠道任务状态后重试。") } - if errors.Is(err, context.DeadlineExceeded) { - err = errors.New(taskTimeoutMessage(task.Type)) + if errors.Is(err, context.DeadlineExceeded) || deadlineExpired { + // 画布 Agent 的单步超时是可恢复事件（运行期会关思考重试同一步）， + // 因此必须与"任务执行超时"区分开，否则只能整轮判死。 + if cloudAgentModelOperation(task) { + err = errors.New(cloudAgentStepTimeoutError + "，已中止这一步") + } else { + err = errors.New(taskTimeoutMessage(task.Type)) + } } s.noteAgentMemoryCompactTask(*task, nil, err) return terminal.handleExecutionFailure(task, err, providerSucceeded, channelSlotFailedBeforeRequest) @@ -307,6 +319,21 @@ func taskLeaseRenewContext(parent context.Context) (context.Context, context.Can return context.WithTimeout(context.WithoutCancel(parent), 5*time.Second) } +// taskExecutionTimeout 解析一次任务的执行墙钟：画布 Agent 的单步调用可以配秒级超时 +// （AgentStepTimeoutSeconds），没配时沿用文本任务超时。秒级粒度是必要的——一轮里每一步 +// 都是分钟级的调用，分钟粒度改不动"某一步卡住"的体验。 +// 超时的表现是任务错误里带 cloudAgentStepTimeoutError 标记，运行期据此关思考重试同一步， +// 而不是把整轮判死（见 cloud_agent_step_timeout.go）。 +func taskExecutionTimeout(task *model.Task, policy RuntimeTaskPolicy) time.Duration { + if task != nil && cloudAgentModelOperation(task) && policy.AgentStepTimeoutSeconds > 0 { + return time.Duration(policy.AgentStepTimeoutSeconds) * time.Second + } + if task == nil { + return time.Duration(policy.DefaultTimeoutMinutes) * time.Minute + } + return taskExecutionTimeoutWithPolicy(task.Type, policy) +} + func taskExecutionTimeoutWithPolicy(taskType string, policy RuntimeTaskPolicy) time.Duration { switch { case strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_"): diff --git a/backend/internal/platform/runtime_policy.go b/backend/internal/platform/runtime_policy.go index 7e5514dc6..f94408123 100644 --- a/backend/internal/platform/runtime_policy.go +++ b/backend/internal/platform/runtime_policy.go @@ -26,6 +26,30 @@ const ( maxRuntimeTimeoutMinutes = 9_999 ) +// 画布 Agent 单步的输出上限与执行时限。上限是"每次模型调用"的边界（思考 + 正文 + 工具参数）， +// 时限是"这一次调用最多等多久"的墙钟；两者一起决定单步最坏耗时。0 在两个口径下都是"该边界不生效"： +// 上限为 0 表示不限制输出（只由执行时限兜底），时限为 0 表示沿用文本任务超时。 +const ( + MinRuntimeAgentStepOutputTokens = 256 + MaxRuntimeAgentStepOutputTokens = 131_072 + DefaultRuntimeAgentStepOutputTokens = 16_384 + MinRuntimeAgentStepTimeoutSeconds = 30 + MaxRuntimeAgentStepTimeoutSeconds = 3_600 + DefaultRuntimeAgentStepTimeout = 0 +) + +// agentStepMaxOutputTokensDefault 读取部署级覆盖并按同一套范围夹取；环境变量无法表达"不限制"（0）， +// 那个档位只在管理端配置里设置。 +func agentStepMaxOutputTokensDefault() int { + value := envInt("CANVAS_AGENT_STEP_MAX_OUTPUT_TOKENS", DefaultRuntimeAgentStepOutputTokens) + return min(max(value, MinRuntimeAgentStepOutputTokens), MaxRuntimeAgentStepOutputTokens) +} + +func agentStepTimeoutSecondsDefault() int { + value := envInt("CANVAS_AGENT_STEP_TIMEOUT_SECONDS", DefaultRuntimeAgentStepTimeout) + return min(value, MaxRuntimeAgentStepTimeoutSeconds) +} + type RuntimeResourcePolicy struct { ResourceUploadMB int64 `json:"resourceUploadMB"` GeneratedFileMB int64 `json:"generatedFileMB"` @@ -50,6 +74,11 @@ type RuntimeTaskPolicy struct { VideoTimeoutMinutes int `json:"videoTimeoutMinutes"` StoryboardTimeoutMinutes int `json:"storyboardTimeoutMinutes"` DefaultTimeoutMinutes int `json:"defaultTimeoutMinutes"` + // AgentStepMaxOutputTokens 是画布 Agent 单步模型调用的输出上限（思考 + 正文 + 工具调用参数）。 + // 0 表示不限制输出，此时单步最坏耗时只由 AgentStepTimeoutSeconds 兜底。 + AgentStepMaxOutputTokens int `json:"agentStepMaxOutputTokens"` + // AgentStepTimeoutSeconds 是画布 Agent 单步模型调用的秒级墙钟；0 表示沿用文本任务超时。 + AgentStepTimeoutSeconds int `json:"agentStepTimeoutSeconds"` } type RuntimeRequestPolicy struct { @@ -123,6 +152,8 @@ func DefaultRuntimePolicy() RuntimePolicySetting { VideoTimeoutMinutes: 60, StoryboardTimeoutMinutes: 20, DefaultTimeoutMinutes: 10, + AgentStepMaxOutputTokens: agentStepMaxOutputTokensDefault(), + AgentStepTimeoutSeconds: agentStepTimeoutSecondsDefault(), }, Request: RuntimeRequestPolicy{ TaskCreatePerMinute: 30, @@ -162,6 +193,7 @@ func selfUseRuntimePolicy() RuntimePolicySetting { ImageTimeoutMinutes: maxRuntimeTimeoutMinutes, TextTimeoutMinutes: maxRuntimeTimeoutMinutes, AudioTimeoutMinutes: maxRuntimeTimeoutMinutes, VideoTimeoutMinutes: maxRuntimeTimeoutMinutes, StoryboardTimeoutMinutes: maxRuntimeTimeoutMinutes, DefaultTimeoutMinutes: maxRuntimeTimeoutMinutes, + AgentStepMaxOutputTokens: MaxRuntimeAgentStepOutputTokens, AgentStepTimeoutSeconds: MaxRuntimeAgentStepTimeoutSeconds, } value.Request = RuntimeRequestPolicy{ TaskCreatePerMinute: maxRuntimeRate, @@ -345,6 +377,12 @@ func validateRuntimePolicy(value RuntimePolicySetting) error { return kernel.BadAuthRequest(fmt.Sprintf("%s必须是 1-%d 分钟的整数", label, maxRuntimeTimeoutMinutes)) } } + if task.AgentStepMaxOutputTokens != 0 && (task.AgentStepMaxOutputTokens < MinRuntimeAgentStepOutputTokens || task.AgentStepMaxOutputTokens > MaxRuntimeAgentStepOutputTokens) { + return kernel.BadAuthRequest(fmt.Sprintf("Agent 单步输出上限必须是 0 或 %d-%d 的整数 (0 表示不限制)", MinRuntimeAgentStepOutputTokens, MaxRuntimeAgentStepOutputTokens)) + } + if task.AgentStepTimeoutSeconds != 0 && (task.AgentStepTimeoutSeconds < MinRuntimeAgentStepTimeoutSeconds || task.AgentStepTimeoutSeconds > MaxRuntimeAgentStepTimeoutSeconds) { + return kernel.BadAuthRequest(fmt.Sprintf("Agent 单步超时必须是 0 或 %d-%d 秒的整数 (0 表示沿用文本任务超时)", MinRuntimeAgentStepTimeoutSeconds, MaxRuntimeAgentStepTimeoutSeconds)) + } request := value.Request for label, item := range map[string]int{ "任务创建频控": request.TaskCreatePerMinute, diff --git a/web/src/pages/admin/settings/runtime-policy-settings-page.tsx b/web/src/pages/admin/settings/runtime-policy-settings-page.tsx index 19ed85ad2..845cd6c55 100644 --- a/web/src/pages/admin/settings/runtime-policy-settings-page.tsx +++ b/web/src/pages/admin/settings/runtime-policy-settings-page.tsx @@ -1,5 +1,5 @@ import { App, Button, Form, InputNumber, Skeleton } from "antd"; -import { AlertTriangle, Database, Gauge, Infinity as InfinityIcon, Network, RefreshCw, RotateCcw, Save, ShieldCheck, TimerReset } from "lucide-react"; +import { AlertTriangle, Bot, Database, Gauge, Infinity as InfinityIcon, Network, RefreshCw, RotateCcw, Save, ShieldCheck, TimerReset } from "lucide-react"; import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from "react"; import { useBlocker } from "react-router"; @@ -51,6 +51,27 @@ const timeoutFields: PolicyField[] = [ { group: "task", name: "defaultTimeoutMinutes", label: "默认任务超时", extra: "未匹配专用类型时使用的最长执行时间。", unit: "分钟", max: 9_999 }, ]; +const agentFields: PolicyField[] = [ + { + group: "task", + name: "agentStepMaxOutputTokens", + label: "单步输出上限", + extra: "画布 Agent 每次模型调用的输出上限，含思考、正文与工具调用参数。0 表示不限制，此时只有单步超时兜底；抬高可以避免长思考模型被截断后返回空内容。", + unit: "token", + min: 0, + max: 131_072, + }, + { + group: "task", + name: "agentStepTimeoutSeconds", + label: "单步超时", + extra: "画布 Agent 单步模型调用的最长等待时间，到点会中止这一步并自动关思考重试一次。0 表示沿用“文本任务超时”，不再单独计时。", + unit: "秒", + min: 0, + max: 3_600, + }, +]; + const rateFields: PolicyField[] = [ { group: "request", name: "taskCreatePerMinute", label: "任务创建", extra: "每账号每分钟允许创建的任务数。", unit: "次/分钟", max: 999_999 }, { group: "request", name: "resourceUploadPerMinute", label: "资源上传", extra: "每账号每分钟上传资源的次数。", unit: "次/分钟", max: 999_999 }, @@ -88,6 +109,15 @@ const policySections: PolicySectionDefinition[] = [ status: , }, { id: "policy-timeout", icon: