From 6f09b313d674482987a56e62ecbd8820f5dd74de Mon Sep 17 00:00:00 2001 From: RuoJi6 Date: Fri, 23 Oct 2026 19:14:35 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20=E5=A4=9A=E6=A8=A1=E5=9E=8B=20Worke?= =?UTF-8?q?r=20=E6=B1=A0=EF=BC=8C=E6=94=AF=E6=8C=81=E6=8C=89=20LLM=20?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=88=86=E9=85=8D=E5=B9=B6=E5=8F=91=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现 #41：系统配置可设置 worker_pools，使本地模型与云端模型等各占用若干 Work Agent 会话并在同一任务内并行执行。空配置保持原 workers 遗留行为。 --- .gitignore | 1 + server/engine.go | 34 ++++- server/server.go | 37 ++++- server/task_llm.go | 91 ++++++++++-- server/worker_pools.go | 127 ++++++++++++++++ server/worker_pools_test.go | 86 +++++++++++ web/src/app/(main)/system/settings/page.tsx | 154 ++++++++++++++++++-- web/src/lib/mock/data.ts | 1 + web/src/lib/types.ts | 2 + 9 files changed, 499 insertions(+), 34 deletions(-) create mode 100644 server/worker_pools.go create mode 100644 server/worker_pools_test.go diff --git a/.gitignore b/.gitignore index c9a7436..96def82 100644 --- a/.gitignore +++ b/.gitignore @@ -60,3 +60,4 @@ Thumbs.db /bench/ /Dockerfile.bench /Dockerfile.bench.v2 +.plan/ diff --git a/server/engine.go b/server/engine.go index b5643b2..e99cf17 100644 --- a/server/engine.go +++ b/server/engine.go @@ -148,6 +148,11 @@ type Engine struct { // (for example an exhausted failover chain) — there is no global-pair fallback. resolve func(t *Task) (*agent.Planner, *agent.Worker) resolveAuthoritative bool + // resolveWorker returns the Work Agent for one spawn slot. profileID 0 → unbound + // task-router worker (legacy); non-zero → worker pinned to that LLM profile. + // For profileID 0, a nil result falls back to snapshotFor; for pinned slots, + // nil means wait/retry (never switch to another model). + resolveWorker func(t *Task, profileID int64) *agent.Worker // readiness reports whether a global LLM provider is configured — the signal behind // Ready()/the llm_configured indicator. Wired once at startup; nil → not ready. readiness func() bool @@ -715,6 +720,12 @@ func (e *Engine) SetAuthoritativeAgentResolver(fn func(t *Task) (*agent.Planner, e.resolveAuthoritative = true } +// SetWorkerResolver installs the per-slot worker factory used by workerLoop when +// multi-model pools assign a profile to a session (work#N). Called once at startup. +func (e *Engine) SetWorkerResolver(fn func(t *Task, profileID int64) *agent.Worker) { + e.resolveWorker = fn +} + // snapshotFor returns the planner/worker a task should run on, from the task-router // resolver. nil,nil means the task is deliberately unavailable (e.g. an exhausted // failover chain); there is no global-pair fallback. @@ -746,7 +757,7 @@ func (e *Engine) ReadyFor(t *Task) bool { // but no-op until an LLM is configured (so a task created while idle picks up // automatically once LLM is set from the UI). func (e *Engine) Run(ctx context.Context, t *Task) { - workers := e.m.Workers() + slots := e.m.WorkerSlots() e.deleteMu.RLock() if e.IsDeleting(t.ID) { e.deleteMu.RUnlock() @@ -757,13 +768,13 @@ func (e *Engine) Run(ctx context.Context, t *Task) { t.Notify() // already running — just nudge a planning round return } - rt := e.registerTaskRoutines(ctx, t.ID, 1+workers) + rt := e.registerTaskRoutines(ctx, t.ID, 1+len(slots)) e.deleteMu.RUnlock() e.touch(t.ID) runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) }) - for i := 0; i < workers; i++ { - name := fmt.Sprintf("work#%d", i+1) - runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name) }) + for _, slot := range slots { + name, profileID := slot.Name, slot.ProfileID + runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name, profileID) }) } e.startDeadlineCoordinator(ctx, t) // 任务级超时定时器(仅 timeout>0;去重) // 仅在「完全没有活动意图(open+running)」时才 kick 首轮规划。带种子意图的任务:种子已 @@ -913,14 +924,23 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) { } } -func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { +func (e *Engine) workerLoop(ctx context.Context, t *Task, name string, profileID int64) { for { select { case <-ctx.Done(): return default: } - _, worker := e.snapshotFor(t) + var worker *agent.Worker + if e.resolveWorker != nil { + worker = e.resolveWorker(t, profileID) + } + // Legacy unbound slots (profileID 0) may fall back to the task-router + // snapshot. Pinned multi-model pool slots must NOT silently switch models + // when the assigned profile is missing/unavailable — wait and retry. + if worker == nil && profileID == 0 { + _, worker = e.snapshotFor(t) + } if worker == nil { if sleepCtx(ctx, 1500*time.Millisecond) { return diff --git a/server/server.go b/server/server.go index 969216c..ec51b8a 100644 --- a/server/server.go +++ b/server/server.go @@ -177,6 +177,11 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi b := s.agentsForTask(t) return b.pl, b.wk }) + // Multi-model Worker pools (#41): each spawn slot may pin a Work Agent to a + // specific LLM profile. profileID 0 keeps the unbound task-router worker. + s.engine.SetWorkerResolver(func(t *Task, profileID int64) *agent.Worker { + return s.workerForTask(t, profileID) + }) // Global readiness (Ready()/llm_configured): a global active LLM provider is // installed. Task-level runnability is separate (ReadyFor → the resolver above). s.engine.SetReadiness(func() bool { @@ -3432,6 +3437,7 @@ func (s *Server) settingsPayload() map[string]any { "global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5)，所有目标流量走它，空=直连 "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 + "worker_pools": workerPoolsPayload(s.m), // 多模型 Worker 池；空=走 workers 遗留模式 "task_concurrency_enabled": concOn, // 任务并发上限开关(默认关) "task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5) // LLM 轮询(故障转移)。默认关；开启后走全局激活配置的 agent 在当前配置不可用时 @@ -3452,6 +3458,18 @@ func (s *Server) settingsPayload() map[string]any { } } + +// workerPoolsPayload always returns a non-nil slice for JSON (empty = legacy mode). +func workerPoolsPayload(m *Manager) []WorkerPoolConfig { + if m == nil { + return []WorkerPoolConfig{} + } + if pools := m.WorkerPools(); pools != nil { + return pools + } + return []WorkerPoolConfig{} +} + // notifyPublicBaseURL 读推送回链用的外部地址。 func notifyPublicBaseURL(pg *db.DB) string { v, _, _ := pg.GetSetting(settingNotifyPublicBaseURL) @@ -3506,7 +3524,8 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空 GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连) PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径 - Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 + Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 + WorkerPools *[]WorkerPoolConfig `json:"worker_pools"` // 多模型池；null=不改，[]=清空回遗留模式 // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 ConcurrencyEnabled *bool `json:"task_concurrency_enabled"` ConcurrencyLimit *int `json:"task_concurrency_limit"` @@ -3585,6 +3604,22 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { return } } + if req.WorkerPools != nil { + if err := s.m.SetWorkerPools(*req.WorkerPools); err != nil { + writeErr(w, 400, err.Error()) + return + } + // 有池配置时把 workers 同步成合计,方便 UI/遗留读数与真实并发一致。 + if pools := *req.WorkerPools; len(pools) > 0 { + total := 0 + for _, p := range pools { + total += p.Workers + } + if total > 0 { + _ = s.m.SetWorkers(total) + } + } + } if req.ConcurrencyEnabled != nil || req.ConcurrencyLimit != nil { // 部分 PUT:未给的字段用当前值兜底,避免只改一个把另一个重置。 curOn, curLimit := s.m.ConcurrencyLimit() diff --git a/server/task_llm.go b/server/task_llm.go index 2cae9ce..48a086f 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -27,6 +27,9 @@ type taskAgentBundle struct { pl *agent.Planner wk *agent.Worker main *agent.MainAgent + // workersByProfile caches Work Agents pinned to a specific LLM profile for + // multi-model pools (issue #41). Key = llm_profiles.id. + workersByProfile map[int64]*agent.Worker } type llmAuditProfile struct { @@ -51,6 +54,11 @@ type taskLLMRuntime struct { s *Server taskID string agentKey string + // pinProfileID, when non-zero, forces every LLM call onto that profile and + // skips agent-binding / task-chain / global resolution. Used by multi-model + // Worker pools so each session (work#N) stays on its assigned model. Quota + // errors do NOT advance the task failover chain (sel.profileID stays 0). + pinProfileID int64 } type taskLLMError struct { @@ -122,6 +130,15 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) { r.s.syncTaskLLMState(pt) t, _ := r.s.m.Task(r.taskID) sel := taskLLMSelection{task: t, revision: pt.LLMChainRevision} + if r.pinProfileID != 0 { + prov, cfg, ok := r.s.providerForProfile(r.pinProfileID) + if !ok { + return sel, fmt.Errorf("pinned LLM profile #%d is missing or invalid", r.pinProfileID) + } + // profileID left 0 so streamTaskLLM/completeTaskLLM never mutate the task chain. + sel.provider, sel.retry = r.s.poolForBinding(r.pinProfileID, prov, cfg), cfg.Retry + return sel, nil + } if prov, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { sel.provider, sel.retry = prov, cfg.Retry return sel, nil @@ -152,6 +169,12 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) { // the caller. If failover switches profiles, the change takes effect on the next // agent run (a fresh Session is built per run in captureRun). func (r *taskLLMRuntime) activeCfg() (agent.Config, bool) { + if r.pinProfileID != 0 { + if _, cfg, ok := r.s.providerForProfile(r.pinProfileID); ok { + return cfg, true + } + return agent.Config{}, false + } taskNum, err := parseTaskID(r.taskID) if err != nil { return agent.Config{}, false @@ -454,6 +477,12 @@ func isRetryableStreamError(err error) bool { // CompactionWindow mirrors current()'s precedence so the context window always // matches the provider the role will actually stream on. func (r *taskLLMRuntime) CompactionWindow() int { + if r.pinProfileID != 0 { + if _, cfg, ok := r.s.providerForProfile(r.pinProfileID); ok { + return cfg.CompactionWindow() + } + return (agent.Config{}).CompactionWindow() + } if _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { return cfg.CompactionWindow() } @@ -560,17 +589,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { workerRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "worker"} mainRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "mainagent"} tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) - window := workerRuntime.CompactionWindow() - wk := agent.NewWorker(workerRuntime, "task-router", s.m.dir, tx, window, s.agentMaxTurns("worker")) - wk.SetFindingRecorder(s.evidenceStore()) - wk.SetCompactionWindowResolver(workerRuntime.CompactionWindow) - wk.SetNonStreaming(workerRuntime.nonStreaming) // 按任务当前激活 profile 的流式开关(每轮读) - wk.SetMaxTokens(workerRuntime.maxTokens) // 同上,输出上限也跟随当前激活 profile - wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读) - wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) - wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) - wk.SetWebSearch(s.webSearchFor("worker")) - wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开) + wk := s.newTaskWorker(workerRuntime, tx) pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner")) pl.SetFindingRecorder(s.evidenceStore()) pl.SetCompactionWindowResolver(plannerRuntime.CompactionWindow) @@ -598,11 +617,61 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { bundle := &taskAgentBundle{ runtime: goalRuntime, plannerRuntime: plannerRuntime, workerRuntime: workerRuntime, mainRuntime: mainRuntime, pl: pl, wk: wk, main: main, + workersByProfile: map[int64]*agent.Worker{}, } s.taskAgents[t.ID] = bundle return bundle } + +// newTaskWorker builds a Work Agent on the given runtime (shared wiring for the +// unbound task worker and per-profile pinned workers). +func (s *Server) newTaskWorker(runtime *taskLLMRuntime, tx *transcript.Store) *agent.Worker { + wk := agent.NewWorker(runtime, "task-router", s.m.dir, tx, runtime.CompactionWindow(), s.agentMaxTurns("worker")) + wk.SetFindingRecorder(s.evidenceStore()) + wk.SetCompactionWindowResolver(runtime.CompactionWindow) + wk.SetNonStreaming(runtime.nonStreaming) + wk.SetMaxTokens(runtime.maxTokens) + wk.SetNoaEnabled(s.m.NoaCompactionEnabled) + wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) + wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) + wk.SetWebSearch(s.webSearchFor("worker")) + wk.SetConstraintInject(s.constraintInjectWorker) + return wk +} + +// workerForTask returns the Work Agent for one Engine spawn slot. +// profileID 0 → unbound task-router worker (requires taskRuntimeAvailable). +// Non-zero → worker pinned to that LLM profile (must build; ignores task-chain exhaustion). +func (s *Server) workerForTask(t *Task, profileID int64) *agent.Worker { + if t == nil { + return nil + } + if profileID == 0 { + if !s.taskRuntimeAvailable(t, "worker") { + return nil + } + return s.agentsForTask(t).wk + } + if _, _, ok := s.providerForProfile(profileID); !ok { + return nil + } + bundle := s.agentsForTask(t) + s.taskAgentMu.Lock() + defer s.taskAgentMu.Unlock() + if bundle.workersByProfile == nil { + bundle.workersByProfile = map[int64]*agent.Worker{} + } + if w := bundle.workersByProfile[profileID]; w != nil { + return w + } + tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) + runtime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "worker", pinProfileID: profileID} + w := s.newTaskWorker(runtime, tx) + bundle.workersByProfile[profileID] = w + return w +} + func (s *Server) syncTaskLLMState(pt *db.Task) { if pt == nil { return diff --git a/server/worker_pools.go b/server/worker_pools.go new file mode 100644 index 0000000..b66c691 --- /dev/null +++ b/server/worker_pools.go @@ -0,0 +1,127 @@ +package server + +import ( + "encoding/json" + "fmt" + "strings" +) + +// settingWorkerPools stores the multi-model Work Agent pool config as a JSON array. +// Empty / unset → legacy single-pool mode using settings.workers + unbound runtime. +const settingWorkerPools = "worker_pools" + +// maxWorkerPoolTotal is the hard upper bound on sum(pool.workers), matching the +// per-task override ceiling used elsewhere so a misconfigured UI can't spawn dozens +// of LLM sessions per task. +const maxWorkerPoolTotal = 50 + +// WorkerPoolConfig is one LLM profile's share of concurrent Work Agents. +type WorkerPoolConfig struct { + ProfileID int64 `json:"profile_id"` + Workers int `json:"workers"` +} + +// WorkerSlot is one Engine.Run worker loop: Name is the claim/session id (work#N); +// ProfileID 0 means unbound (legacy task-router resolution). +type WorkerSlot struct { + Name string + ProfileID int64 +} + +// validateWorkerPools checks a pools slice before persistence. Empty is allowed +// (clears multi-model mode). Non-empty requires every entry valid and total in 1..50. +func validateWorkerPools(pools []WorkerPoolConfig) error { + if len(pools) == 0 { + return nil + } + total := 0 + for i, p := range pools { + if p.ProfileID <= 0 { + return fmt.Errorf("worker_pools[%d].profile_id 必须 >0", i) + } + if p.Workers <= 0 { + return fmt.Errorf("worker_pools[%d].workers 必须 >0", i) + } + total += p.Workers + } + if total > maxWorkerPoolTotal { + return fmt.Errorf("worker_pools 合计 Worker 数最多 %d（当前 %d）", maxWorkerPoolTotal, total) + } + return nil +} + +// expandWorkerSlots turns pool configs into a spawn plan. Empty pools → legacy: +// N unbound slots named work#1..N where N = legacyWorkers (clamped to >=1). +func expandWorkerSlots(pools []WorkerPoolConfig, legacyWorkers int) []WorkerSlot { + if len(pools) == 0 { + n := legacyWorkers + if n <= 0 { + n = defaultWorkers + } + out := make([]WorkerSlot, n) + for i := 0; i < n; i++ { + out[i] = WorkerSlot{Name: fmt.Sprintf("work#%d", i+1), ProfileID: 0} + } + return out + } + var out []WorkerSlot + for _, p := range pools { + if p.ProfileID <= 0 || p.Workers <= 0 { + continue + } + for i := 0; i < p.Workers; i++ { + out = append(out, WorkerSlot{ + Name: fmt.Sprintf("work#%d", len(out)+1), + ProfileID: p.ProfileID, + }) + } + } + if len(out) == 0 { + return expandWorkerSlots(nil, legacyWorkers) + } + return out +} + +// WorkerPools returns the configured multi-model pools (nil/empty = legacy mode). +// Invalid entries in the stored JSON are skipped. +func (m *Manager) WorkerPools() []WorkerPoolConfig { + if m == nil || m.pg == nil { + return nil + } + v, ok, err := m.pg.GetSetting(settingWorkerPools) + if err != nil || !ok || strings.TrimSpace(v) == "" { + return nil + } + var raw []WorkerPoolConfig + if json.Unmarshal([]byte(v), &raw) != nil { + return nil + } + out := make([]WorkerPoolConfig, 0, len(raw)) + for _, p := range raw { + if p.ProfileID > 0 && p.Workers > 0 { + out = append(out, p) + } + } + return out +} + +// SetWorkerPools persists the multi-model pool config. An empty slice clears the +// setting (back to legacy single-pool Workers()). Applies to tasks started afterwards. +func (m *Manager) SetWorkerPools(pools []WorkerPoolConfig) error { + if err := validateWorkerPools(pools); err != nil { + return err + } + if len(pools) == 0 { + return m.pg.SetSetting(settingWorkerPools, "") + } + b, err := json.Marshal(pools) + if err != nil { + return err + } + return m.pg.SetSetting(settingWorkerPools, string(b)) +} + +// WorkerSlots returns the expanded spawn plan for Engine.Run (read per task start). +func (m *Manager) WorkerSlots() []WorkerSlot { + return expandWorkerSlots(m.WorkerPools(), m.Workers()) +} diff --git a/server/worker_pools_test.go b/server/worker_pools_test.go new file mode 100644 index 0000000..78ca527 --- /dev/null +++ b/server/worker_pools_test.go @@ -0,0 +1,86 @@ +package server + +import ( + "testing" +) + +func TestExpandWorkerSlotsLegacy(t *testing.T) { + slots := expandWorkerSlots(nil, 3) + if len(slots) != 3 { + t.Fatalf("len=%d want 3", len(slots)) + } + for i, s := range slots { + want := "work#" + itoa(i+1) + if s.Name != want || s.ProfileID != 0 { + t.Fatalf("slot[%d]=%+v want name=%s profile=0", i, s, want) + } + } + slots = expandWorkerSlots(nil, 0) + if len(slots) != defaultWorkers { + t.Fatalf("legacy 0 → defaultWorkers: got %d", len(slots)) + } +} + +func TestExpandWorkerSlotsMulti(t *testing.T) { + slots := expandWorkerSlots([]WorkerPoolConfig{ + {ProfileID: 1, Workers: 2}, + {ProfileID: 2, Workers: 3}, + }, 3) + if len(slots) != 5 { + t.Fatalf("len=%d want 5", len(slots)) + } + want := []int64{1, 1, 2, 2, 2} + for i, s := range slots { + if s.ProfileID != want[i] || s.Name != "work#"+itoa(i+1) { + t.Fatalf("slot[%d]=%+v want profile=%d", i, s, want[i]) + } + } +} + +func TestExpandWorkerSlotsSkipsInvalidFallsBack(t *testing.T) { + slots := expandWorkerSlots([]WorkerPoolConfig{ + {ProfileID: 0, Workers: 2}, + {ProfileID: 1, Workers: 0}, + }, 4) + if len(slots) != 4 || slots[0].ProfileID != 0 { + t.Fatalf("all-invalid pools should fall back to legacy: %+v", slots) + } +} + +func TestValidateWorkerPools(t *testing.T) { + if err := validateWorkerPools(nil); err != nil { + t.Fatal(err) + } + if err := validateWorkerPools([]WorkerPoolConfig{}); err != nil { + t.Fatal(err) + } + if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 0, Workers: 1}}); err == nil { + t.Fatal("expected profile_id error") + } + if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 1, Workers: 0}}); err == nil { + t.Fatal("expected workers error") + } + big := make([]WorkerPoolConfig, 0, 3) + big = append(big, WorkerPoolConfig{ProfileID: 1, Workers: 40}) + big = append(big, WorkerPoolConfig{ProfileID: 2, Workers: 20}) + if err := validateWorkerPools(big); err == nil { + t.Fatal("expected sum>50 error") + } + if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 1, Workers: 3}, {ProfileID: 2, Workers: 3}}); err != nil { + t.Fatal(err) + } +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + var b [16]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index fea869f..50420b8 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -2,7 +2,7 @@ import * as React from "react"; -import { CpuIcon, FlaskConicalIcon, KeyboardIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon } from "lucide-react"; +import { CpuIcon, FlaskConicalIcon, KeyboardIcon, PlusIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon, Trash2Icon } from "lucide-react"; import { toast } from "sonner"; import { Button } from "@/components/ui/button"; @@ -14,7 +14,7 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@ import { Switch } from "@/components/ui/switch"; import { api } from "@/lib/api"; import { CHAT_SEND_MODE_OPTIONS, type ChatSendMode, setChatSendMode, useChatSendMode } from "@/lib/chat-send-mode"; -import type { Settings } from "@/lib/types"; +import type { LLMProfile, Settings } from "@/lib/types"; import { BasicAuthCard } from "./_components/basic-auth-card"; import { UpdateCard } from "./_components/update-card"; @@ -45,6 +45,8 @@ export default function SystemSettingsPage() { const [pyInterp, setPyInterp] = React.useState(""); const [workers, setWorkers] = React.useState("3"); const [savingWorkers, setSavingWorkers] = React.useState(false); + const [llmProfiles, setLlmProfiles] = React.useState([]); + const [workerPools, setWorkerPools] = React.useState<{ profile_id: string; workers: string }[]>([]); // 操作约束注入范围(默认都开)。 const [injectPlanner, setInjectPlanner] = React.useState(true); const [injectWorker, setInjectWorker] = React.useState(true); @@ -67,23 +69,53 @@ export default function SystemSettingsPage() { setGlobalProxyInput(s.global_proxy ?? ""); setPyInterp(s.python_interpreter ?? ""); setWorkers(String(s.workers ?? 3)); + setWorkerPools( + (s.worker_pools ?? []).map((p) => ({ + profile_id: String(p.profile_id), + workers: String(p.workers), + })), + ); setInjectPlanner(s.constraints_inject_planner !== false); setInjectWorker(s.constraints_inject_worker !== false); setNoaCompaction(!!s.noa_compaction); }, []); const saveWorkers = () => { - const n = Number(workers); - if (!Number.isInteger(n) || n <= 0) { - toast.error("并发数必须是大于 0 的整数"); + const pools: { profile_id: number; workers: number }[] = []; + for (let i = 0; i < workerPools.length; i++) { + const row = workerPools[i]; + const pid = Number(row.profile_id); + const n = Number(row.workers); + if (!Number.isInteger(pid) || pid <= 0) { + toast.error(`第 ${i + 1} 行：请选择 LLM 配置`); + return; + } + if (!Number.isInteger(n) || n <= 0) { + toast.error(`第 ${i + 1} 行：Worker 数必须是大于 0 的整数`); + return; + } + pools.push({ profile_id: pid, workers: n }); + } + const legacy = Number(workers); + if (pools.length === 0 && (!Number.isInteger(legacy) || legacy <= 0)) { + toast.error("默认并发数必须是大于 0 的整数"); + return; + } + const total = pools.length > 0 ? pools.reduce((s, p) => s + p.workers, 0) : legacy; + if (total > 50) { + toast.error("Worker 合计最多 50"); return; } setSavingWorkers(true); api - .setSettings({ workers: n }) + .setSettings({ workers: total, worker_pools: pools }) .then((s) => { apply(s); - toast.success("已保存并发工作 agent 数（对之后启动的任务生效）"); + toast.success( + pools.length > 0 + ? `已保存多模型 Worker 池（共 ${total} 个会话；对之后启动的任务生效）` + : "已保存并发工作 agent 数（对之后启动的任务生效）", + ); }) .catch((e) => toast.error("保存失败：" + (e as Error).message)) .finally(() => setSavingWorkers(false)); @@ -110,9 +142,11 @@ export default function SystemSettingsPage() { }; React.useEffect(() => { - api - .settings() - .then(apply) + Promise.all([api.settings(), api.llmProfiles().catch(() => [] as LLMProfile[])]) + .then(([s, profiles]) => { + apply(s); + setLlmProfiles(profiles); + }) .catch(() => undefined) .finally(() => setLoaded(true)); }, [apply]); @@ -748,24 +782,114 @@ export default function SystemSettingsPage() { - 工作并发 · Work Agent 数 + 工作并发 · Work Agent / 多模型池 - 每个任务并发运行的工作 agent 数量（默认 3）。数值越大并发探测越多、消耗也越高。修改后 + 可为多个 LLM 配置各分配若干 Worker；同一任务内多模型并行执行，按会话（work#N）绑定模型。 + 未配置模型池时，使用下方「默认并发」走单一模型（与历史行为一致）。修改后 对之后启动的任务生效，正在运行的任务不受影响。 - -
+ +
+ 0} onChange={(e) => setWorkers(e.target.value)} /> + {workerPools.length > 0 ? ( +

已配置模型池时以各池合计为准，默认并发输入已锁定。

+ ) : null} +
+ +
+
+ + +
+ {llmProfiles.length === 0 ? ( +

尚未配置 LLM：请先到「系统 → LLM 配置」添加并激活。

+ ) : null} + {workerPools.map((row, idx) => ( +
+ + + setWorkerPools((rows) => + rows.map((r, i) => (i === idx ? { ...r, workers: e.target.value } : r)), + ) + } + /> + Workers + +
+ ))} +

+ 当前合计： + {workerPools.length > 0 + ? workerPools.reduce((s, r) => s + (Number(r.workers) || 0), 0) + : Number(workers) || 0}{" "} + 个 Worker 会话 +

+
+ +
diff --git a/web/src/lib/mock/data.ts b/web/src/lib/mock/data.ts index 531df7b..9b8a69c 100644 --- a/web/src/lib/mock/data.ts +++ b/web/src/lib/mock/data.ts @@ -3338,6 +3338,7 @@ export const settings: Settings = { global_proxy: "", python_interpreter: "/usr/bin/python3", workers: 3, + worker_pools: [], llm_pool_enabled: true, llm_pool_bind_fallback: false, noa_compaction: false, diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 54e26a6..5a7188d 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -968,6 +968,8 @@ export interface Settings { global_proxy?: string; python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测) workers?: number; // 并发工作 agent 数(默认3)；对之后启动的任务生效 + // 多模型 Worker 池：每个 LLM 配置各自占用若干并发会话。空/缺省=走上面的 workers 遗留模式。 + worker_pools?: { profile_id: number; workers: number }[]; // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 task_concurrency_enabled?: boolean; // 默认 false task_concurrency_limit?: number; // 开启后默认 5 From 6ffe04cf3dc82a20d7ec16d73e534de679213b57 Mon Sep 17 00:00:00 2001 From: RuoJi6 Date: Sat, 24 Oct 2026 21:17:59 +0800 Subject: [PATCH 2/4] chore: reset #41 multi-model worker files to origin/main for Codex rewrite --- .gitignore | 1 - server/engine.go | 34 +---- server/server.go | 37 +---- server/task_llm.go | 91 ++---------- server/worker_pools.go | 127 ---------------- server/worker_pools_test.go | 86 ----------- web/src/app/(main)/system/settings/page.tsx | 154 ++------------------ web/src/lib/mock/data.ts | 1 - web/src/lib/types.ts | 2 - 9 files changed, 34 insertions(+), 499 deletions(-) delete mode 100644 server/worker_pools.go delete mode 100644 server/worker_pools_test.go diff --git a/.gitignore b/.gitignore index 96def82..c9a7436 100644 --- a/.gitignore +++ b/.gitignore @@ -60,4 +60,3 @@ Thumbs.db /bench/ /Dockerfile.bench /Dockerfile.bench.v2 -.plan/ diff --git a/server/engine.go b/server/engine.go index e99cf17..b5643b2 100644 --- a/server/engine.go +++ b/server/engine.go @@ -148,11 +148,6 @@ type Engine struct { // (for example an exhausted failover chain) — there is no global-pair fallback. resolve func(t *Task) (*agent.Planner, *agent.Worker) resolveAuthoritative bool - // resolveWorker returns the Work Agent for one spawn slot. profileID 0 → unbound - // task-router worker (legacy); non-zero → worker pinned to that LLM profile. - // For profileID 0, a nil result falls back to snapshotFor; for pinned slots, - // nil means wait/retry (never switch to another model). - resolveWorker func(t *Task, profileID int64) *agent.Worker // readiness reports whether a global LLM provider is configured — the signal behind // Ready()/the llm_configured indicator. Wired once at startup; nil → not ready. readiness func() bool @@ -720,12 +715,6 @@ func (e *Engine) SetAuthoritativeAgentResolver(fn func(t *Task) (*agent.Planner, e.resolveAuthoritative = true } -// SetWorkerResolver installs the per-slot worker factory used by workerLoop when -// multi-model pools assign a profile to a session (work#N). Called once at startup. -func (e *Engine) SetWorkerResolver(fn func(t *Task, profileID int64) *agent.Worker) { - e.resolveWorker = fn -} - // snapshotFor returns the planner/worker a task should run on, from the task-router // resolver. nil,nil means the task is deliberately unavailable (e.g. an exhausted // failover chain); there is no global-pair fallback. @@ -757,7 +746,7 @@ func (e *Engine) ReadyFor(t *Task) bool { // but no-op until an LLM is configured (so a task created while idle picks up // automatically once LLM is set from the UI). func (e *Engine) Run(ctx context.Context, t *Task) { - slots := e.m.WorkerSlots() + workers := e.m.Workers() e.deleteMu.RLock() if e.IsDeleting(t.ID) { e.deleteMu.RUnlock() @@ -768,13 +757,13 @@ func (e *Engine) Run(ctx context.Context, t *Task) { t.Notify() // already running — just nudge a planning round return } - rt := e.registerTaskRoutines(ctx, t.ID, 1+len(slots)) + rt := e.registerTaskRoutines(ctx, t.ID, 1+workers) e.deleteMu.RUnlock() e.touch(t.ID) runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) }) - for _, slot := range slots { - name, profileID := slot.Name, slot.ProfileID - runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name, profileID) }) + for i := 0; i < workers; i++ { + name := fmt.Sprintf("work#%d", i+1) + runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name) }) } e.startDeadlineCoordinator(ctx, t) // 任务级超时定时器(仅 timeout>0;去重) // 仅在「完全没有活动意图(open+running)」时才 kick 首轮规划。带种子意图的任务:种子已 @@ -924,23 +913,14 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) { } } -func (e *Engine) workerLoop(ctx context.Context, t *Task, name string, profileID int64) { +func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { for { select { case <-ctx.Done(): return default: } - var worker *agent.Worker - if e.resolveWorker != nil { - worker = e.resolveWorker(t, profileID) - } - // Legacy unbound slots (profileID 0) may fall back to the task-router - // snapshot. Pinned multi-model pool slots must NOT silently switch models - // when the assigned profile is missing/unavailable — wait and retry. - if worker == nil && profileID == 0 { - _, worker = e.snapshotFor(t) - } + _, worker := e.snapshotFor(t) if worker == nil { if sleepCtx(ctx, 1500*time.Millisecond) { return diff --git a/server/server.go b/server/server.go index ec51b8a..969216c 100644 --- a/server/server.go +++ b/server/server.go @@ -177,11 +177,6 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi b := s.agentsForTask(t) return b.pl, b.wk }) - // Multi-model Worker pools (#41): each spawn slot may pin a Work Agent to a - // specific LLM profile. profileID 0 keeps the unbound task-router worker. - s.engine.SetWorkerResolver(func(t *Task, profileID int64) *agent.Worker { - return s.workerForTask(t, profileID) - }) // Global readiness (Ready()/llm_configured): a global active LLM provider is // installed. Task-level runnability is separate (ReadyFor → the resolver above). s.engine.SetReadiness(func() bool { @@ -3437,7 +3432,6 @@ func (s *Server) settingsPayload() map[string]any { "global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5)，所有目标流量走它，空=直连 "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 - "worker_pools": workerPoolsPayload(s.m), // 多模型 Worker 池；空=走 workers 遗留模式 "task_concurrency_enabled": concOn, // 任务并发上限开关(默认关) "task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5) // LLM 轮询(故障转移)。默认关；开启后走全局激活配置的 agent 在当前配置不可用时 @@ -3458,18 +3452,6 @@ func (s *Server) settingsPayload() map[string]any { } } - -// workerPoolsPayload always returns a non-nil slice for JSON (empty = legacy mode). -func workerPoolsPayload(m *Manager) []WorkerPoolConfig { - if m == nil { - return []WorkerPoolConfig{} - } - if pools := m.WorkerPools(); pools != nil { - return pools - } - return []WorkerPoolConfig{} -} - // notifyPublicBaseURL 读推送回链用的外部地址。 func notifyPublicBaseURL(pg *db.DB) string { v, _, _ := pg.GetSetting(settingNotifyPublicBaseURL) @@ -3524,8 +3506,7 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空 GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连) PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径 - Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 - WorkerPools *[]WorkerPoolConfig `json:"worker_pools"` // 多模型池；null=不改，[]=清空回遗留模式 + Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 ConcurrencyEnabled *bool `json:"task_concurrency_enabled"` ConcurrencyLimit *int `json:"task_concurrency_limit"` @@ -3604,22 +3585,6 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { return } } - if req.WorkerPools != nil { - if err := s.m.SetWorkerPools(*req.WorkerPools); err != nil { - writeErr(w, 400, err.Error()) - return - } - // 有池配置时把 workers 同步成合计,方便 UI/遗留读数与真实并发一致。 - if pools := *req.WorkerPools; len(pools) > 0 { - total := 0 - for _, p := range pools { - total += p.Workers - } - if total > 0 { - _ = s.m.SetWorkers(total) - } - } - } if req.ConcurrencyEnabled != nil || req.ConcurrencyLimit != nil { // 部分 PUT:未给的字段用当前值兜底,避免只改一个把另一个重置。 curOn, curLimit := s.m.ConcurrencyLimit() diff --git a/server/task_llm.go b/server/task_llm.go index 48a086f..2cae9ce 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -27,9 +27,6 @@ type taskAgentBundle struct { pl *agent.Planner wk *agent.Worker main *agent.MainAgent - // workersByProfile caches Work Agents pinned to a specific LLM profile for - // multi-model pools (issue #41). Key = llm_profiles.id. - workersByProfile map[int64]*agent.Worker } type llmAuditProfile struct { @@ -54,11 +51,6 @@ type taskLLMRuntime struct { s *Server taskID string agentKey string - // pinProfileID, when non-zero, forces every LLM call onto that profile and - // skips agent-binding / task-chain / global resolution. Used by multi-model - // Worker pools so each session (work#N) stays on its assigned model. Quota - // errors do NOT advance the task failover chain (sel.profileID stays 0). - pinProfileID int64 } type taskLLMError struct { @@ -130,15 +122,6 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) { r.s.syncTaskLLMState(pt) t, _ := r.s.m.Task(r.taskID) sel := taskLLMSelection{task: t, revision: pt.LLMChainRevision} - if r.pinProfileID != 0 { - prov, cfg, ok := r.s.providerForProfile(r.pinProfileID) - if !ok { - return sel, fmt.Errorf("pinned LLM profile #%d is missing or invalid", r.pinProfileID) - } - // profileID left 0 so streamTaskLLM/completeTaskLLM never mutate the task chain. - sel.provider, sel.retry = r.s.poolForBinding(r.pinProfileID, prov, cfg), cfg.Retry - return sel, nil - } if prov, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { sel.provider, sel.retry = prov, cfg.Retry return sel, nil @@ -169,12 +152,6 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) { // the caller. If failover switches profiles, the change takes effect on the next // agent run (a fresh Session is built per run in captureRun). func (r *taskLLMRuntime) activeCfg() (agent.Config, bool) { - if r.pinProfileID != 0 { - if _, cfg, ok := r.s.providerForProfile(r.pinProfileID); ok { - return cfg, true - } - return agent.Config{}, false - } taskNum, err := parseTaskID(r.taskID) if err != nil { return agent.Config{}, false @@ -477,12 +454,6 @@ func isRetryableStreamError(err error) bool { // CompactionWindow mirrors current()'s precedence so the context window always // matches the provider the role will actually stream on. func (r *taskLLMRuntime) CompactionWindow() int { - if r.pinProfileID != 0 { - if _, cfg, ok := r.s.providerForProfile(r.pinProfileID); ok { - return cfg.CompactionWindow() - } - return (agent.Config{}).CompactionWindow() - } if _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { return cfg.CompactionWindow() } @@ -589,7 +560,17 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { workerRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "worker"} mainRuntime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "mainagent"} tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) - wk := s.newTaskWorker(workerRuntime, tx) + window := workerRuntime.CompactionWindow() + wk := agent.NewWorker(workerRuntime, "task-router", s.m.dir, tx, window, s.agentMaxTurns("worker")) + wk.SetFindingRecorder(s.evidenceStore()) + wk.SetCompactionWindowResolver(workerRuntime.CompactionWindow) + wk.SetNonStreaming(workerRuntime.nonStreaming) // 按任务当前激活 profile 的流式开关(每轮读) + wk.SetMaxTokens(workerRuntime.maxTokens) // 同上,输出上限也跟随当前激活 profile + wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读) + wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) + wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) + wk.SetWebSearch(s.webSearchFor("worker")) + wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开) pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner")) pl.SetFindingRecorder(s.evidenceStore()) pl.SetCompactionWindowResolver(plannerRuntime.CompactionWindow) @@ -617,61 +598,11 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { bundle := &taskAgentBundle{ runtime: goalRuntime, plannerRuntime: plannerRuntime, workerRuntime: workerRuntime, mainRuntime: mainRuntime, pl: pl, wk: wk, main: main, - workersByProfile: map[int64]*agent.Worker{}, } s.taskAgents[t.ID] = bundle return bundle } - -// newTaskWorker builds a Work Agent on the given runtime (shared wiring for the -// unbound task worker and per-profile pinned workers). -func (s *Server) newTaskWorker(runtime *taskLLMRuntime, tx *transcript.Store) *agent.Worker { - wk := agent.NewWorker(runtime, "task-router", s.m.dir, tx, runtime.CompactionWindow(), s.agentMaxTurns("worker")) - wk.SetFindingRecorder(s.evidenceStore()) - wk.SetCompactionWindowResolver(runtime.CompactionWindow) - wk.SetNonStreaming(runtime.nonStreaming) - wk.SetMaxTokens(runtime.maxTokens) - wk.SetNoaEnabled(s.m.NoaCompactionEnabled) - wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) - wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) - wk.SetWebSearch(s.webSearchFor("worker")) - wk.SetConstraintInject(s.constraintInjectWorker) - return wk -} - -// workerForTask returns the Work Agent for one Engine spawn slot. -// profileID 0 → unbound task-router worker (requires taskRuntimeAvailable). -// Non-zero → worker pinned to that LLM profile (must build; ignores task-chain exhaustion). -func (s *Server) workerForTask(t *Task, profileID int64) *agent.Worker { - if t == nil { - return nil - } - if profileID == 0 { - if !s.taskRuntimeAvailable(t, "worker") { - return nil - } - return s.agentsForTask(t).wk - } - if _, _, ok := s.providerForProfile(profileID); !ok { - return nil - } - bundle := s.agentsForTask(t) - s.taskAgentMu.Lock() - defer s.taskAgentMu.Unlock() - if bundle.workersByProfile == nil { - bundle.workersByProfile = map[int64]*agent.Worker{} - } - if w := bundle.workersByProfile[profileID]; w != nil { - return w - } - tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) - runtime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "worker", pinProfileID: profileID} - w := s.newTaskWorker(runtime, tx) - bundle.workersByProfile[profileID] = w - return w -} - func (s *Server) syncTaskLLMState(pt *db.Task) { if pt == nil { return diff --git a/server/worker_pools.go b/server/worker_pools.go deleted file mode 100644 index b66c691..0000000 --- a/server/worker_pools.go +++ /dev/null @@ -1,127 +0,0 @@ -package server - -import ( - "encoding/json" - "fmt" - "strings" -) - -// settingWorkerPools stores the multi-model Work Agent pool config as a JSON array. -// Empty / unset → legacy single-pool mode using settings.workers + unbound runtime. -const settingWorkerPools = "worker_pools" - -// maxWorkerPoolTotal is the hard upper bound on sum(pool.workers), matching the -// per-task override ceiling used elsewhere so a misconfigured UI can't spawn dozens -// of LLM sessions per task. -const maxWorkerPoolTotal = 50 - -// WorkerPoolConfig is one LLM profile's share of concurrent Work Agents. -type WorkerPoolConfig struct { - ProfileID int64 `json:"profile_id"` - Workers int `json:"workers"` -} - -// WorkerSlot is one Engine.Run worker loop: Name is the claim/session id (work#N); -// ProfileID 0 means unbound (legacy task-router resolution). -type WorkerSlot struct { - Name string - ProfileID int64 -} - -// validateWorkerPools checks a pools slice before persistence. Empty is allowed -// (clears multi-model mode). Non-empty requires every entry valid and total in 1..50. -func validateWorkerPools(pools []WorkerPoolConfig) error { - if len(pools) == 0 { - return nil - } - total := 0 - for i, p := range pools { - if p.ProfileID <= 0 { - return fmt.Errorf("worker_pools[%d].profile_id 必须 >0", i) - } - if p.Workers <= 0 { - return fmt.Errorf("worker_pools[%d].workers 必须 >0", i) - } - total += p.Workers - } - if total > maxWorkerPoolTotal { - return fmt.Errorf("worker_pools 合计 Worker 数最多 %d（当前 %d）", maxWorkerPoolTotal, total) - } - return nil -} - -// expandWorkerSlots turns pool configs into a spawn plan. Empty pools → legacy: -// N unbound slots named work#1..N where N = legacyWorkers (clamped to >=1). -func expandWorkerSlots(pools []WorkerPoolConfig, legacyWorkers int) []WorkerSlot { - if len(pools) == 0 { - n := legacyWorkers - if n <= 0 { - n = defaultWorkers - } - out := make([]WorkerSlot, n) - for i := 0; i < n; i++ { - out[i] = WorkerSlot{Name: fmt.Sprintf("work#%d", i+1), ProfileID: 0} - } - return out - } - var out []WorkerSlot - for _, p := range pools { - if p.ProfileID <= 0 || p.Workers <= 0 { - continue - } - for i := 0; i < p.Workers; i++ { - out = append(out, WorkerSlot{ - Name: fmt.Sprintf("work#%d", len(out)+1), - ProfileID: p.ProfileID, - }) - } - } - if len(out) == 0 { - return expandWorkerSlots(nil, legacyWorkers) - } - return out -} - -// WorkerPools returns the configured multi-model pools (nil/empty = legacy mode). -// Invalid entries in the stored JSON are skipped. -func (m *Manager) WorkerPools() []WorkerPoolConfig { - if m == nil || m.pg == nil { - return nil - } - v, ok, err := m.pg.GetSetting(settingWorkerPools) - if err != nil || !ok || strings.TrimSpace(v) == "" { - return nil - } - var raw []WorkerPoolConfig - if json.Unmarshal([]byte(v), &raw) != nil { - return nil - } - out := make([]WorkerPoolConfig, 0, len(raw)) - for _, p := range raw { - if p.ProfileID > 0 && p.Workers > 0 { - out = append(out, p) - } - } - return out -} - -// SetWorkerPools persists the multi-model pool config. An empty slice clears the -// setting (back to legacy single-pool Workers()). Applies to tasks started afterwards. -func (m *Manager) SetWorkerPools(pools []WorkerPoolConfig) error { - if err := validateWorkerPools(pools); err != nil { - return err - } - if len(pools) == 0 { - return m.pg.SetSetting(settingWorkerPools, "") - } - b, err := json.Marshal(pools) - if err != nil { - return err - } - return m.pg.SetSetting(settingWorkerPools, string(b)) -} - -// WorkerSlots returns the expanded spawn plan for Engine.Run (read per task start). -func (m *Manager) WorkerSlots() []WorkerSlot { - return expandWorkerSlots(m.WorkerPools(), m.Workers()) -} diff --git a/server/worker_pools_test.go b/server/worker_pools_test.go deleted file mode 100644 index 78ca527..0000000 --- a/server/worker_pools_test.go +++ /dev/null @@ -1,86 +0,0 @@ -package server - -import ( - "testing" -) - -func TestExpandWorkerSlotsLegacy(t *testing.T) { - slots := expandWorkerSlots(nil, 3) - if len(slots) != 3 { - t.Fatalf("len=%d want 3", len(slots)) - } - for i, s := range slots { - want := "work#" + itoa(i+1) - if s.Name != want || s.ProfileID != 0 { - t.Fatalf("slot[%d]=%+v want name=%s profile=0", i, s, want) - } - } - slots = expandWorkerSlots(nil, 0) - if len(slots) != defaultWorkers { - t.Fatalf("legacy 0 → defaultWorkers: got %d", len(slots)) - } -} - -func TestExpandWorkerSlotsMulti(t *testing.T) { - slots := expandWorkerSlots([]WorkerPoolConfig{ - {ProfileID: 1, Workers: 2}, - {ProfileID: 2, Workers: 3}, - }, 3) - if len(slots) != 5 { - t.Fatalf("len=%d want 5", len(slots)) - } - want := []int64{1, 1, 2, 2, 2} - for i, s := range slots { - if s.ProfileID != want[i] || s.Name != "work#"+itoa(i+1) { - t.Fatalf("slot[%d]=%+v want profile=%d", i, s, want[i]) - } - } -} - -func TestExpandWorkerSlotsSkipsInvalidFallsBack(t *testing.T) { - slots := expandWorkerSlots([]WorkerPoolConfig{ - {ProfileID: 0, Workers: 2}, - {ProfileID: 1, Workers: 0}, - }, 4) - if len(slots) != 4 || slots[0].ProfileID != 0 { - t.Fatalf("all-invalid pools should fall back to legacy: %+v", slots) - } -} - -func TestValidateWorkerPools(t *testing.T) { - if err := validateWorkerPools(nil); err != nil { - t.Fatal(err) - } - if err := validateWorkerPools([]WorkerPoolConfig{}); err != nil { - t.Fatal(err) - } - if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 0, Workers: 1}}); err == nil { - t.Fatal("expected profile_id error") - } - if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 1, Workers: 0}}); err == nil { - t.Fatal("expected workers error") - } - big := make([]WorkerPoolConfig, 0, 3) - big = append(big, WorkerPoolConfig{ProfileID: 1, Workers: 40}) - big = append(big, WorkerPoolConfig{ProfileID: 2, Workers: 20}) - if err := validateWorkerPools(big); err == nil { - t.Fatal("expected sum>50 error") - } - if err := validateWorkerPools([]WorkerPoolConfig{{ProfileID: 1, Workers: 3}, {ProfileID: 2, Workers: 3}}); err != nil { - t.Fatal(err) - } -} - -func itoa(n int) string { - if n == 0 { - return "0" - } - var b [16]byte - i := len(b) - for n > 0 { - i-- - b[i] = byte('0' + n%10) - n /= 10 - } - return string(b[i:]) -} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index 50420b8..fea869f 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -2,7 +2,7 @@ import * as React from "react"; -import { CpuIcon, FlaskConicalIcon, KeyboardIcon, PlusIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon, Trash2Icon } from "lucide-react"; +import { CpuIcon, FlaskConicalIcon, KeyboardIcon, RadioTowerIcon, SearchIcon, ShieldAlertIcon } from "lucide-react"; import { toast } from "sonner"; import { Button } from "@/components/ui/button"; @@ -14,7 +14,7 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@ import { Switch } from "@/components/ui/switch"; import { api } from "@/lib/api"; import { CHAT_SEND_MODE_OPTIONS, type ChatSendMode, setChatSendMode, useChatSendMode } from "@/lib/chat-send-mode"; -import type { LLMProfile, Settings } from "@/lib/types"; +import type { Settings } from "@/lib/types"; import { BasicAuthCard } from "./_components/basic-auth-card"; import { UpdateCard } from "./_components/update-card"; @@ -45,8 +45,6 @@ export default function SystemSettingsPage() { const [pyInterp, setPyInterp] = React.useState(""); const [workers, setWorkers] = React.useState("3"); const [savingWorkers, setSavingWorkers] = React.useState(false); - const [llmProfiles, setLlmProfiles] = React.useState([]); - const [workerPools, setWorkerPools] = React.useState<{ profile_id: string; workers: string }[]>([]); // 操作约束注入范围(默认都开)。 const [injectPlanner, setInjectPlanner] = React.useState(true); const [injectWorker, setInjectWorker] = React.useState(true); @@ -69,53 +67,23 @@ export default function SystemSettingsPage() { setGlobalProxyInput(s.global_proxy ?? ""); setPyInterp(s.python_interpreter ?? ""); setWorkers(String(s.workers ?? 3)); - setWorkerPools( - (s.worker_pools ?? []).map((p) => ({ - profile_id: String(p.profile_id), - workers: String(p.workers), - })), - ); setInjectPlanner(s.constraints_inject_planner !== false); setInjectWorker(s.constraints_inject_worker !== false); setNoaCompaction(!!s.noa_compaction); }, []); const saveWorkers = () => { - const pools: { profile_id: number; workers: number }[] = []; - for (let i = 0; i < workerPools.length; i++) { - const row = workerPools[i]; - const pid = Number(row.profile_id); - const n = Number(row.workers); - if (!Number.isInteger(pid) || pid <= 0) { - toast.error(`第 ${i + 1} 行：请选择 LLM 配置`); - return; - } - if (!Number.isInteger(n) || n <= 0) { - toast.error(`第 ${i + 1} 行：Worker 数必须是大于 0 的整数`); - return; - } - pools.push({ profile_id: pid, workers: n }); - } - const legacy = Number(workers); - if (pools.length === 0 && (!Number.isInteger(legacy) || legacy <= 0)) { - toast.error("默认并发数必须是大于 0 的整数"); - return; - } - const total = pools.length > 0 ? pools.reduce((s, p) => s + p.workers, 0) : legacy; - if (total > 50) { - toast.error("Worker 合计最多 50"); + const n = Number(workers); + if (!Number.isInteger(n) || n <= 0) { + toast.error("并发数必须是大于 0 的整数"); return; } setSavingWorkers(true); api - .setSettings({ workers: total, worker_pools: pools }) + .setSettings({ workers: n }) .then((s) => { apply(s); - toast.success( - pools.length > 0 - ? `已保存多模型 Worker 池（共 ${total} 个会话；对之后启动的任务生效）` - : "已保存并发工作 agent 数（对之后启动的任务生效）", - ); + toast.success("已保存并发工作 agent 数（对之后启动的任务生效）"); }) .catch((e) => toast.error("保存失败：" + (e as Error).message)) .finally(() => setSavingWorkers(false)); @@ -142,11 +110,9 @@ export default function SystemSettingsPage() { }; React.useEffect(() => { - Promise.all([api.settings(), api.llmProfiles().catch(() => [] as LLMProfile[])]) - .then(([s, profiles]) => { - apply(s); - setLlmProfiles(profiles); - }) + api + .settings() + .then(apply) .catch(() => undefined) .finally(() => setLoaded(true)); }, [apply]); @@ -782,114 +748,24 @@ export default function SystemSettingsPage() { - 工作并发 · Work Agent / 多模型池 + 工作并发 · Work Agent 数 - 可为多个 LLM 配置各分配若干 Worker；同一任务内多模型并行执行，按会话（work#N）绑定模型。 - 未配置模型池时，使用下方「默认并发」走单一模型（与历史行为一致）。修改后 + 每个任务并发运行的工作 agent 数量（默认 3）。数值越大并发探测越多、消耗也越高。修改后 对之后启动的任务生效，正在运行的任务不受影响。 - -
- + +
0} + disabled={!loaded || savingWorkers} onChange={(e) => setWorkers(e.target.value)} /> - {workerPools.length > 0 ? ( -

已配置模型池时以各池合计为准，默认并发输入已锁定。

- ) : null} -
- -
-
- - -
- {llmProfiles.length === 0 ? ( -

尚未配置 LLM：请先到「系统 → LLM 配置」添加并激活。

- ) : null} - {workerPools.map((row, idx) => ( -
- - - setWorkerPools((rows) => - rows.map((r, i) => (i === idx ? { ...r, workers: e.target.value } : r)), - ) - } - /> - Workers - -
- ))} -

- 当前合计： - {workerPools.length > 0 - ? workerPools.reduce((s, r) => s + (Number(r.workers) || 0), 0) - : Number(workers) || 0}{" "} - 个 Worker 会话 -

-
- -
diff --git a/web/src/lib/mock/data.ts b/web/src/lib/mock/data.ts index 9b8a69c..531df7b 100644 --- a/web/src/lib/mock/data.ts +++ b/web/src/lib/mock/data.ts @@ -3338,7 +3338,6 @@ export const settings: Settings = { global_proxy: "", python_interpreter: "/usr/bin/python3", workers: 3, - worker_pools: [], llm_pool_enabled: true, llm_pool_bind_fallback: false, noa_compaction: false, diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 5a7188d..54e26a6 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -968,8 +968,6 @@ export interface Settings { global_proxy?: string; python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测) workers?: number; // 并发工作 agent 数(默认3)；对之后启动的任务生效 - // 多模型 Worker 池：每个 LLM 配置各自占用若干并发会话。空/缺省=走上面的 workers 遗留模式。 - worker_pools?: { profile_id: number; workers: number }[]; // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 task_concurrency_enabled?: boolean; // 默认 false task_concurrency_limit?: number; // 开启后默认 5 From d1b2b5a996f40f2cfd1530f19748e22369337064 Mon Sep 17 00:00:00 2001 From: RuoJi6 Date: Sat, 24 Oct 2026 22:22:05 +0800 Subject: [PATCH 3/4] =?UTF-8?q?feat:=20=E5=A4=9A=E6=A8=A1=E5=9E=8B=20Worke?= =?UTF-8?q?r=20=E6=B1=A0=EF=BC=88#41=EF=BC=8CCodex=20=E9=87=8D=E5=86=99?= =?UTF-8?q?=EF=BC=9A=E7=BB=91=E5=AE=9A=E9=85=8D=E7=BD=AE=E4=B8=8D=E5=8F=AF?= =?UTF-8?q?=E7=94=A8=E6=97=B6=E6=8A=A5=E9=94=99=E8=80=8C=E9=9D=9E=E5=9B=9E?= =?UTF-8?q?=E9=80=80=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code authored by Codex CLI (gpt-6-astra main implementation; gpt-6-sol / gpt-6-astra continuation passes). --- server/engine.go | 81 +++++- server/manager.go | 149 +++++++++++ server/server.go | 51 +++- server/task_llm.go | 77 +++++- server/worker_pools_test.go | 263 ++++++++++++++++++++ web/src/app/(main)/system/settings/page.tsx | 121 ++++++++- web/src/lib/mock/data.ts | 1 + web/src/lib/mock/handler.ts | 13 +- web/src/lib/types.ts | 6 + 9 files changed, 733 insertions(+), 29 deletions(-) create mode 100644 server/worker_pools_test.go diff --git a/server/engine.go b/server/engine.go index b5643b2..9a8c07b 100644 --- a/server/engine.go +++ b/server/engine.go @@ -148,11 +148,18 @@ type Engine struct { // (for example an exhausted failover chain) — there is no global-pair fallback. resolve func(t *Task) (*agent.Planner, *agent.Worker) resolveAuthoritative bool + workerResolve func(t *Task, slot WorkerSlot) (*agent.Worker, error) + slotState sync.Map // task/slot -> slotAvailability // readiness reports whether a global LLM provider is configured — the signal behind // Ready()/the llm_configured indicator. Wired once at startup; nil → not ready. readiness func() bool } +type slotAvailability struct { + available bool + lastEmit time.Time +} + type taskRuntime struct { ctx context.Context cancel context.CancelFunc @@ -285,6 +292,13 @@ func (e *Engine) StopTask(taskID string) { e.stamped.Delete(taskID) e.inflight.Delete(taskID) e.coordStarted.Delete(taskID) + e.slotState.Range(func(key, _ any) bool { + keyText, ok := key.(string) + if ok && strings.HasPrefix(keyText, taskID+"/") { + e.slotState.Delete(key) + } + return true + }) e.deleteMu.Lock() e.deleting.Delete(taskID) e.deleteMu.Unlock() @@ -715,6 +729,37 @@ func (e *Engine) SetAuthoritativeAgentResolver(fn func(t *Task) (*agent.Planner, e.resolveAuthoritative = true } +func (e *Engine) SetWorkerResolver(fn func(t *Task, slot WorkerSlot) (*agent.Worker, error)) { + e.workerResolve = fn +} + +func (e *Engine) reportSlotAvailability(t *Task, slot WorkerSlot, err error) { + key := t.ID + "/" + slot.Name + now := time.Now() + value, loaded := e.slotState.LoadOrStore(key, slotAvailability{available: err == nil, lastEmit: now}) + state := value.(slotAvailability) + if err == nil { + if loaded && !state.available { + e.slotState.Store(key, slotAvailability{available: true, lastEmit: now}) + e.emitActivity(t, db.Activity{Worker: slot.Name, Kind: "text", Summary: slot.Name + " 绑定的 LLM 配置已恢复，可以继续领取任务"}) + } + return + } + if loaded && state.available { + state.available = false + state.lastEmit = now + e.slotState.Store(key, state) + } else if loaded && now.Sub(state.lastEmit) < 5*time.Minute { + return + } else { + state = slotAvailability{available: false, lastEmit: now} + e.slotState.Store(key, state) + } + summary := fmt.Sprintf("%s 绑定的 LLM 配置 #%d 不可用（已删除或配置无效），该会话暂停领取任务，不会回退到默认模型", slot.Name, slot.ProfileID) + e.emitActivity(t, db.Activity{Worker: slot.Name, Kind: "text", IsError: true, Summary: summary, Detail: err.Error()}) + log.Printf("[worker %s] task %s pinned profile #%d unavailable: %v", slot.Name, t.ID, slot.ProfileID, err) +} + // snapshotFor returns the planner/worker a task should run on, from the task-router // resolver. nil,nil means the task is deliberately unavailable (e.g. an exhausted // failover chain); there is no global-pair fallback. @@ -746,7 +791,11 @@ func (e *Engine) ReadyFor(t *Task) bool { // but no-op until an LLM is configured (so a task created while idle picks up // automatically once LLM is set from the UI). func (e *Engine) Run(ctx context.Context, t *Task) { - workers := e.m.Workers() + slots, err := ExpandWorkerSlots(e.m.WorkerPools(), e.m.Workers()) + if err != nil { + log.Printf("[engine] task %s worker pool invalid, using legacy workers: %v", t.ID, err) + slots, _ = ExpandWorkerSlots(nil, e.m.Workers()) + } e.deleteMu.RLock() if e.IsDeleting(t.ID) { e.deleteMu.RUnlock() @@ -757,13 +806,13 @@ func (e *Engine) Run(ctx context.Context, t *Task) { t.Notify() // already running — just nudge a planning round return } - rt := e.registerTaskRoutines(ctx, t.ID, 1+workers) + rt := e.registerTaskRoutines(ctx, t.ID, 1+len(slots)) e.deleteMu.RUnlock() e.touch(t.ID) runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) }) - for i := 0; i < workers; i++ { - name := fmt.Sprintf("work#%d", i+1) - runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name) }) + for _, slot := range slots { + slot := slot + runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, slot) }) } e.startDeadlineCoordinator(ctx, t) // 任务级超时定时器(仅 timeout>0;去重) // 仅在「完全没有活动意图(open+running)」时才 kick 首轮规划。带种子意图的任务:种子已 @@ -913,14 +962,32 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) { } } -func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { +func (e *Engine) workerLoop(ctx context.Context, t *Task, slot WorkerSlot) { + name := slot.Name for { select { case <-ctx.Done(): return default: } - _, worker := e.snapshotFor(t) + var worker *agent.Worker + var resolveErr error + if slot.ProfileID > 0 { + if e.workerResolve == nil { + resolveErr = errors.New("pinned worker resolver unavailable") + } else { + worker, resolveErr = e.workerResolve(t, slot) + if worker == nil && resolveErr == nil { + resolveErr = errors.New("pinned worker resolver returned no worker") + } + } + e.reportSlotAvailability(t, slot, resolveErr) + if resolveErr != nil { + worker = nil + } + } else { + _, worker = e.snapshotFor(t) + } if worker == nil { if sleepCtx(ctx, 1500*time.Millisecond) { return diff --git a/server/manager.go b/server/manager.go index 09fdcd3..76f72ae 100644 --- a/server/manager.go +++ b/server/manager.go @@ -265,6 +265,7 @@ const ( // only routes the search backend) and the per-profile LLM proxy. settingGlobalProxy = "global_proxy" settingWorkers = "workers" + settingWorkerPools = "worker_pools" settingLLMRecord = "llm_record" // LLM 轮询(故障转移)。默认关闭——开启后走「全局激活配置」的 agent 在当前配置 // 不可用(余额不足/key 失效/限流/服务异常)时自动切到下一个配置。 @@ -291,6 +292,102 @@ const ( defaultConcurrencyLimit = 5 ) +// WorkerPool binds a number of worker slots to one LLM profile. Profile IDs are +// encoded as JSON strings to match the frontend LLMProfile API. +type WorkerPool struct { + ProfileID int64 `json:"profile_id,string"` + Workers int `json:"workers"` +} + +func (p *WorkerPool) UnmarshalJSON(data []byte) error { + var raw struct { + ProfileID json.RawMessage `json:"profile_id"` + Workers int `json:"workers"` + } + if err := json.Unmarshal(data, &raw); err != nil { + return err + } + var id int64 + if err := json.Unmarshal(raw.ProfileID, &id); err != nil { + var text string + if err := json.Unmarshal(raw.ProfileID, &text); err != nil { + return fmt.Errorf("profile_id 必须是整数或数字字符串") + } + parsed, err := strconv.ParseInt(strings.TrimSpace(text), 10, 64) + if err != nil { + return fmt.Errorf("profile_id 必须是整数或数字字符串") + } + id = parsed + } + p.ProfileID, p.Workers = id, raw.Workers + return nil +} + +// WorkerSlot is the concrete routing plan for one task run. +type WorkerSlot struct { + Name string + ProfileID int64 // zero means legacy task-router resolution +} + +// ValidateWorkerPools validates pool shape without requiring a database. Pool +// entries with duplicate profile IDs are rejected so each profile has one +// unambiguous slot group in the saved order. +func ValidateWorkerPools(pools []WorkerPool) error { + if len(pools) == 0 { + return nil + } + seen := make(map[int64]bool, len(pools)) + total := 0 + for _, pool := range pools { + if pool.ProfileID <= 0 { + return fmt.Errorf("worker_pools 中 profile_id 必须大于 0") + } + if pool.Workers <= 0 { + return fmt.Errorf("worker_pools 中 workers 必须大于 0") + } + if seen[pool.ProfileID] { + return fmt.Errorf("worker_pools 中 profile_id %d 重复", pool.ProfileID) + } + seen[pool.ProfileID] = true + if total > 50-pool.Workers { + return fmt.Errorf("worker_pools 的 Worker 总数必须在 1 到 50 之间") + } + total += pool.Workers + } + if total < 1 || total > 50 { + return fmt.Errorf("worker_pools 的 Worker 总数必须在 1 到 50 之间") + } + return nil +} + +// ExpandWorkerSlots expands pools in their saved order; invalid or empty legacy +// settings use the historical worker count (default 3). +func ExpandWorkerSlots(pools []WorkerPool, legacy int) ([]WorkerSlot, error) { + if err := ValidateWorkerPools(pools); err != nil { + return nil, err + } + if len(pools) == 0 { + if legacy <= 0 { + legacy = defaultWorkers + } + slots := make([]WorkerSlot, legacy) + for i := range slots { + slots[i] = WorkerSlot{Name: fmt.Sprintf("work#%d", i+1)} + } + return slots, nil + } + slots := make([]WorkerSlot, 0) + for _, pool := range pools { + for i := 0; i < pool.Workers; i++ { + slots = append(slots, WorkerSlot{ProfileID: pool.ProfileID}) + } + } + for i := range slots { + slots[i].Name = fmt.Sprintf("work#%d", i+1) + } + return slots, nil +} + // ConcurrencyLimit returns whether the simultaneous-running-task cap is enabled and // its limit (default 5 when enabled but unset). limit is always >=1 when enabled. func (m *Manager) ConcurrencyLimit() (enabled bool, limit int) { @@ -332,6 +429,58 @@ func (m *Manager) Workers() int { return n } +func (m *Manager) WorkerPools() []WorkerPool { + v, ok, err := m.pg.GetSetting(settingWorkerPools) + if err != nil || !ok || strings.TrimSpace(v) == "" { + return []WorkerPool{} + } + var pools []WorkerPool + if json.Unmarshal([]byte(v), &pools) != nil || ValidateWorkerPools(pools) != nil { + return []WorkerPool{} + } + if pools == nil { + return []WorkerPool{} + } + return pools +} + +// SetWorkerPools validates profile references at save time and synchronizes the +// legacy workers setting with the pool total. An empty slice restores legacy mode. +func (m *Manager) SetWorkerPools(pools []WorkerPool) error { + if pools == nil { + pools = []WorkerPool{} + } + if err := ValidateWorkerPools(pools); err != nil { + return err + } + if len(pools) > 0 { + for _, pool := range pools { + profile, err := m.pg.ProfileByID(pool.ProfileID) + if err != nil { + return err + } + if profile == nil { + return fmt.Errorf("worker_pools 中 profile_id %d 不存在", pool.ProfileID) + } + } + } + b, err := json.Marshal(pools) + if err != nil { + return err + } + if err := m.pg.SetSetting(settingWorkerPools, string(b)); err != nil { + return err + } + if len(pools) > 0 { + total := 0 + for _, pool := range pools { + total += pool.Workers + } + return m.pg.SetSetting(settingWorkers, strconv.Itoa(total)) + } + return nil +} + // SetWorkers persists the concurrent work-agent count. Values <=0 are rejected. func (m *Manager) SetWorkers(n int) error { if n <= 0 { diff --git a/server/server.go b/server/server.go index 969216c..fc3e0b5 100644 --- a/server/server.go +++ b/server/server.go @@ -177,6 +177,9 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi b := s.agentsForTask(t) return b.pl, b.wk }) + s.engine.SetWorkerResolver(func(t *Task, slot WorkerSlot) (*agent.Worker, error) { + return s.workerForTaskSlot(t, slot) + }) // Global readiness (Ready()/llm_configured): a global active LLM provider is // installed. Task-level runnability is separate (ReadyFor → the resolver above). s.engine.SetReadiness(func() bool { @@ -575,7 +578,8 @@ func (s *Server) chatUnavailableReason() string { // providerForProfile returns a cached provider+cfg for a profile id, so every agent // bound/pinned to the same profile shares one provider instance (one rate limiter). -// ok=false when the profile is missing/invalid → caller falls back to the global pair. +// ok=false when the profile is missing/invalid; pinned callers surface that error +// while legacy callers retain their existing fallback behavior. func (s *Server) providerForProfile(id int64) (llm.Provider, agent.Config, bool) { s.provCacheMu.Lock() if e := s.provByProfile[id]; e != nil { @@ -3432,6 +3436,7 @@ func (s *Server) settingsPayload() map[string]any { "global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5)，所有目标流量走它，空=直连 "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 + "worker_pools": s.m.WorkerPools(), // 非空时按池顺序绑定 worker 会话 "task_concurrency_enabled": concOn, // 任务并发上限开关(默认关) "task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5) // LLM 轮询(故障转移)。默认关；开启后走全局激活配置的 agent 在当前配置不可用时 @@ -3499,14 +3504,15 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { LLMRecord *bool `json:"llm_record"` // LLM 录制开关（默认关）；即时生效，无需重建 agent // Web search. WebSearchEnabled/Backend toggle the tool + backend; BraveKey/TavilyKey // are optional — omit (null) to leave a stored key untouched, send "" to clear. - WebSearchEnabled *bool `json:"web_search_enabled"` - WebSearchBackend *string `json:"web_search_backend"` - BraveKey *string `json:"brave_search_api_key"` - TavilyKey *string `json:"tavily_search_api_key"` - WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空 - GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连) - PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径 - Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 + WebSearchEnabled *bool `json:"web_search_enabled"` + WebSearchBackend *string `json:"web_search_backend"` + BraveKey *string `json:"brave_search_api_key"` + TavilyKey *string `json:"tavily_search_api_key"` + WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空 + GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连) + PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径 + Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 + WorkerPools *[]WorkerPool `json:"worker_pools"` // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 ConcurrencyEnabled *bool `json:"task_concurrency_enabled"` ConcurrencyLimit *int `json:"task_concurrency_limit"` @@ -3579,11 +3585,34 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { return } } - if req.Workers != nil { - if err := s.m.SetWorkers(*req.Workers); err != nil { + // A non-empty pool list takes precedence when both fields are present and + // synchronizes workers to its total. An empty list clears pools; when workers + // is also present it sets the legacy count for the newly restored legacy mode. + if req.WorkerPools != nil { + if len(*req.WorkerPools) == 0 && req.Workers != nil && *req.Workers <= 0 { + writeErr(w, 400, "workers 必须 >0") + return + } + if err := s.m.SetWorkerPools(*req.WorkerPools); err != nil { writeErr(w, 400, err.Error()) return } + if len(*req.WorkerPools) == 0 && req.Workers != nil { + if err := s.m.SetWorkers(*req.Workers); err != nil { + writeErr(w, 400, err.Error()) + return + } + } + } else if req.Workers != nil { + // An omitted worker_pools field leaves the saved mode unchanged. If a + // non-empty pool configuration is already saved, its total remains the + // authoritative concurrency and workers is deliberately ignored. + if len(s.m.WorkerPools()) == 0 { + if err := s.m.SetWorkers(*req.Workers); err != nil { + writeErr(w, 400, err.Error()) + return + } + } } if req.ConcurrencyEnabled != nil || req.ConcurrencyLimit != nil { // 部分 PUT:未给的字段用当前值兜底,避免只改一个把另一个重置。 diff --git a/server/task_llm.go b/server/task_llm.go index 2cae9ce..8063a3e 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -27,6 +27,7 @@ type taskAgentBundle struct { pl *agent.Planner wk *agent.Worker main *agent.MainAgent + pinnedWorkers map[int64]*agent.Worker } type llmAuditProfile struct { @@ -48,9 +49,10 @@ type llmActivityMetadata struct { } type taskLLMRuntime struct { - s *Server - taskID string - agentKey string + s *Server + taskID string + agentKey string + pinProfileID *int64 } type taskLLMError struct { @@ -101,13 +103,24 @@ type taskLLMStreamHooks struct { transition func(taskLLMSelection, db.TaskLLMTransition, error) } -// current resolves the provider this role runs on, by precedence: -// Agent 绑定 → 任务 LLM 配置链 → 全局/环境配置。 +// current resolves the provider this role runs on. Pinned workers use only their +// assigned profile; legacy task agents use Agent 绑定 → 任务 LLM 配置链 → 全局/环境配置. // 绑定优先于任务链：某个角色被显式指定了模型，就一直跑在那个模型上；绑定不存在或 // 构建失败时才降级到任务链，任务链为空时再降级到全局配置。 // 返回的 profile id 只有走任务链时才非零 —— streamTaskLLM 以此判断额度错误是否 // 应该推进任务的故障转移状态（绑定/全局路径不改任务链状态，沿用既有语义）。 func (r *taskLLMRuntime) current() (taskLLMSelection, error) { + if r.pinProfileID != nil { + // Pinned worker runtimes are exclusive: resolve their profile before + // touching bindings or the task failover state. A missing or invalid + // profile is an error, never an invitation to use another model. + prov, cfg, ok := r.s.providerForProfile(*r.pinProfileID) + if !ok { + return taskLLMSelection{}, fmt.Errorf("LLM profile #%d is missing or invalid", *r.pinProfileID) + } + t, _ := r.s.m.Task(r.taskID) + return taskLLMSelection{task: t, provider: prov, retry: cfg.Retry}, nil + } taskNum, err := parseTaskID(r.taskID) if err != nil { return taskLLMSelection{}, err @@ -152,6 +165,10 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) { // the caller. If failover switches profiles, the change takes effect on the next // agent run (a fresh Session is built per run in captureRun). func (r *taskLLMRuntime) activeCfg() (agent.Config, bool) { + if r.pinProfileID != nil { + _, cfg, ok := r.s.providerForProfile(*r.pinProfileID) + return cfg, ok + } taskNum, err := parseTaskID(r.taskID) if err != nil { return agent.Config{}, false @@ -454,6 +471,12 @@ func isRetryableStreamError(err error) bool { // CompactionWindow mirrors current()'s precedence so the context window always // matches the provider the role will actually stream on. func (r *taskLLMRuntime) CompactionWindow() int { + if r.pinProfileID != nil { + if _, cfg, ok := r.s.providerForProfile(*r.pinProfileID); ok { + return cfg.CompactionWindow() + } + return (agent.Config{}).CompactionWindow() + } if _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { return cfg.CompactionWindow() } @@ -597,12 +620,54 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { main.SetSteerWork(s.engine.SteerWork) // steer_work：人对运行中 work 实时纠偏 bundle := &taskAgentBundle{ runtime: goalRuntime, plannerRuntime: plannerRuntime, workerRuntime: workerRuntime, - mainRuntime: mainRuntime, pl: pl, wk: wk, main: main, + mainRuntime: mainRuntime, pl: pl, wk: wk, main: main, pinnedWorkers: map[int64]*agent.Worker{}, } s.taskAgents[t.ID] = bundle return bundle } +func (s *Server) workerForTaskSlot(t *Task, slot WorkerSlot) (*agent.Worker, error) { + if slot.ProfileID <= 0 { + return nil, errors.New("worker slot is not pinned") + } + _, cfg, ok := s.providerForProfile(slot.ProfileID) + if !ok { + return nil, fmt.Errorf("LLM profile #%d is missing or invalid", slot.ProfileID) + } + s.taskAgentMu.Lock() + bundle := s.taskAgents[t.ID] + s.taskAgentMu.Unlock() + if bundle == nil { + bundle = s.agentsForTask(t) + } + s.taskAgentMu.Lock() + defer s.taskAgentMu.Unlock() + if current := s.taskAgents[t.ID]; current != nil { + bundle = current + } + if bundle.pinnedWorkers == nil { + bundle.pinnedWorkers = make(map[int64]*agent.Worker) + } + if worker := bundle.pinnedWorkers[slot.ProfileID]; worker != nil { + return worker, nil + } + id := slot.ProfileID + runtime := &taskLLMRuntime{s: s, taskID: t.ID, agentKey: "worker", pinProfileID: &id} + tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) + worker := agent.NewWorker(runtime, cfg.Model, s.m.dir, tx, runtime.CompactionWindow(), s.agentMaxTurns("worker")) + worker.SetFindingRecorder(s.evidenceStore()) + worker.SetCompactionWindowResolver(runtime.CompactionWindow) + worker.SetNonStreaming(runtime.nonStreaming) + worker.SetMaxTokens(runtime.maxTokens) + worker.SetNoaEnabled(s.m.NoaCompactionEnabled) + worker.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) + worker.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) + worker.SetWebSearch(s.webSearchFor("worker")) + worker.SetConstraintInject(s.constraintInjectWorker) + bundle.pinnedWorkers[id] = worker + return worker, nil +} + func (s *Server) syncTaskLLMState(pt *db.Task) { if pt == nil { return diff --git a/server/worker_pools_test.go b/server/worker_pools_test.go new file mode 100644 index 0000000..36f40aa --- /dev/null +++ b/server/worker_pools_test.go @@ -0,0 +1,263 @@ +package server + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/Autumn-27/artex/agent" + "github.com/Autumn-27/artex/db" +) + +func TestExpandWorkerSlotsLegacy(t *testing.T) { + for _, tc := range []struct { + name string + count int + want int + }{ + {"configured", 2, 2}, + {"zero defaults", 0, 3}, + {"negative defaults", -1, 3}, + } { + t.Run(tc.name, func(t *testing.T) { + slots, err := ExpandWorkerSlots(nil, tc.count) + if err != nil || len(slots) != tc.want { + t.Fatalf("slots=%v err=%v, want %d", slots, err, tc.want) + } + for i, slot := range slots { + if slot.Name != "work#"+strconv.Itoa(i+1) || slot.ProfileID != 0 { + t.Fatalf("slot[%d]=%+v", i, slot) + } + } + }) + } +} + +func TestExpandWorkerSlotsPoolOrder(t *testing.T) { + slots, err := ExpandWorkerSlots([]WorkerPool{{ProfileID: 7, Workers: 2}, {ProfileID: 12, Workers: 3}}, 99) + if err != nil || len(slots) != 5 { + t.Fatalf("slots=%v err=%v", slots, err) + } + for i, slot := range slots { + want := int64(7) + if i >= 2 { + want = 12 + } + if slot.Name != "work#"+strconv.Itoa(i+1) || slot.ProfileID != want { + t.Fatalf("slot[%d]=%+v want profile %d", i, slot, want) + } + } +} + +func TestValidateWorkerPools(t *testing.T) { + for _, tc := range []struct { + name string + pools []WorkerPool + valid bool + }{ + {"empty", nil, true}, + {"bad profile", []WorkerPool{{ProfileID: 0, Workers: 1}}, false}, + {"bad workers", []WorkerPool{{ProfileID: 1, Workers: 0}}, false}, + {"duplicate", []WorkerPool{{ProfileID: 1, Workers: 1}, {ProfileID: 1, Workers: 1}}, false}, + {"too many", []WorkerPool{{ProfileID: 1, Workers: 51}}, false}, + {"valid", []WorkerPool{{ProfileID: 1, Workers: 1}}, true}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := ValidateWorkerPools(tc.pools) == nil; got != tc.valid { + t.Fatalf("valid=%v want %v", got, tc.valid) + } + }) + } +} + +func TestWorkerPoolSettingsRoundTrip(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable: %v", err) + } + defer m.Close() + pID, err := m.pg.SaveProfile(&db.LLMProfile{Name: "pool-test", Format: "openai", Model: "pool-test", APIKey: "pool-test-key"}) + if err != nil { + t.Fatal(err) + } + defer m.pg.DeleteProfile(pID) + oldPools, _, _ := m.pg.GetSetting(settingWorkerPools) + oldWorkers, workersSet, _ := m.pg.GetSetting(settingWorkers) + defer func() { + if workersSet { + _ = m.pg.SetSetting(settingWorkers, oldWorkers) + } else { + _, _ = m.pg.Exec("DELETE FROM settings WHERE key=$1", settingWorkers) + } + if oldPools != "" { + _ = m.pg.SetSetting(settingWorkerPools, oldPools) + } else { + _, _ = m.pg.Exec("DELETE FROM settings WHERE key=$1", settingWorkerPools) + } + }() + if err := m.SetWorkerPools([]WorkerPool{{ProfileID: 9223372036854775807, Workers: 1}}); err == nil { + t.Fatal("nonexistent profile accepted") + } + if err := m.SetWorkerPools([]WorkerPool{{ProfileID: pID, Workers: 2}}); err != nil { + t.Fatal(err) + } + got := m.WorkerPools() + if len(got) != 1 || got[0].ProfileID != pID || got[0].Workers != 2 || m.Workers() != 2 { + t.Fatalf("round trip pools=%+v workers=%d", got, m.Workers()) + } + if err := m.SetWorkerPools(nil); err != nil { + t.Fatal(err) + } + if got := m.WorkerPools(); got == nil || len(got) != 0 { + t.Fatalf("cleared pools=%+v", got) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s := New(ctx, m, t.TempDir(), t.TempDir(), t.TempDir()) + if _, err := m.pg.Exec("DELETE FROM settings WHERE key=$1", settingWorkerPools); err != nil { + t.Fatal(err) + } + res := httptest.NewRecorder() + s.getSettings(res, httptest.NewRequest(http.MethodGet, "/api/settings", nil)) + var unset map[string]any + if err := json.Unmarshal(res.Body.Bytes(), &unset); err != nil { + t.Fatal(err) + } + if pools, ok := unset["worker_pools"].([]any); !ok || pools == nil || len(pools) != 0 { + t.Fatalf("GET unset worker_pools=%#v, want empty array", unset["worker_pools"]) + } + put := httptest.NewRequest(http.MethodPut, "/api/settings", strings.NewReader(`{"workers":2,"worker_pools":[{"profile_id":"`+strconv.FormatInt(pID, 10)+`","workers":2}]}`)) + putResult := httptest.NewRecorder() + s.putSettings(putResult, put) + if putResult.Code != http.StatusOK { + t.Fatalf("PUT worker pools status=%d body=%s", putResult.Code, putResult.Body.String()) + } + if m.Workers() != 2 || len(m.WorkerPools()) != 1 { + t.Fatalf("PUT worker pools did not sync settings: workers=%d pools=%+v", m.Workers(), m.WorkerPools()) + } + workersOnly := httptest.NewRequest(http.MethodPut, "/api/settings", strings.NewReader(`{"workers":9}`)) + workersOnlyResult := httptest.NewRecorder() + s.putSettings(workersOnlyResult, workersOnly) + if workersOnlyResult.Code != http.StatusOK || m.Workers() != 2 { + t.Fatalf("workers-only PUT changed pool total: status=%d workers=%d", workersOnlyResult.Code, m.Workers()) + } + req := httptest.NewRequest(http.MethodGet, "/api/settings", nil) + res = httptest.NewRecorder() + s.getSettings(res, req) + if res.Code != http.StatusOK { + t.Fatalf("GET settings status=%d", res.Code) + } + var payload map[string]any + if err := json.Unmarshal(res.Body.Bytes(), &payload); err != nil { + t.Fatal(err) + } + pools, ok := payload["worker_pools"].([]any) + if !ok || len(pools) != 1 || pools[0].(map[string]any)["profile_id"] != strconv.FormatInt(pID, 10) { + t.Fatalf("GET worker_pools=%#v, want string profile id", payload["worker_pools"]) + } + clear := httptest.NewRequest(http.MethodPut, "/api/settings", strings.NewReader(`{"workers":4,"worker_pools":[]}`)) + clearResult := httptest.NewRecorder() + s.putSettings(clearResult, clear) + if clearResult.Code != http.StatusOK || m.Workers() != 4 || len(m.WorkerPools()) != 0 { + t.Fatalf("PUT legacy clear status=%d body=%s workers=%d pools=%+v", clearResult.Code, clearResult.Body.String(), m.Workers(), m.WorkerPools()) + } + res = httptest.NewRecorder() + s.getSettings(res, httptest.NewRequest(http.MethodGet, "/api/settings", nil)) + var cleared map[string]any + if err := json.Unmarshal(res.Body.Bytes(), &cleared); err != nil { + t.Fatal(err) + } + if pools, ok := cleared["worker_pools"].([]any); !ok || pools == nil || len(pools) != 0 { + t.Fatalf("GET cleared worker_pools=%#v, want empty array", cleared["worker_pools"]) + } +} + +func TestPinnedRuntimeMissingProfileDoesNotFallback(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable: %v", err) + } + defer m.Close() + task, err := m.CreateTask("pinned runtime test", "test", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer m.DeleteTask(task.ID, DeleteTaskOptions{}) + fallback, err := m.pg.SaveProfile(&db.LLMProfile{Name: "pinned-fallback", Format: "openai", Model: "fallback", APIKey: "fallback-key"}) + if err != nil { + t.Fatal(err) + } + defer m.pg.DeleteProfile(fallback) + if _, err := m.ReplaceTaskLLMProfiles(task.ID, []int64{fallback}, fallback); err != nil { + t.Fatal(err) + } + missing := int64(9223372036854775807) + s := &Server{m: m, llmOn: true, llmProv: &scriptedLLMProvider{}} + runtime := &taskLLMRuntime{s: s, taskID: task.ID, agentKey: "worker", pinProfileID: &missing} + if _, err := runtime.current(); err == nil { + t.Fatal("missing pinned profile silently fell back") + } + if _, ok := runtime.activeCfg(); ok || runtime.nonStreaming() || runtime.maxTokens() != 0 { + t.Fatal("missing pinned profile exposed fallback runtime settings") + } +} + +func TestPinnedWorkerLoopDoesNotConsultTaskRouterAndReportsFailureOnce(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable: %v", err) + } + defer m.Close() + task, err := m.CreateTask("pinned worker loop", "test", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer m.DeleteTask(task.ID, DeleteTaskOptions{}) + + e := NewEngine(m) + var routerCalls atomic.Int32 + e.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) { + routerCalls.Add(1) + return nil, nil + }) + e.SetWorkerResolver(func(*Task, WorkerSlot) (*agent.Worker, error) { + return nil, errors.New("profile unavailable") + }) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + e.workerLoop(ctx, task, WorkerSlot{Name: "work#4", ProfileID: 12}) + close(done) + }() + time.Sleep(100 * time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("worker loop did not stop") + } + if routerCalls.Load() != 0 { + t.Fatalf("pinned worker consulted task router %d times", routerCalls.Load()) + } + activities, _, err := task.Store.ActivityList(nil, 0, 20) + if err != nil { + t.Fatal(err) + } + var failures int + for _, activity := range activities { + if activity.Worker == "work#4" && activity.IsError { + failures++ + } + } + if failures != 1 { + t.Fatalf("pinned failure activities=%d, want one throttled entry: %+v", failures, activities) + } +} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index fea869f..84bfe45 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -14,7 +14,7 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@ import { Switch } from "@/components/ui/switch"; import { api } from "@/lib/api"; import { CHAT_SEND_MODE_OPTIONS, type ChatSendMode, setChatSendMode, useChatSendMode } from "@/lib/chat-send-mode"; -import type { Settings } from "@/lib/types"; +import type { LLMProfile, Settings, WorkerPool } from "@/lib/types"; import { BasicAuthCard } from "./_components/basic-auth-card"; import { UpdateCard } from "./_components/update-card"; @@ -45,6 +45,8 @@ export default function SystemSettingsPage() { const [pyInterp, setPyInterp] = React.useState(""); const [workers, setWorkers] = React.useState("3"); const [savingWorkers, setSavingWorkers] = React.useState(false); + const [workerPools, setWorkerPools] = React.useState([]); + const [llmProfiles, setLlmProfiles] = React.useState([]); // 操作约束注入范围(默认都开)。 const [injectPlanner, setInjectPlanner] = React.useState(true); const [injectWorker, setInjectWorker] = React.useState(true); @@ -67,6 +69,7 @@ export default function SystemSettingsPage() { setGlobalProxyInput(s.global_proxy ?? ""); setPyInterp(s.python_interpreter ?? ""); setWorkers(String(s.workers ?? 3)); + setWorkerPools(s.worker_pools); setInjectPlanner(s.constraints_inject_planner !== false); setInjectWorker(s.constraints_inject_worker !== false); setNoaCompaction(!!s.noa_compaction); @@ -74,13 +77,46 @@ export default function SystemSettingsPage() { const saveWorkers = () => { const n = Number(workers); + if (workerPools.length > 0) { + const total = workerPools.reduce((sum, pool) => sum + Number(pool.workers), 0); + if (workerPools.some((pool) => !pool.profile_id)) { + toast.error("请选择每个模型池的 LLM 配置"); + return; + } + if (workerPools.some((pool) => !Number.isInteger(Number(pool.workers)) || Number(pool.workers) <= 0)) { + toast.error("每个模型池的 Worker 数必须是大于 0 的整数"); + return; + } + if (total > 50) { + toast.error("Worker 总数不能超过 50"); + return; + } + const ids = workerPools.map((pool) => pool.profile_id); + if (new Set(ids).size !== ids.length) { + toast.error("不能重复选择同一个 LLM 配置"); + return; + } + setSavingWorkers(true); + api + .setSettings({ + workers: total, + worker_pools: workerPools.map((pool) => ({ ...pool, workers: Number(pool.workers) })), + }) + .then((s) => { + apply(s); + toast.success("已保存多模型 Worker 池（对之后启动的任务生效）"); + }) + .catch((e) => toast.error(`保存失败：${(e as Error).message}`)) + .finally(() => setSavingWorkers(false)); + return; + } if (!Number.isInteger(n) || n <= 0) { toast.error("并发数必须是大于 0 的整数"); return; } setSavingWorkers(true); api - .setSettings({ workers: n }) + .setSettings({ workers: n, worker_pools: [] }) .then((s) => { apply(s); toast.success("已保存并发工作 agent 数（对之后启动的任务生效）"); @@ -89,6 +125,8 @@ export default function SystemSettingsPage() { .finally(() => setSavingWorkers(false)); }; + const addWorkerPool = () => setWorkerPools((pools) => [...pools, { profile_id: "", workers: 1 }]); + const savePython = () => { setSaving(true); api @@ -117,6 +155,13 @@ export default function SystemSettingsPage() { .finally(() => setLoaded(true)); }, [apply]); + React.useEffect(() => { + api + .llmProfiles() + .then(setLlmProfiles) + .catch(() => setLlmProfiles([])); + }, []); + const toggleTraffic = (v: boolean) => { setTrafficCapture(v); // optimistic setSaving(true); @@ -756,6 +801,73 @@ export default function SystemSettingsPage() { +
+
+ + +
+ {workerPools.map((pool, index) => { + const selected = llmProfiles.find((profile) => profile.id === pool.profile_id); + const placeholder = + selected?.name || (pool.profile_id ? `配置 #${pool.profile_id} 已删除/不可用` : "选择 LLM 配置"); + return ( + // biome-ignore lint/suspicious/noArrayIndexKey: pool rows are edited in place and only deleted. +
+ + + setWorkerPools((current) => + current.map((item, itemIndex) => + itemIndex === index ? { ...item, workers: Number(event.target.value) } : item, + ), + ) + } + /> + +
+ ); + })} + {workerPools.length > 0 ? ( +
+ 当前合计：{workerPools.reduce((sum, pool) => sum + (Number(pool.workers) || 0), 0)} 个 Worker 会话 +
+ ) : null} +
0} onChange={(e) => setWorkers(e.target.value)} /> + {workerPools.length > 0 ? ( + 已配置模型池时由各池数量决定并发。 + ) : null} diff --git a/web/src/lib/mock/data.ts b/web/src/lib/mock/data.ts index 531df7b..9b8a69c 100644 --- a/web/src/lib/mock/data.ts +++ b/web/src/lib/mock/data.ts @@ -3338,6 +3338,7 @@ export const settings: Settings = { global_proxy: "", python_interpreter: "/usr/bin/python3", workers: 3, + worker_pools: [], llm_pool_enabled: true, llm_pool_bind_fallback: false, noa_compaction: false, diff --git a/web/src/lib/mock/handler.ts b/web/src/lib/mock/handler.ts index 2f15a69..185d5a7 100644 --- a/web/src/lib/mock/handler.ts +++ b/web/src/lib/mock/handler.ts @@ -2249,7 +2249,15 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re if (path === "/traffic") return D.traffic; if (path === "/traffic/exchange") return D.trafficDetail; if (path === "/settings" && m === "GET") return D.settings; - if (path === "/settings" && m === "PUT") return { ...D.settings, ...b }; + if (path === "/settings" && m === "PUT") { + const next = { ...D.settings, ...b }; + if (!Array.isArray(next.worker_pools)) next.worker_pools = D.settings.worker_pools; + if (next.worker_pools.length > 0) { + next.workers = next.worker_pools.reduce((sum, pool) => sum + Number(pool.workers), 0); + } + Object.assign(D.settings, next); + return next; + } if (path === "/settings/basic-auth" && m === "GET") return { enabled: false, username: "", password_set: false }; if (path === "/settings/basic-auth" && m === "PUT") throw new Error("演示模式不支持修改 HTTP Basic Auth，请在实际部署中配置"); @@ -2467,7 +2475,8 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re const from = q.get("from") || ""; const to = q.get("to") || ""; if (status && !["pending", "allowed", "denied", "timeout"].includes(status)) throw new Error("无效审批状态"); - if (decisionSource && !["model", "rule", "manual", "unknown"].includes(decisionSource)) throw new Error("无效判定来源"); + if (decisionSource && !["model", "rule", "manual", "unknown"].includes(decisionSource)) + throw new Error("无效判定来源"); if ([...search].length > 200) throw new Error("搜索关键词不能超过 200 个字符"); if (search.startsWith("#") && !/^#[1-9]\d*$/.test(search)) throw new Error("审批编号必须为 # 后接正整数"); for (const value of [from, to]) { diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 54e26a6..e080c12 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -968,6 +968,7 @@ export interface Settings { global_proxy?: string; python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测) workers?: number; // 并发工作 agent 数(默认3)；对之后启动的任务生效 + worker_pools: WorkerPool[]; // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 task_concurrency_enabled?: boolean; // 默认 false task_concurrency_limit?: number; // 开启后默认 5 @@ -989,6 +990,11 @@ export interface Settings { notify_digest_interval_min?: number; // 汇总模式周期（分钟），默认 30 } +export interface WorkerPool { + profile_id: string; + workers: number; +} + // ---- 漏洞 IM 推送 ---- // NotificationFilter 是渠道的过滤条件，字段全部可选，缺省即不过滤。 From 06801bc5aaf6939aeb2a33074e1b76300c51a3d5 Mon Sep 17 00:00:00 2001 From: RuoJi6 Date: Sat, 24 Oct 2026 22:55:34 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix:=20Worker=20=E6=B1=A0=E8=AF=84=E5=AE=A1?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=88=E7=B3=BB=E7=BB=9F=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=E5=8F=AF=E8=A7=81=E6=8A=A5=E9=94=99=E3=80=81=E6=9A=82=E5=81=9C?= =?UTF-8?q?=E6=84=8F=E5=9B=BE=E7=BB=AD=E8=B7=91=E4=BF=9D=E6=8C=81=E7=BB=91?= =?UTF-8?q?=E5=AE=9A=E6=A8=A1=E5=9E=8B=E3=80=81=E5=8A=A0=E8=BD=BD=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E4=B8=8D=E6=B8=85=E7=A9=BA=E6=B1=A0=E3=80=81=E5=8E=9F?= =?UTF-8?q?=E5=AD=90=E5=86=99=E5=85=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code authored by Codex CLI gpt-6.1-sol (addresses Grok review #1). --- db/settings.go | 27 ++++ db/settings_test.go | 28 +++++ server/engine.go | 39 +++++- server/manager.go | 27 ++-- server/server.go | 12 +- server/worker_pools_test.go | 132 +++++++++++++++++++- web/src/app/(main)/system/settings/page.tsx | 34 ++++- web/src/lib/mock/handler.ts | 3 +- 8 files changed, 270 insertions(+), 32 deletions(-) create mode 100644 db/settings_test.go diff --git a/db/settings.go b/db/settings.go index f8fafcc..5c3727c 100644 --- a/db/settings.go +++ b/db/settings.go @@ -3,6 +3,7 @@ package db import ( "context" "database/sql" + "sort" ) // Settings is a tiny key-value store for global app config the UI toggles at @@ -33,6 +34,32 @@ ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`, key return err } +// SetSettingsAtomic upserts several settings in one transaction. Keys are +// sorted so concurrent callers acquire row locks in a deterministic order. +func (d *DB) SetSettingsAtomic(ctx context.Context, values map[string]string) error { + if len(values) == 0 { + return nil + } + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck // commit makes this a no-op + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + if _, err := tx.ExecContext(ctx, ` +INSERT INTO settings(key, value) VALUES ($1, $2) +ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`, key, values[key]); err != nil { + return err + } + } + return tx.Commit() +} + // InsertSettingIfAbsent 只在 key 尚未存在时写入，已存在则原样保留并返回 inserted=false。 // 给 auth.password_hash 这类"只允许首次设置"的键用：判定落在数据库的主键约束上， // 调用方先 GetSetting 再写的那种检查就只是快速失败路径——读出错或并发撞车时， diff --git a/db/settings_test.go b/db/settings_test.go new file mode 100644 index 0000000..9de68a9 --- /dev/null +++ b/db/settings_test.go @@ -0,0 +1,28 @@ +package db + +import ( + "context" + "fmt" + "testing" + "time" +) + +func TestSetSettingsAtomic(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v)", err) + } + defer d.Close() + prefix := fmt.Sprintf("test.atomic_settings.%d", time.Now().UnixNano()) + defer func() { _, _ = d.Exec(`DELETE FROM settings WHERE key LIKE $1`, prefix+".%") }() + values := map[string]string{prefix + ".worker_pools": "[]", prefix + ".workers": "3"} + if err := d.SetSettingsAtomic(context.Background(), values); err != nil { + t.Fatal(err) + } + for key, want := range values { + got, ok, err := d.GetSetting(key) + if err != nil || !ok || got != want { + t.Fatalf("key %q: value=%q ok=%v err=%v, want %q", key, got, ok, err, want) + } + } +} diff --git a/server/engine.go b/server/engine.go index 9a8c07b..be30998 100644 --- a/server/engine.go +++ b/server/engine.go @@ -150,6 +150,7 @@ type Engine struct { resolveAuthoritative bool workerResolve func(t *Task, slot WorkerSlot) (*agent.Worker, error) slotState sync.Map // task/slot -> slotAvailability + slotPlans sync.Map // task -> map[slot name]profile ID; immutable for a run // readiness reports whether a global LLM provider is configured — the signal behind // Ready()/the llm_configured indicator. Wired once at startup; nil → not ready. readiness func() bool @@ -292,6 +293,7 @@ func (e *Engine) StopTask(taskID string) { e.stamped.Delete(taskID) e.inflight.Delete(taskID) e.coordStarted.Delete(taskID) + e.slotPlans.Delete(taskID) e.slotState.Range(func(key, _ any) bool { keyText, ok := key.(string) if ok && strings.HasPrefix(keyText, taskID+"/") { @@ -741,7 +743,7 @@ func (e *Engine) reportSlotAvailability(t *Task, slot WorkerSlot, err error) { if err == nil { if loaded && !state.available { e.slotState.Store(key, slotAvailability{available: true, lastEmit: now}) - e.emitActivity(t, db.Activity{Worker: slot.Name, Kind: "text", Summary: slot.Name + " 绑定的 LLM 配置已恢复，可以继续领取任务"}) + e.emitActivity(t, db.Activity{Worker: "system", Kind: "text", Summary: fmt.Sprintf("%s 绑定的 LLM 配置 #%d 已恢复，可以继续领取任务", slot.Name, slot.ProfileID)}) } return } @@ -756,7 +758,7 @@ func (e *Engine) reportSlotAvailability(t *Task, slot WorkerSlot, err error) { e.slotState.Store(key, state) } summary := fmt.Sprintf("%s 绑定的 LLM 配置 #%d 不可用（已删除或配置无效），该会话暂停领取任务，不会回退到默认模型", slot.Name, slot.ProfileID) - e.emitActivity(t, db.Activity{Worker: slot.Name, Kind: "text", IsError: true, Summary: summary, Detail: err.Error()}) + e.emitActivity(t, db.Activity{Worker: "system", Kind: "text", IsError: true, Summary: summary, Detail: err.Error()}) log.Printf("[worker %s] task %s pinned profile #%d unavailable: %v", slot.Name, t.ID, slot.ProfileID, err) } @@ -807,6 +809,11 @@ func (e *Engine) Run(ctx context.Context, t *Task) { return } rt := e.registerTaskRoutines(ctx, t.ID, 1+len(slots)) + plan := make(map[string]int64, len(slots)) + for _, slot := range slots { + plan[slot.Name] = slot.ProfileID + } + e.slotPlans.Store(t.ID, plan) e.deleteMu.RUnlock() e.touch(t.ID) runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) }) @@ -1245,10 +1252,6 @@ func (e *Engine) runDetachedIntent(ctx context.Context, t *Task, intentID int64, e.decInflight(t.ID) } }() - _, worker := e.snapshotFor(t) - if worker == nil { - return fmt.Errorf("worker 尚未就绪") - } node, err := t.Store.GetNode(intentID) if err != nil { return err @@ -1256,6 +1259,30 @@ func (e *Engine) runDetachedIntent(ctx context.Context, t *Task, intentID int64, if node == nil || node.Kind != db.KindIntent { return fmt.Errorf("intent not found") } + owner := node.Owner + var worker *agent.Worker + if value, ok := e.slotPlans.Load(t.ID); ok { + if plan, ok := value.(map[string]int64); ok { + if profileID, pinned := plan[owner]; pinned && profileID > 0 { + if e.workerResolve == nil { + return fmt.Errorf("意图所属的 LLM 配置 #%d 不可用：pinned worker resolver unavailable", profileID) + } + worker, err = e.workerResolve(t, WorkerSlot{Name: owner, ProfileID: profileID}) + if err != nil || worker == nil { + if err == nil { + err = errors.New("worker resolver returned no worker") + } + return fmt.Errorf("意图所属的 LLM 配置 #%d 不可用：%w", profileID, err) + } + } + } + } + if worker == nil { + _, worker = e.snapshotFor(t) + if worker == nil { + return fmt.Errorf("worker 尚未就绪") + } + } changed, err := t.Store.CompareAndSetIntentState(intentID, "paused", "running") if err != nil { return err diff --git a/server/manager.go b/server/manager.go index 76f72ae..b2ff3c7 100644 --- a/server/manager.go +++ b/server/manager.go @@ -1,6 +1,7 @@ package server import ( + "context" "database/sql" "encoding/json" "errors" @@ -444,9 +445,10 @@ func (m *Manager) WorkerPools() []WorkerPool { return pools } -// SetWorkerPools validates profile references at save time and synchronizes the -// legacy workers setting with the pool total. An empty slice restores legacy mode. -func (m *Manager) SetWorkerPools(pools []WorkerPool) error { +// SetWorkerPools validates profile references at save time and atomically +// synchronizes the legacy workers setting with the pool total. An empty slice +// restores legacy mode; an optional legacy count is used when clearing pools. +func (m *Manager) SetWorkerPools(pools []WorkerPool, legacyWorkers ...int) error { if pools == nil { pools = []WorkerPool{} } @@ -468,17 +470,22 @@ func (m *Manager) SetWorkerPools(pools []WorkerPool) error { if err != nil { return err } - if err := m.pg.SetSetting(settingWorkerPools, string(b)); err != nil { - return err - } + workerCount := m.Workers() if len(pools) > 0 { - total := 0 + workerCount = 0 for _, pool := range pools { - total += pool.Workers + workerCount += pool.Workers } - return m.pg.SetSetting(settingWorkers, strconv.Itoa(total)) + } else if len(legacyWorkers) > 0 { + if legacyWorkers[0] <= 0 { + return fmt.Errorf("workers 必须 >0") + } + workerCount = legacyWorkers[0] } - return nil + return m.pg.SetSettingsAtomic(context.Background(), map[string]string{ + settingWorkerPools: string(b), + settingWorkers: strconv.Itoa(workerCount), + }) } // SetWorkers persists the concurrent work-agent count. Values <=0 are rejected. diff --git a/server/server.go b/server/server.go index fc3e0b5..71cf168 100644 --- a/server/server.go +++ b/server/server.go @@ -3593,16 +3593,14 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { writeErr(w, 400, "workers 必须 >0") return } - if err := s.m.SetWorkerPools(*req.WorkerPools); err != nil { + legacyWorkers := []int(nil) + if req.Workers != nil { + legacyWorkers = []int{*req.Workers} + } + if err := s.m.SetWorkerPools(*req.WorkerPools, legacyWorkers...); err != nil { writeErr(w, 400, err.Error()) return } - if len(*req.WorkerPools) == 0 && req.Workers != nil { - if err := s.m.SetWorkers(*req.Workers); err != nil { - writeErr(w, 400, err.Error()) - return - } - } } else if req.Workers != nil { // An omitted worker_pools field leaves the saved mode unchanged. If a // non-empty pool configuration is already saved, its total remains the diff --git a/server/worker_pools_test.go b/server/worker_pools_test.go index 36f40aa..4facf4a 100644 --- a/server/worker_pools_test.go +++ b/server/worker_pools_test.go @@ -14,6 +14,7 @@ import ( "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/db" + "github.com/Autumn-27/norma/llm" ) func TestExpandWorkerSlotsLegacy(t *testing.T) { @@ -210,6 +211,51 @@ func TestPinnedRuntimeMissingProfileDoesNotFallback(t *testing.T) { } } +func TestPinnedRuntimeQuotaDoesNotAdvanceTaskFailover(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable: %v", err) + } + defer m.Close() + task, err := m.CreateTask("pinned quota", "test", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer m.DeleteTask(task.ID, DeleteTaskOptions{}) + profileID, err := m.pg.SaveProfile(&db.LLMProfile{Name: "pinned-quota", Format: "openai", Model: "pinned", APIKey: "key"}) + if err != nil { + t.Fatal(err) + } + defer m.pg.DeleteProfile(profileID) + if _, err := m.ReplaceTaskLLMProfiles(task.ID, []int64{profileID}, profileID); err != nil { + t.Fatal(err) + } + before, err := m.pg.GetTask(taskIDInt(t, task.ID)) + if err != nil || before == nil || before.ActiveLLMProfileID == nil { + t.Fatalf("load task failover state: task=%+v err=%v", before, err) + } + provider := &scriptedLLMProvider{err: errors.New("insufficient_quota")} + s := &Server{m: m, provByProfile: map[int64]*provEntry{profileID: {prov: provider, cfg: agent.Config{}}}} + id := profileID + runtime := &taskLLMRuntime{s: s, taskID: task.ID, agentKey: "worker", pinProfileID: &id} + if _, _, _, err := runtime.Complete(context.Background(), llm.CompletionRequest{}); err == nil { + t.Fatal("pinned quota error unexpectedly succeeded") + } + after, err := m.pg.GetTask(taskIDInt(t, task.ID)) + if err != nil || after == nil || after.LLMChainRevision != before.LLMChainRevision || after.ActiveLLMProfileID == nil || *after.ActiveLLMProfileID != *before.ActiveLLMProfileID { + t.Fatalf("pinned quota changed task failover state: before=%+v after=%+v err=%v", before, after, err) + } +} + +func taskIDInt(t *testing.T, id string) int64 { + t.Helper() + n, err := strconv.ParseInt(id, 10, 64) + if err != nil { + t.Fatal(err) + } + return n +} + func TestPinnedWorkerLoopDoesNotConsultTaskRouterAndReportsFailureOnce(t *testing.T) { m, err := NewManager(t.TempDir(), "") if err != nil { @@ -253,7 +299,7 @@ func TestPinnedWorkerLoopDoesNotConsultTaskRouterAndReportsFailureOnce(t *testin } var failures int for _, activity := range activities { - if activity.Worker == "work#4" && activity.IsError { + if activity.Worker == "system" && activity.IsError { failures++ } } @@ -261,3 +307,87 @@ func TestPinnedWorkerLoopDoesNotConsultTaskRouterAndReportsFailureOnce(t *testin t.Fatalf("pinned failure activities=%d, want one throttled entry: %+v", failures, activities) } } + +func TestPinnedWorkerAvailabilityRecoveryUsesSystemSession(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable: %v", err) + } + defer m.Close() + task, err := m.CreateTask("pinned availability", "test", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer m.DeleteTask(task.ID, DeleteTaskOptions{}) + e := NewEngine(m) + slot := WorkerSlot{Name: "work#4", ProfileID: 12} + e.reportSlotAvailability(task, slot, errors.New("profile unavailable")) + e.reportSlotAvailability(task, slot, nil) + activities, _, err := task.Store.ActivityList(nil, 0, 20) + if err != nil { + t.Fatal(err) + } + var failure, recovery int + for _, activity := range activities { + if activity.Worker != "system" { + continue + } + if activity.IsError && strings.Contains(activity.Summary, "work#4") && strings.Contains(activity.Summary, "#12") { + failure++ + } + if !activity.IsError && strings.Contains(activity.Summary, "work#4") && strings.Contains(activity.Summary, "#12") { + recovery++ + } + } + if failure != 1 || recovery != 1 { + t.Fatalf("system availability activities failure=%d recovery=%d: %+v", failure, recovery, activities) + } +} + +func TestDetachedIntentPinnedOwnerUsesPinnedResolver(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable: %v", err) + } + defer m.Close() + task, err := m.CreateTask("detached pinned owner", "test", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer m.DeleteTask(task.ID, DeleteTaskOptions{}) + intentID, err := task.Store.AddIntent(map[string]any{"text": "continue"}, 0, nil, "test") + if err != nil { + t.Fatal(err) + } + if ok, err := task.Store.ClaimIntent(intentID, "work#4"); err != nil || !ok { + t.Fatalf("claim intent: ok=%v err=%v", ok, err) + } + if changed, err := task.Store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed { + t.Fatalf("pause intent: changed=%v err=%v", changed, err) + } + e := NewEngine(m) + e.slotPlans.Store(task.ID, map[string]int64{"work#4": 12}) + var pinnedCalls, snapshotCalls atomic.Int32 + e.SetWorkerResolver(func(_ *Task, slot WorkerSlot) (*agent.Worker, error) { + pinnedCalls.Add(1) + if slot.Name != "work#4" || slot.ProfileID != 12 { + t.Fatalf("unexpected pinned slot: %+v", slot) + } + return nil, errors.New("profile unavailable") + }) + e.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) { + snapshotCalls.Add(1) + return nil, nil + }) + err = e.runDetachedIntent(context.Background(), task, intentID, "request", "human", "agent") + if err == nil || !strings.Contains(err.Error(), "LLM 配置 #12 不可用") { + t.Fatalf("runDetachedIntent err=%v", err) + } + if pinnedCalls.Load() != 1 || snapshotCalls.Load() != 0 { + t.Fatalf("pinned calls=%d snapshot calls=%d", pinnedCalls.Load(), snapshotCalls.Load()) + } + node, err := task.Store.GetNode(intentID) + if err != nil || node == nil || node.State != "paused" { + t.Fatalf("intent changed after unavailable resolver: node=%+v err=%v", node, err) + } +} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index 84bfe45..535e9a0 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -40,6 +40,7 @@ export default function SystemSettingsPage() { const [savingGlobalProxy, setSavingGlobalProxy] = React.useState(false); const [testing, setTesting] = React.useState(false); const [loaded, setLoaded] = React.useState(false); + const [settingsLoaded, setSettingsLoaded] = React.useState(false); const [saving, setSaving] = React.useState(false); const [savingKey, setSavingKey] = React.useState(false); const [pyInterp, setPyInterp] = React.useState(""); @@ -69,13 +70,17 @@ export default function SystemSettingsPage() { setGlobalProxyInput(s.global_proxy ?? ""); setPyInterp(s.python_interpreter ?? ""); setWorkers(String(s.workers ?? 3)); - setWorkerPools(s.worker_pools); + setWorkerPools(s.worker_pools ?? []); setInjectPlanner(s.constraints_inject_planner !== false); setInjectWorker(s.constraints_inject_worker !== false); setNoaCompaction(!!s.noa_compaction); }, []); const saveWorkers = () => { + if (!settingsLoaded) { + toast.error("设置加载失败，请刷新后再保存 Worker 配置"); + return; + } const n = Number(workers); if (workerPools.length > 0) { const total = workerPools.reduce((sum, pool) => sum + Number(pool.workers), 0); @@ -150,8 +155,14 @@ export default function SystemSettingsPage() { React.useEffect(() => { api .settings() - .then(apply) - .catch(() => undefined) + .then((s) => { + apply(s); + setSettingsLoaded(true); + }) + .catch(() => { + setSettingsLoaded(false); + toast.error("设置加载失败，Worker 配置已锁定，避免覆盖服务器上的模型池"); + }) .finally(() => setLoaded(true)); }, [apply]); @@ -804,7 +815,12 @@ export default function SystemSettingsPage() {
-
@@ -817,6 +833,7 @@ export default function SystemSettingsPage() {