From 8047bb7aca8aee46b90068a92056f7e2c4927000 Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 19 Aug 2026 10:05:01 -0400 Subject: [PATCH] =?UTF-8?q?feat(tasks):=20=E4=BB=BB=E5=8A=A1=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E4=B8=8A=E9=99=90(=E5=8F=AF=E9=80=89,=E8=B6=85?= =?UTF-8?q?=E9=99=90=E6=8E=92=E9=98=9F=E8=87=AA=E5=8A=A8=E8=A1=A5=E4=BD=8D?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 任务页新建任务左侧加「并发设置」齿轮:开关(默认关) + 并发数量(开启默认5)。 开启后新建任务若「运行中」数已达上限,进入「排队中」不启动;有空位时后台 每 5s 自动把最早排队的任务补位启动。 - db: tasks 加持久化 queued 列(schema + ALTER 补旧库);Task.Queued/SetQueued。 - manager: ConcurrencyLimit/SetConcurrency(settings KV);SetTaskQueued(落库+同步内存)。 - 准入与促进器(goals.go):launchTask 统一准入,满则挂起排队;抽出 startTaskEngine 复用;occupiesSlot/runningTaskCount/queueIfAtCapacity;reconcileConcurrency 每 scheduler tick 补位(关闭则放行全部排队),concMu 串行化防超发;启动重载/setActive/ reviveTask 三处'排队则不启动'门。 - settings: /api/settings 增 task_concurrency_enabled/limit;PUT 后立即协调一次。 - web: 齿轮 Dialog(Switch+数量);新增 queued/「排队中」状态(类型/状态色/筛选)。 - 超限只排队新任务,不打断在跑的;排队任务不消耗 LLM(目标拆解推迟到出队)。 go build/vet、tsc 通过;DB queued 往返测试通过。 --- db/schema.sql | 3 + db/tasks.go | 12 +- db/tasks_test.go | 17 +++ server/goals.go | 128 ++++++++++++++++++--- server/manager.go | 54 ++++++++- server/scheduler.go | 1 + server/server.go | 60 ++++++++-- web/src/app/(main)/function/tasks/page.tsx | 109 +++++++++++++++++- web/src/lib/status.ts | 1 + web/src/lib/types.ts | 5 +- 10 files changed, 360 insertions(+), 30 deletions(-) diff --git a/db/schema.sql b/db/schema.sql index 380ee9d..053cfe5 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -234,6 +234,7 @@ CREATE TABLE IF NOT EXISTS tasks ( status TEXT NOT NULL DEFAULT 'created' CHECK (status IN ('created','running','paused','done','failed','timeout')), paused BOOLEAN NOT NULL DEFAULT false, + queued BOOLEAN NOT NULL DEFAULT false, llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL, company_id BIGINT REFERENCES companies(id) ON DELETE SET NULL, parent_ref TEXT, @@ -253,6 +254,8 @@ CREATE TRIGGER trg_tasks_upd BEFORE UPDATE ON tasks FOR EACH ROW EXECUTE FUNCTION set_updated_at(); -- planner 心跳触发间隔(秒);补旧库。默认 300s(5min)。见 docs/planner-trigger-impl-plan.md ALTER TABLE tasks ADD COLUMN IF NOT EXISTS plan_heartbeat_seconds INTEGER NOT NULL DEFAULT 300; +-- 并发上限挂起态;补旧库。true=因并发上限排队、等待空位自动启动。 +ALTER TABLE tasks ADD COLUMN IF NOT EXISTS queued BOOLEAN NOT NULL DEFAULT false; -- 任务测试范围（资产覆盖度的分母 + 授权边界）。 -- 自动填(source='auto')：insertAssets 顶层按 worker 显式插入的资产类型加保守范围 diff --git a/db/tasks.go b/db/tasks.go index 03e15e3..86dfac6 100644 --- a/db/tasks.go +++ b/db/tasks.go @@ -13,6 +13,7 @@ type Task struct { ExplorationID int64 `json:"exploration_id"` Status string `json:"status"` Paused bool `json:"paused"` + Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑 LLMProfileID *int64 `json:"llm_profile_id,omitempty"` ParentRef string `json:"parent_ref,omitempty"` // 父任务 id(编排 spawn 记录;空=顶层) CreatedAt time.Time `json:"created_at"` @@ -84,11 +85,11 @@ RETURNING id, status, paused, created_at`, description, goal, expID, llmProfileI return t, tx.Commit() } -const taskCols = `id, description, goal, exploration_id, status, paused, llm_profile_id, COALESCE(parent_ref,''), created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), first_run_at, deadline_at` +const taskCols = `id, description, goal, exploration_id, status, paused, queued, llm_profile_id, COALESCE(parent_ref,''), created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), first_run_at, deadline_at` func scanTask(sc interface{ Scan(...any) error }) (*Task, error) { var t Task - if err := sc.Scan(&t.ID, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.LLMProfileID, &t.ParentRef, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.FirstRunAt, &t.DeadlineAt); err != nil { + if err := sc.Scan(&t.ID, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.LLMProfileID, &t.ParentRef, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.FirstRunAt, &t.DeadlineAt); err != nil { return nil, err } return &t, nil @@ -136,6 +137,13 @@ func (d *DB) SetPaused(id int64, paused bool) error { return err } +// SetQueued persists a task's queued flag (并发上限挂起态). A queued task is held +// (engine not started) until the concurrency reconciler promotes it into a free slot. +func (d *DB) SetQueued(id int64, queued bool) error { + _, err := d.Exec(`UPDATE tasks SET queued=$1 WHERE id=$2`, queued, id) + return err +} + // SetStatus updates a task's lifecycle status. Entering a terminal state // (done/failed/timeout) stamps completed_at once (COALESCE keeps the first stamp // stable); moving back to a non-terminal state clears it, so a re-run has no stale diff --git a/db/tasks_test.go b/db/tasks_test.go index f2e9def..d407e03 100644 --- a/db/tasks_test.go +++ b/db/tasks_test.go @@ -27,6 +27,23 @@ func TestTaskLifecycleAndDeleteCascade(t *testing.T) { if err != nil || got == nil || !got.Paused { t.Fatalf("paused not persisted: %+v err=%v", got, err) } + if got.Queued { + t.Fatalf("new task should not be queued: %+v", got) + } + + // queued (concurrency-hold) flag round-trips independently of paused + if err := d.SetQueued(tk.ID, true); err != nil { + t.Fatal(err) + } + if g, _ := d.GetTask(tk.ID); g == nil || !g.Queued { + t.Fatalf("queued not persisted: %+v", g) + } + if err := d.SetQueued(tk.ID, false); err != nil { + t.Fatal(err) + } + if g, _ := d.GetTask(tk.ID); g == nil || g.Queued { + t.Fatalf("queued not cleared: %+v", g) + } // list contains it list, _ := d.ListTasks() diff --git a/server/goals.go b/server/goals.go index 09e23b1..2a72c60 100644 --- a/server/goals.go +++ b/server/goals.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "sort" "strconv" "strings" @@ -29,21 +30,116 @@ func (s *Server) launchTask(t *Task, seedText string, seedFirstIntent bool) { if seedFirstIntent { s.seedFirstIntent(t) } - go func() { - s.engine.emitActivity(t, db.Activity{Worker: "planner", Kind: "round", - Summary: "第 0 轮目标拆解"}) - goals := s.createGoals(s.ctx, t, func(r db.Activity) { - s.engine.emitActivity(t, r) - }) - for _, g := range goals { - summary := g.Text - if g.VulnClass != "" { - summary = fmt.Sprintf("[%s] %s", g.VulnClass, g.Text) - } - s.engine.emitActivity(t, db.Activity{Worker: "planner", Kind: "text", Summary: summary}) + // 并发上限:满了就挂起排队(不做目标拆解、不启动引擎),由 reconcileConcurrency 补位启动。 + if s.queueIfAtCapacity(t) { + return + } + go s.startTaskEngine(t) +} + +// startTaskEngine does the actual running work: 第0轮目标拆解 + 启动引擎循环。 +// Shared by fresh launches and by the concurrency reconciler promoting a queued task. +func (s *Server) startTaskEngine(t *Task) { + s.engine.emitActivity(t, db.Activity{Worker: "planner", Kind: "round", + Summary: "第 0 轮目标拆解"}) + goals := s.createGoals(s.ctx, t, func(r db.Activity) { + s.engine.emitActivity(t, r) + }) + for _, g := range goals { + summary := g.Text + if g.VulnClass != "" { + summary = fmt.Sprintf("[%s] %s", g.VulnClass, g.Text) } - s.engine.Run(s.ctx, t) - }() + s.engine.emitActivity(t, db.Activity{Worker: "planner", Kind: "text", Summary: summary}) + } + s.engine.Run(s.ctx, t) +} + +// occupiesSlot reports whether a task counts against the concurrency cap: it is +// live (not terminal), not queued, and not paused. A just-created / decomposing +// task counts too (it will run momentarily). +func (s *Server) occupiesSlot(t *Task) bool { + if t == nil || t.Queued || isTerminalStatus(t.Status) { + return false + } + return !t.Paused && !s.engine.IsPaused(t.ID) +} + +// runningTaskCount counts tasks occupying a concurrency slot, excluding excludeID +// (pass the candidate's id when deciding admission so it doesn't count itself). +func (s *Server) runningTaskCount(excludeID string) int { + n := 0 + for _, t := range s.m.List() { + if t.ID == excludeID { + continue + } + if s.occupiesSlot(t) { + n++ + } + } + return n +} + +// queueIfAtCapacity holds a task in the queued state when the concurrency cap is +// enabled and already full. Returns true if the task was queued (caller must not +// start it). No-op (false) when the feature is off or a slot is free. +func (s *Server) queueIfAtCapacity(t *Task) bool { + s.concMu.Lock() + defer s.concMu.Unlock() + enabled, limit := s.m.ConcurrencyLimit() + if !enabled || s.runningTaskCount(t.ID) < limit { + return false + } + if err := s.m.SetTaskQueued(t.ID, true); err != nil { + log.Printf("[concurrency] task %s 置排队失败: %v", t.ID, err) + return false // 落库失败宁可放它跑,也别静默丢任务 + } + t.Queued = true + s.engine.emitActivity(t, db.Activity{Worker: "planner", Kind: "text", + Summary: fmt.Sprintf("已排队：达到并发上限 %d，等待空位后自动开始", limit)}) + log.Printf("[concurrency] task %s 排队(并发上限 %d)", t.ID, limit) + return true +} + +// reconcileConcurrency promotes queued tasks into free slots. Runs on every +// scheduler tick. When the feature is disabled it releases ALL queued tasks; +// when enabled it starts the oldest-queued tasks up to (limit - running). +func (s *Server) reconcileConcurrency() { + s.concMu.Lock() + defer s.concMu.Unlock() + var queued []*Task + for _, t := range s.m.List() { + if t.Queued && !isTerminalStatus(t.Status) { + queued = append(queued, t) + } + } + if len(queued) == 0 { + return + } + // 最早排队的先启动。 + sort.Slice(queued, func(i, j int) bool { return queued[i].CreatedAt < queued[j].CreatedAt }) + + enabled, limit := s.m.ConcurrencyLimit() + slots := len(queued) // feature off → 全部放行 + if enabled { + if !s.engine.Ready() { + return // 引擎未就绪(无 LLM):继续挂起,别空跑 + } + slots = limit - s.runningTaskCount("") + } + for _, t := range queued { + if slots <= 0 { + break + } + if err := s.m.SetTaskQueued(t.ID, false); err != nil { + log.Printf("[concurrency] task %s 解除排队失败: %v", t.ID, err) + continue + } + t.Queued = false + log.Printf("[concurrency] task %s 出队启动", t.ID) + go s.startTaskEngine(t) + slots-- + } } // reviveTask 让一个已停下的任务重新跑起来:把终态(done/failed/timeout)拉回 running、 @@ -56,6 +152,10 @@ func (s *Server) reviveTask(t *Task) { if t == nil { return } + // 排队中的任务不复活启动——它本就等待并发空位,交给 reconcileConcurrency。 + if t.Queued { + return + } // 终态 → 拉回 running(SetTaskStatus 会清 completed_at 并同步内存 handle 的 Status, // 使 planner/worker 循环的终态门放行)。 if isTerminalStatus(t.Status) { diff --git a/server/manager.go b/server/manager.go index 63b4091..454b229 100644 --- a/server/manager.go +++ b/server/manager.go @@ -32,6 +32,7 @@ type Task struct { CreatedAt int64 `json:"created_at"` CompletedAt int64 `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成 Paused bool `json:"paused"` + Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑 ParentRef string `json:"parent_ref,omitempty"` // 父任务 id(编排 spawn 记录) LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // 指定运行本任务 planner/worker 的 LLM 配置;nil=用全局激活配置 Status string `json:"status"` // persisted lifecycle status (done/failed/timeout 为终态；空/其它则由运行态推导) @@ -85,12 +86,45 @@ const ( settingWebSearchProxy = "web_search_proxy" settingWorkers = "workers" settingLLMRecord = "llm_record" + // 任务并发上限:开关 + 上限数。默认关闭;开启后默认上限 5(见 defaultConcurrencyLimit)。 + settingConcurrencyOn = "task_concurrency_enabled" + settingConcurrencyLimit = "task_concurrency_limit" // defaultWebSearchBackend is used when web search is on but no backend was picked. defaultWebSearchBackend = "ddgs" // defaultWorkers is the concurrent work-agent count when the setting is unset. defaultWorkers = 3 + // defaultConcurrencyLimit is the simultaneous-running-task cap when the feature + // is enabled but no explicit limit was saved. + defaultConcurrencyLimit = 5 ) +// 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) { + on, _, _ := m.pg.GetSetting(settingConcurrencyOn) + if strings.TrimSpace(on) != "true" { + return false, 0 + } + limit = defaultConcurrencyLimit + if v, ok, _ := m.pg.GetSetting(settingConcurrencyLimit); ok { + if n, err := strconv.Atoi(strings.TrimSpace(v)); err == nil && n >= 1 { + limit = n + } + } + return true, limit +} + +// SetConcurrency persists the running-task concurrency cap. limit<1 is clamped to 1. +func (m *Manager) SetConcurrency(enabled bool, limit int) error { + if limit < 1 { + limit = defaultConcurrencyLimit + } + if err := m.pg.SetSetting(settingConcurrencyLimit, strconv.Itoa(limit)); err != nil { + return err + } + return m.pg.SetSetting(settingConcurrencyOn, strconv.FormatBool(enabled)) +} + // Workers returns the configured concurrent work-agent count (default 3). Read // per-task at engine.Run, so a change applies to tasks started afterwards. func (m *Manager) Workers() int { @@ -457,7 +491,7 @@ func unixOrZero(t *time.Time) int64 { func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Interceptor) *Task { return &Task{ ID: strconv.FormatInt(pt.ID, 10), ExpID: pt.ExplorationID, - Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, + Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, Queued: pt.Queued, CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef, LLMProfileID: pt.LLMProfileID, TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds, @@ -526,6 +560,24 @@ func (m *Manager) SetTaskPaused(id string, paused bool) error { return m.pg.SetPaused(n, paused) } +// SetTaskQueued persists a task's queued (concurrency-hold) state and syncs the +// in-memory handle so listTasks/占位统计 see it immediately. +func (m *Manager) SetTaskQueued(id string, queued bool) error { + n, err := strconv.ParseInt(id, 10, 64) + if err != nil { + return err + } + if err := m.pg.SetQueued(n, queued); err != nil { + return err + } + m.mu.Lock() + if t := m.tasks[id]; t != nil { + t.Queued = queued + } + m.mu.Unlock() + return nil +} + // TaskStatus returns a task's current in-memory status (empty if unknown). func (m *Manager) TaskStatus(id string) string { m.mu.RLock() diff --git a/server/scheduler.go b/server/scheduler.go index 01fb6d1..97cd0b2 100644 --- a/server/scheduler.go +++ b/server/scheduler.go @@ -50,6 +50,7 @@ func (sc *Scheduler) Run(ctx context.Context) { case <-ctx.Done(): return case <-t.C: + sc.s.reconcileConcurrency() // 并发上限:有空位就把排队任务补位启动 sc.step() } } diff --git a/server/server.go b/server/server.go index 55c40f6..99f8710 100644 --- a/server/server.go +++ b/server/server.go @@ -42,6 +42,11 @@ type Server struct { skillDir string // root directory for skill subdirectories jwtKey []byte // HS256 signing key loaded from / generated into dataDir/jwt.key + // concMu serializes concurrency-cap decisions (admission + reconcile) so a + // scheduler tick and an HTTP settings change / task creation can't both count + // free slots off the same snapshot and over-promote past the limit. + concMu sync.Mutex + cfgMu sync.Mutex mainAgent *agent.MainAgent // nil when no LLM provider is configured chatAgent *agent.ChatAgent // conversational runner for the chat page; nil w/o LLM @@ -227,7 +232,8 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi } // resume the active task's engine; other tasks resume when opened (setActive). // (a restored-paused task's loops still start but idle until resumed.) - if t := m.ActiveTask(); t != nil { + // 排队中的任务不在此启动——留给 reconcileConcurrency 按空位补位。 + if t := m.ActiveTask(); t != nil && !t.Queued { s.engine.Run(ctx, t) } return s @@ -846,6 +852,8 @@ func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) { switch { case isTerminalStatus(t.Status): // 持久化终态优先（done/failed/timeout） status = t.Status + case t.Queued: // 因并发上限挂起、等待空位(尚未启动) + status = "queued" case t.Paused || s.engine.IsPaused(t.ID): status = "paused" case s.engine.Ready() && s.engine.Started(t.ID): @@ -880,7 +888,8 @@ func (s *Server) setActive(w http.ResponseWriter, r *http.Request) { return } // resume the engine for the opened task (idempotent — no-op if already running). - if t, ok := s.m.Task(req.ID); ok { + // 排队中的任务:仅设为活跃可查看,不启动引擎(维持并发上限,由 reconcile 补位)。 + if t, ok := s.m.Task(req.ID); ok && !t.Queued { s.engine.Run(s.ctx, t) } writeJSON(w, 200, map[string]any{"active": req.ID}) @@ -2145,16 +2154,22 @@ func (s *Server) getSettings(w http.ResponseWriter, r *http.Request) { func (s *Server) settingsPayload() map[string]any { on, backend, braveKey, tavilyKey, proxy := s.m.WebSearch() pyStored, _, _ := s.m.pg.GetSetting(settingPythonInterp) + concOn, concLimit := s.m.ConcurrencyLimit() + if concLimit == 0 { + concLimit = defaultConcurrencyLimit // 关闭时也回显一个合理默认值给 UI + } return map[string]any{ - "traffic_capture": s.m.TrafficEnabled(), - "llm_record": s.m.LLMRecordEnabled(), - "web_search_enabled": on, - "web_search_backend": backend, - "brave_key_set": strings.TrimSpace(braveKey) != "", - "tavily_key_set": strings.TrimSpace(tavilyKey) != "", - "web_search_proxy": proxy, // 独立出口代理(http/https/socks5)，空=直连 - "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) - "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 + "traffic_capture": s.m.TrafficEnabled(), + "llm_record": s.m.LLMRecordEnabled(), + "web_search_enabled": on, + "web_search_backend": backend, + "brave_key_set": strings.TrimSpace(braveKey) != "", + "tavily_key_set": strings.TrimSpace(tavilyKey) != "", + "web_search_proxy": proxy, // 独立出口代理(http/https/socks5)，空=直连 + "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) + "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 + "task_concurrency_enabled": concOn, // 任务并发上限开关(默认关) + "task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5) } } @@ -2188,6 +2203,9 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空 PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径 Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效 + // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 + ConcurrencyEnabled *bool `json:"task_concurrency_enabled"` + ConcurrencyLimit *int `json:"task_concurrency_limit"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeErr(w, 400, err.Error()) @@ -2199,6 +2217,26 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { return } } + if req.ConcurrencyEnabled != nil || req.ConcurrencyLimit != nil { + // 部分 PUT:未给的字段用当前值兜底,避免只改一个把另一个重置。 + curOn, curLimit := s.m.ConcurrencyLimit() + if curLimit == 0 { + curLimit = defaultConcurrencyLimit + } + on, limit := curOn, curLimit + if req.ConcurrencyEnabled != nil { + on = *req.ConcurrencyEnabled + } + if req.ConcurrencyLimit != nil { + limit = *req.ConcurrencyLimit + } + if err := s.m.SetConcurrency(on, limit); err != nil { + writeErr(w, 400, err.Error()) + return + } + // 立即协调一次:关闭时放行全部排队,调高上限时补位启动,不必等下一个 tick。 + go s.reconcileConcurrency() + } if req.PythonInterp != nil { if err := s.m.pg.SetSetting(settingPythonInterp, strings.TrimSpace(*req.PythonInterp)); err != nil { writeErr(w, 500, err.Error()) diff --git a/web/src/app/(main)/function/tasks/page.tsx b/web/src/app/(main)/function/tasks/page.tsx index a6a668e..fa7fac9 100644 --- a/web/src/app/(main)/function/tasks/page.tsx +++ b/web/src/app/(main)/function/tasks/page.tsx @@ -13,6 +13,7 @@ import { ChevronRightIcon, PaperclipIcon, Loader2Icon, + SlidersHorizontalIcon, } from "lucide-react"; import { StatusBadge } from "@/components/status-badge"; @@ -30,8 +31,19 @@ import { import { Button } from "@/components/ui/button"; import { Card, CardContent } from "@/components/ui/card"; import { Checkbox } from "@/components/ui/checkbox"; +import { + Dialog, + DialogClose, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, + DialogTrigger, +} from "@/components/ui/dialog"; import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; +import { Switch } from "@/components/ui/switch"; import { Textarea } from "@/components/ui/textarea"; import { Table, @@ -143,6 +155,7 @@ function fmtDateTime(unix?: number): string { const STATUS_OPTIONS: { value: TaskStatus; label: string }[] = [ { value: "created", label: "已创建" }, + { value: "queued", label: "排队中" }, { value: "running", label: "运行中" }, { value: "paused", label: "已暂停" }, { value: "done", label: "已完成" }, @@ -263,6 +276,7 @@ export default function TasksPage() { {filtered.length}/{tasks.length} 条 + @@ -423,6 +437,99 @@ const TaskRow = React.memo(function TaskRow({ ); }); +// ConcurrencySettingsDialog is the 齿轮 button left of 新建任务: toggle the +// simultaneous-running-task cap and pick the limit. Default off; default limit 5. +// Persisted globally via /api/settings — applies to tasks created afterwards. +function ConcurrencySettingsDialog() { + const [open, setOpen] = React.useState(false); + const [enabled, setEnabled] = React.useState(false); + const [limit, setLimit] = React.useState("5"); + const [loading, setLoading] = React.useState(false); + const [saving, setSaving] = React.useState(false); + + // 每次打开都拉取当前值,避免显示陈旧状态。 + React.useEffect(() => { + if (!open) return; + setLoading(true); + api + .settings() + .then((s) => { + setEnabled(!!s.task_concurrency_enabled); + setLimit(String(s.task_concurrency_limit ?? 5)); + }) + .catch(() => {}) + .finally(() => setLoading(false)); + }, [open]); + + async function save() { + const n = Math.max(1, Math.floor(Number(limit) || 5)); + setSaving(true); + try { + await api.setSettings({ task_concurrency_enabled: enabled, task_concurrency_limit: n }); + toast.success(enabled ? `已开启并发限制：最多同时运行 ${n} 个任务` : "已关闭任务并发限制"); + setOpen(false); + } catch (e) { + toast.error("保存失败：" + (e as Error).message); + } finally { + setSaving(false); + } + } + + return ( + + + + + + + 任务并发限制 + + 限制同时「运行中」的任务数量。关闭时不限制；开启后新建任务若已达上限，会进入「排队中」，有空位时自动开始。 + + + +
+
+
+ + 默认关闭（不限制并发） +
+ +
+ + {enabled && ( +
+ + setLimit(e.target.value)} + disabled={loading} + /> + 开启后默认 5，最小 1。 +
+ )} +
+ + + + + + + + + + ); +} + // CreateTaskSheet is the 新建任务 drawer. Its form state lives HERE, not in TasksPage: with // 描述/目标 held by the page component every keystroke re-rendered the whole task table // behind the drawer (plus its sticky column and 20 AlertDialog trees), which showed up as @@ -499,7 +606,7 @@ function CreateTaskSheet({ onCreated }: { onCreated: () => void }) { return ( - diff --git a/web/src/lib/status.ts b/web/src/lib/status.ts index 07d4d97..ab3245e 100644 --- a/web/src/lib/status.ts +++ b/web/src/lib/status.ts @@ -54,6 +54,7 @@ const intent: Record = { const task: Record = { created: { label: "已创建", tone: "slate" }, + queued: { label: "排队中", tone: "amber" }, running: { label: "运行中", tone: "blue" }, paused: { label: "已暂停", tone: "amber" }, done: { label: "已完成", tone: "green" }, diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 3444edd..50b578b 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -1,7 +1,7 @@ // ARTEX domain model — types used across the UI. // Derived from the functional spec (section 7: 关键数据形状). -export type TaskStatus = "created" | "running" | "paused" | "done" | "failed" | "timeout"; +export type TaskStatus = "created" | "queued" | "running" | "paused" | "done" | "failed" | "timeout"; export type EngineMode = "exploring" | "paused" | "stalled" | "idle"; export interface Task { @@ -498,6 +498,9 @@ export interface Settings { web_search_proxy?: string; python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测) workers?: number; // 并发工作 agent 数(默认3)；对之后启动的任务生效 + // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 + task_concurrency_enabled?: boolean; // 默认 false + task_concurrency_limit?: number; // 开启后默认 5 } // ---- LLM config ----