diff --git a/db/schema.sql b/db/schema.sql
index 1fe2311..4a35aab 100644
--- a/db/schema.sql
+++ b/db/schema.sql
@@ -524,6 +524,7 @@ CREATE TABLE IF NOT EXISTS tasks (
timeout_seconds INTEGER NOT NULL DEFAULT 0,
plan_heartbeat_seconds INTEGER NOT NULL DEFAULT 300,
coverage_enabled BOOLEAN NOT NULL DEFAULT true,
+ worker_count INTEGER NOT NULL DEFAULT 0, -- 0=用全局默认 workers
pinned_at TIMESTAMPTZ,
first_run_at TIMESTAMPTZ,
deadline_at TIMESTAMPTZ,
@@ -546,6 +547,8 @@ ALTER TABLE tasks ADD COLUMN IF NOT EXISTS queued BOOLEAN NOT NULL DEFAULT false
-- 给 agent 开放 add_task_scope/list_untested_assets;false=全部关闭(见 task_scope.go)。
-- 存量任务默认 true 保持原行为;company 关联(task_scope kind=company)不受此开关影响。
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS coverage_enabled BOOLEAN NOT NULL DEFAULT true;
+-- 任务级 Worker 运行数量;0=用全局默认(settings.workers)。补旧库。
+ALTER TABLE tasks ADD COLUMN IF NOT EXISTS worker_count INTEGER NOT NULL DEFAULT 0;
-- queued_at makes admission FIFO reflect the actual enqueue order rather than the
-- task creation order. queue_mode distinguishes first bootstrap from resuming an
-- exploration that already owns goals/history.
@@ -617,16 +620,21 @@ CREATE TABLE IF NOT EXISTS task_templates (
nkey TEXT NOT NULL UNIQUE,
description TEXT NOT NULL,
goal TEXT NOT NULL,
+ remark TEXT NOT NULL DEFAULT '',
-- 预设的任务分类；分类删除时置空（与 tasks.category_id 一致，不阻断）。
category_id BIGINT REFERENCES task_categories(id) ON DELETE SET NULL,
-- 预设的任务级拦截/允许规则快照(AssetInterceptRuleInput 数组)；应用模板时灌进新任务。
intercept_rules JSONB NOT NULL DEFAULT '[]',
+ -- 预设的任务级 Worker 运行数量;0=用全局默认。
+ worker_count INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- 补旧库(已发版,加列带 IF NOT EXISTS)。
ALTER TABLE task_templates ADD COLUMN IF NOT EXISTS category_id BIGINT REFERENCES task_categories(id) ON DELETE SET NULL;
ALTER TABLE task_templates ADD COLUMN IF NOT EXISTS intercept_rules JSONB NOT NULL DEFAULT '[]';
+ALTER TABLE task_templates ADD COLUMN IF NOT EXISTS remark TEXT NOT NULL DEFAULT '';
+ALTER TABLE task_templates ADD COLUMN IF NOT EXISTS worker_count INTEGER NOT NULL DEFAULT 0;
CREATE INDEX IF NOT EXISTS idx_task_templates_updated ON task_templates(updated_at DESC, id DESC);
DROP TRIGGER IF EXISTS trg_task_templates_upd ON task_templates;
CREATE TRIGGER trg_task_templates_upd BEFORE UPDATE ON task_templates
diff --git a/db/task_archives_restore.go b/db/task_archives_restore.go
index 01094ca..56f102b 100644
--- a/db/task_archives_restore.go
+++ b/db/task_archives_restore.go
@@ -122,7 +122,7 @@ AND NOT EXISTS (
if _, err := tx.Exec(`UPDATE tasks SET
name='',remark='',category_id=NULL,description='',goal='',paused=true,queued=false,queued_at=NULL,queue_mode='',
llm_profile_id=NULL,active_llm_profile_id=NULL,llm_chain_revision=llm_chain_revision+1,
- company_id=NULL,parent_ref=NULL,timeout_seconds=0,coverage_enabled=true,pinned_at=NULL,
+ company_id=NULL,parent_ref=NULL,timeout_seconds=0,coverage_enabled=true,worker_count=0,pinned_at=NULL,
first_run_at=NULL,deadline_at=NULL,archived_at=now(),deleted_at=now()
WHERE id=$1`, taskID); err != nil {
return err
@@ -322,7 +322,7 @@ func restoreTaskStub(tx *sql.Tx, row map[string]any, taskID, remaining int64) er
llm_profile_id=archived.llm_profile_id,active_llm_profile_id=archived.active_llm_profile_id,
llm_chain_revision=archived.llm_chain_revision,company_id=archived.company_id,parent_ref=archived.parent_ref,
timeout_seconds=archived.timeout_seconds,plan_heartbeat_seconds=archived.plan_heartbeat_seconds,
- coverage_enabled=archived.coverage_enabled,pinned_at=archived.pinned_at,first_run_at=archived.first_run_at,
+ coverage_enabled=archived.coverage_enabled,worker_count=COALESCE(archived.worker_count,0),pinned_at=archived.pinned_at,first_run_at=archived.first_run_at,
deadline_at=CASE WHEN archived.paused AND $3>0 THEN now()+make_interval(secs=>$3::double precision)
ELSE archived.deadline_at END,
deleted_at=NULL,archived_at=NULL,completed_at=archived.completed_at,
diff --git a/db/task_archives_test.go b/db/task_archives_test.go
index 93a1364..bc4274a 100644
--- a/db/task_archives_test.go
+++ b/db/task_archives_test.go
@@ -132,7 +132,9 @@ func TestTaskArchiveDatabaseRoundTrip(t *testing.T) {
if err := d.EnsureLLMUsageTable(); err != nil {
t.Fatal(err)
}
- task, err := d.CreateTaskWithOptions("archive database roundtrip", "restore exact graph", TaskCreateOptions{Name: "cold task"})
+ task, err := d.CreateTaskWithOptions("archive database roundtrip", "restore exact graph", TaskCreateOptions{
+ Name: "cold task", WorkerCount: 7,
+ })
if err != nil {
t.Fatal(err)
}
@@ -297,7 +299,7 @@ VALUES($1,$2,0,'quota_exhausted','balance exhausted',$3,$4,$3)`, task.ID, llmPro
if err != nil || live == nil {
t.Fatalf("restored task = %+v, %v", live, err)
}
- if live.Name != "cold task" || live.Remark != remark || !live.Paused {
+ if live.Name != "cold task" || live.Remark != remark || live.WorkerCount != 7 || !live.Paused {
t.Fatalf("restored task metadata mismatch: %+v", live)
}
if err := d.QueryRow(`SELECT count(*) FROM exploration_nodes WHERE exploration_id=$1`, task.ExplorationID).Scan(&nodes); err != nil {
diff --git a/db/task_options_test.go b/db/task_options_test.go
new file mode 100644
index 0000000..100bda7
--- /dev/null
+++ b/db/task_options_test.go
@@ -0,0 +1,102 @@
+package db
+
+import (
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestTaskAndTemplateRemarkLimitIsShared(t *testing.T) {
+ boundary := strings.Repeat("界", MaxTaskRemarkRunes)
+ if _, _, err := normalizeTaskTemplateInput(TaskTemplateInput{Name: "n", Description: "d", Goal: "g", Remark: boundary}); err != nil {
+ t.Fatalf("boundary remark rejected: %v", err)
+ }
+ if _, _, err := normalizeTaskTemplateInput(TaskTemplateInput{Name: "n", Description: "d", Goal: "g", Remark: boundary + "界"}); err == nil {
+ t.Fatal("oversized template remark accepted")
+ }
+}
+
+func TestWorkerCountNormalizationAndLimit(t *testing.T) {
+ if got := normalizeWorkerCount(-1); got != 0 {
+ t.Fatalf("negative worker count = %d, want 0", got)
+ }
+ if _, _, err := normalizeTaskTemplateInput(TaskTemplateInput{Name: "n", Description: "d", Goal: "g", WorkerCount: MaxTaskWorkerCount}); err != nil {
+ t.Fatalf("maximum worker count rejected: %v", err)
+ }
+ if _, _, err := normalizeTaskTemplateInput(TaskTemplateInput{Name: "n", Description: "d", Goal: "g", WorkerCount: MaxTaskWorkerCount + 1}); err == nil {
+ t.Fatal("oversized worker count accepted")
+ }
+}
+
+func TestTaskOptionsPersistAndValidateLimits(t *testing.T) {
+ d, err := Open(testDSN(t))
+ if err != nil {
+ t.Skipf("postgres unavailable (%v) — skipping", err)
+ }
+ defer d.Close()
+
+ if _, err := d.CreateTaskWithOptions("oversized remark", "goal", TaskCreateOptions{
+ Remark: strings.Repeat("界", MaxTaskRemarkRunes+1),
+ }); err == nil {
+ t.Fatal("2001-rune task remark accepted")
+ }
+ if _, err := d.CreateTaskWithOptions("oversized workers", "goal", TaskCreateOptions{
+ WorkerCount: MaxTaskWorkerCount + 1,
+ }); err == nil {
+ t.Fatal("51-worker task accepted")
+ }
+
+ remark := strings.Repeat("界", MaxTaskRemarkRunes)
+ task, err := d.CreateTaskWithOptions("task options boundary", "goal", TaskCreateOptions{
+ Remark: remark, WorkerCount: MaxTaskWorkerCount,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+ gotTask, err := d.GetTask(task.ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if gotTask == nil || gotTask.Remark != remark || gotTask.WorkerCount != MaxTaskWorkerCount {
+ t.Fatalf("task options round trip = %+v", gotTask)
+ }
+
+ name := fmt.Sprintf("options template %d", time.Now().UnixNano())
+ templateRemark := "模板备注 📝"
+ template, err := d.CreateTaskTemplate(TaskTemplateInput{
+ Name: name, Description: "template description", Goal: "template goal",
+ Remark: templateRemark, WorkerCount: MaxTaskWorkerCount,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _, _ = d.DeleteTaskTemplate(template.ID) })
+ gotTemplate, err := d.GetTaskTemplate(template.ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if gotTemplate == nil || gotTemplate.Remark != templateRemark || gotTemplate.WorkerCount != MaxTaskWorkerCount {
+ t.Fatalf("created template options round trip = %+v", gotTemplate)
+ }
+
+ updatedRemark := strings.Repeat("改", MaxTaskRemarkRunes)
+ updatedWorkers := MaxTaskWorkerCount - 1
+ updated, err := d.PatchTaskTemplate(template.ID, TaskTemplatePatch{
+ Remark: &updatedRemark, WorkerCount: &updatedWorkers,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if updated.Remark != updatedRemark || updated.WorkerCount != updatedWorkers {
+ t.Fatalf("updated template options = %+v", updated)
+ }
+ gotTemplate, err = d.GetTaskTemplate(template.ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if gotTemplate == nil || gotTemplate.Remark != updatedRemark || gotTemplate.WorkerCount != updatedWorkers {
+ t.Fatalf("updated template options round trip = %+v", gotTemplate)
+ }
+}
diff --git a/db/task_templates.go b/db/task_templates.go
index 80e3116..5401139 100644
--- a/db/task_templates.go
+++ b/db/task_templates.go
@@ -23,16 +23,18 @@ var (
ErrTaskTemplateNotFound = errors.New("task template not found")
)
-// TaskTemplate is a reusable task preset (description/goal + optional category
-// and task-level intercept/allow rules).
+// TaskTemplate is a reusable task preset (description/goal + optional remark,
+// worker count, category, and task-level intercept/allow rules).
type TaskTemplate struct {
ID int64 `json:"id"`
Name string `json:"name"`
NKey string `json:"-"`
Description string `json:"description"`
Goal string `json:"goal"`
+ Remark string `json:"remark"`
CategoryID *int64 `json:"category_id"`
InterceptRules []TaskInterceptRuleInput `json:"intercept_rules"`
+ WorkerCount int `json:"worker_count"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
@@ -42,8 +44,10 @@ type TaskTemplateInput struct {
Name string
Description string
Goal string
+ Remark string
CategoryID *int64
InterceptRules []TaskInterceptRuleInput
+ WorkerCount int
}
// TaskTemplatePatch changes only fields flagged as set. Name/Description/Goal use
@@ -53,18 +57,20 @@ type TaskTemplatePatch struct {
Name *string
Description *string
Goal *string
+ Remark *string
CategoryID *int64
SetCategoryID bool
InterceptRules []TaskInterceptRuleInput
SetInterceptRules bool
+ WorkerCount *int
}
-const taskTemplateCols = `id, name, nkey, description, goal, category_id, intercept_rules, created_at, updated_at`
+const taskTemplateCols = `id, name, nkey, description, goal, COALESCE(remark,''), category_id, intercept_rules, COALESCE(worker_count,0), created_at, updated_at`
func scanTaskTemplate(row interface{ Scan(...any) error }) (TaskTemplate, error) {
var t TaskTemplate
var rulesRaw []byte
- if err := row.Scan(&t.ID, &t.Name, &t.NKey, &t.Description, &t.Goal, &t.CategoryID, &rulesRaw, &t.CreatedAt, &t.UpdatedAt); err != nil {
+ if err := row.Scan(&t.ID, &t.Name, &t.NKey, &t.Description, &t.Goal, &t.Remark, &t.CategoryID, &rulesRaw, &t.WorkerCount, &t.CreatedAt, &t.UpdatedAt); err != nil {
return t, err
}
t.InterceptRules = []TaskInterceptRuleInput{}
@@ -112,11 +118,21 @@ func normalizeTaskTemplateInput(in TaskTemplateInput) (TaskTemplateInput, string
case utf8.RuneCountInString(in.Goal) > MaxTaskTemplateTextRunes:
return in, "", fmt.Errorf("%w: goal exceeds %d characters", ErrTaskTemplateInvalid, MaxTaskTemplateTextRunes)
}
+ in.Remark = strings.TrimSpace(in.Remark)
+ // Keep template remarks within the same shared limit as task remarks so
+ // applying a template cannot produce a rejected create-task payload.
+ if utf8.RuneCountInString(in.Remark) > MaxTaskRemarkRunes {
+ return in, "", fmt.Errorf("%w: remark exceeds %d characters", ErrTaskTemplateInvalid, MaxTaskRemarkRunes)
+ }
+ in.WorkerCount = normalizeWorkerCount(in.WorkerCount)
+ if err := validateWorkerCount(in.WorkerCount); err != nil {
+ return in, "", fmt.Errorf("%w: %v", ErrTaskTemplateInvalid, err)
+ }
return in, taskTemplateNKey(in.Name), nil
}
func normalizeTaskTemplatePatch(patch TaskTemplatePatch) (TaskTemplatePatch, *string, error) {
- if patch.Name == nil && patch.Description == nil && patch.Goal == nil && !patch.SetCategoryID && !patch.SetInterceptRules {
+ if patch.Name == nil && patch.Description == nil && patch.Goal == nil && patch.Remark == nil && !patch.SetCategoryID && !patch.SetInterceptRules && patch.WorkerCount == nil {
return patch, nil, fmt.Errorf("%w: no fields supplied", ErrTaskTemplateInvalid)
}
var nkey *string
@@ -152,6 +168,20 @@ func normalizeTaskTemplatePatch(patch TaskTemplatePatch) (TaskTemplatePatch, *st
}
patch.Goal = &goal
}
+ if patch.Remark != nil {
+ remark := strings.TrimSpace(*patch.Remark)
+ if utf8.RuneCountInString(remark) > MaxTaskRemarkRunes {
+ return patch, nil, fmt.Errorf("%w: remark exceeds %d characters", ErrTaskTemplateInvalid, MaxTaskRemarkRunes)
+ }
+ patch.Remark = &remark
+ }
+ if patch.WorkerCount != nil {
+ n := normalizeWorkerCount(*patch.WorkerCount)
+ if err := validateWorkerCount(n); err != nil {
+ return patch, nil, fmt.Errorf("%w: %v", ErrTaskTemplateInvalid, err)
+ }
+ patch.WorkerCount = &n
+ }
return patch, nkey, nil
}
@@ -171,10 +201,10 @@ func (d *DB) CreateTaskTemplate(in TaskTemplateInput) (*TaskTemplate, error) {
return nil, err
}
t, err := scanTaskTemplate(d.QueryRow(`
-INSERT INTO task_templates(name, nkey, description, goal, category_id, intercept_rules)
-VALUES ($1,$2,$3,$4,$5,$6)
+INSERT INTO task_templates(name, nkey, description, goal, remark, category_id, intercept_rules, worker_count)
+VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
ON CONFLICT (nkey) DO NOTHING
-RETURNING `+taskTemplateCols, in.Name, nkey, in.Description, in.Goal, in.CategoryID, rulesJSON))
+RETURNING `+taskTemplateCols, in.Name, nkey, in.Description, in.Goal, in.Remark, in.CategoryID, rulesJSON, in.WorkerCount))
if err == sql.ErrNoRows {
return nil, ErrTaskTemplateNameConflict
}
@@ -221,7 +251,8 @@ func (d *DB) UpdateTaskTemplate(id int64, in TaskTemplateInput) (*TaskTemplate,
return nil, err
}
return d.PatchTaskTemplate(id, TaskTemplatePatch{
- Name: &in.Name, Description: &in.Description, Goal: &in.Goal,
+ Name: &in.Name, Description: &in.Description, Goal: &in.Goal, Remark: &in.Remark,
+ WorkerCount: &in.WorkerCount,
})
}
@@ -242,16 +273,20 @@ SET name=CASE WHEN $2 THEN $3::text ELSE name END,
nkey=CASE WHEN $2 THEN $4::text ELSE nkey END,
description=CASE WHEN $5 THEN $6::text ELSE description END,
goal=CASE WHEN $7 THEN $8::text ELSE goal END,
- category_id=CASE WHEN $9 THEN $10::bigint ELSE category_id END,
- intercept_rules=CASE WHEN $11 THEN $12::jsonb ELSE intercept_rules END
+ remark=CASE WHEN $9 THEN $10::text ELSE remark END,
+ category_id=CASE WHEN $11 THEN $12::bigint ELSE category_id END,
+ intercept_rules=CASE WHEN $13 THEN $14::jsonb ELSE intercept_rules END,
+ worker_count=CASE WHEN $15 THEN $16::integer ELSE worker_count END
WHERE id=$1
RETURNING `+taskTemplateCols,
id,
patch.Name != nil, patch.Name, nkey,
patch.Description != nil, patch.Description,
patch.Goal != nil, patch.Goal,
+ patch.Remark != nil, patch.Remark,
patch.SetCategoryID, patch.CategoryID,
patch.SetInterceptRules, rulesJSON,
+ patch.WorkerCount != nil, patch.WorkerCount,
))
if err == sql.ErrNoRows {
return nil, ErrTaskTemplateNotFound
diff --git a/db/tasks.go b/db/tasks.go
index b675bea..c17f1b6 100644
--- a/db/tasks.go
+++ b/db/tasks.go
@@ -6,7 +6,9 @@ import (
"errors"
"fmt"
"strconv"
+ "strings"
"time"
+ "unicode/utf8"
)
// Task is a row in the task registry (1:1 with an exploration).
@@ -51,6 +53,9 @@ type Task struct {
// list_untested_assets、态势里不注入 coverage 块(scope 字段仍保留)。company 关联
// (task_scope kind=company)与此开关无关，永不受影响。见 db/task_scope.go。
CoverageEnabled bool `json:"coverage_enabled"`
+ // WorkerCount is the per-task concurrent work-agent count. 0 means use the
+ // global settings.workers default when the engine starts this task.
+ WorkerCount int `json:"worker_count"`
}
// TaskDeleteResult reports optional related-data cleanup performed in the same
@@ -132,6 +137,29 @@ func normalizeHeartbeat(sec int) int {
return sec
}
+// MaxTaskWorkerCount is the hard upper bound for a per-task worker override.
+const MaxTaskWorkerCount = 50
+
+// MaxTaskRemarkRunes is the shared limit for task and template remarks.
+const MaxTaskRemarkRunes = 2000
+
+// normalizeWorkerCount converts negative per-task worker overrides to 0.
+// The upper bound is enforced separately by validateWorkerCount; 0 means
+// "use the global settings.workers default".
+func normalizeWorkerCount(n int) int {
+ if n < 0 {
+ return 0
+ }
+ return n
+}
+
+func validateWorkerCount(n int) error {
+ if n > MaxTaskWorkerCount {
+ return fmt.Errorf("Worker 运行数量最多 %d", MaxTaskWorkerCount)
+ }
+ return nil
+}
+
func (d *DB) CreateTask(description, goal string, llmProfileID *int64, timeoutSeconds, planHeartbeatSeconds int) (*Task, error) {
var ids []int64
if llmProfileID != nil {
@@ -156,6 +184,10 @@ type TaskCreateOptions struct {
// CoverageEnabled 是「资产覆盖度功能」开关;nil=默认开(true)，让不关心该开关的创建
// 路径(编排 spawn、老 API)沿用原行为。仅 web 创建任务时可显式传 false 关闭。
CoverageEnabled *bool
+ // WorkerCount 是本任务并发运行的工作 Agent 数;0/省略=用全局默认。上限 50。
+ WorkerCount int
+ // Remark 是创建时可带的任务备注(最多 2000 字);空=无备注。
+ Remark string
// InterceptRules 是任务级资产拦截规则,创建时随任务在同一事务内写入 task_intercept_rules。
InterceptRules []TaskInterceptRuleInput
}
@@ -200,6 +232,14 @@ VALUES ($1, 'fact', $2, 0, 'origin', 'system')`, expID, string(originPayload));
}
opts.PlanHeartbeatSeconds = normalizeHeartbeat(opts.PlanHeartbeatSeconds)
coverageEnabled := opts.CoverageEnabled == nil || *opts.CoverageEnabled
+ opts.WorkerCount = normalizeWorkerCount(opts.WorkerCount)
+ if err := validateWorkerCount(opts.WorkerCount); err != nil {
+ return nil, err
+ }
+ opts.Remark = strings.TrimSpace(opts.Remark)
+ if utf8.RuneCountInString(opts.Remark) > MaxTaskRemarkRunes {
+ return nil, fmt.Errorf("备注最多 %d 个字符", MaxTaskRemarkRunes)
+ }
var categoryName string
if opts.CategoryID != nil {
if *opts.CategoryID <= 0 {
@@ -225,12 +265,12 @@ VALUES ($1, 'fact', $2, 0, 'origin', 'system')`, expID, string(originPayload));
SourceTaskIDs: append([]int64(nil), opts.SourceTaskIDs...),
CompanyIDs: append([]int64(nil), opts.CompanyIDs...),
TimeoutSeconds: opts.TimeoutSeconds, PlanHeartbeatSeconds: opts.PlanHeartbeatSeconds,
- CoverageEnabled: coverageEnabled,
+ CoverageEnabled: coverageEnabled, WorkerCount: opts.WorkerCount, Remark: opts.Remark,
}
if err := tx.QueryRow(`
-INSERT INTO tasks(name, category_id, description, goal, exploration_id, llm_profile_id, active_llm_profile_id, timeout_seconds, plan_heartbeat_seconds, coverage_enabled)
-VALUES ($1,$2,$3,$4,$5,$6,$6,$7,$8,$9)
-RETURNING id, status, paused, created_at`, opts.Name, opts.CategoryID, description, goal, expID, active, opts.TimeoutSeconds, opts.PlanHeartbeatSeconds, coverageEnabled).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil {
+INSERT INTO tasks(name, remark, category_id, description, goal, exploration_id, llm_profile_id, active_llm_profile_id, timeout_seconds, plan_heartbeat_seconds, coverage_enabled, worker_count)
+VALUES ($1,$2,$3,$4,$5,$6,$7,$7,$8,$9,$10,$11)
+RETURNING id, status, paused, created_at`, opts.Name, opts.Remark, opts.CategoryID, description, goal, expID, active, opts.TimeoutSeconds, opts.PlanHeartbeatSeconds, coverageEnabled, opts.WorkerCount).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil {
return nil, err
}
if err := insertTaskRelations(tx, t.ID, opts.SourceTaskIDs); err != nil {
@@ -363,11 +403,11 @@ func insertTaskLLMProfiles(tx *sql.Tx, taskID int64, profileIDs []int64) error {
const taskCols = `id, COALESCE(name,''), remark, category_id,
COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''),
-description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), first_run_at, deadline_at`
+description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), COALESCE(worker_count,0), first_run_at, deadline_at`
func scanTask(sc interface{ Scan(...any) error }) (*Task, error) {
var t Task
- if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.CoverageEnabled, &t.FirstRunAt, &t.DeadlineAt); err != nil {
+ if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.CoverageEnabled, &t.WorkerCount, &t.FirstRunAt, &t.DeadlineAt); err != nil {
return nil, err
}
t.Pinned = t.PinnedAt != nil
diff --git a/server/assembly_test.go b/server/assembly_test.go
index 1eabeba..1ea94bf 100644
--- a/server/assembly_test.go
+++ b/server/assembly_test.go
@@ -6,9 +6,9 @@ import (
"path/filepath"
"testing"
- actool "github.com/Autumn-27/norma/tool"
"github.com/Autumn-27/artex/agent"
"github.com/Autumn-27/artex/db"
+ actool "github.com/Autumn-27/norma/tool"
)
// hasSkillTool reports whether the packed tool set contains the Skill meta-tool.
diff --git a/server/dto.go b/server/dto.go
index 60785a5..339c1f2 100644
--- a/server/dto.go
+++ b/server/dto.go
@@ -64,6 +64,7 @@ type TaskDTO struct {
ArchiveBlockedBy string `json:"archive_blocked_by_task_id,omitempty"`
CompanyIDs []int64 `json:"company_ids"`
CoverageEnabled bool `json:"coverage_enabled"` // 资产覆盖度功能开关(创建时定)
+ WorkerCount int `json:"worker_count"` // 本任务 Worker 数;0=用全局默认
}
// FindingSeverityDTO 是任务列表里按严重度分档的漏洞计数（严重/高/中/低）。
@@ -138,6 +139,7 @@ func taskDTO(t *Task, status string) TaskDTO {
SourceTaskIDs: sourceIDs,
CompanyIDs: lifecycle.CompanyIDs,
CoverageEnabled: t.CoverageEnabled,
+ WorkerCount: t.WorkerCount,
}
}
diff --git a/server/engine.go b/server/engine.go
index b5643b2..5d07a03 100644
--- a/server/engine.go
+++ b/server/engine.go
@@ -746,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) {
- workers := e.m.Workers()
+ workers := workerCountForTask(t, e.m.Workers())
e.deleteMu.RLock()
if e.IsDeleting(t.ID) {
e.deleteMu.RUnlock()
@@ -776,6 +776,18 @@ func (e *Engine) Run(ctx context.Context, t *Task) {
}
}
+func workerCountForTask(t *Task, global int) int {
+ if t != nil && t.WorkerCount > 0 {
+ // Direct database edits can bypass validation; cap those values so the
+ // engine never starts more than the supported per-task worker count.
+ if t.WorkerCount > db.MaxTaskWorkerCount {
+ return db.MaxTaskWorkerCount
+ }
+ return t.WorkerCount
+ }
+ return global
+}
+
// plannerHeartbeatInterval 解析任务的 planner 心跳间隔。db.CreateTask 已归一
// (低于 600 一律抬到 600);这里再兜一次底,防内存态异常值。
func plannerHeartbeatInterval(t *Task) time.Duration {
diff --git a/server/engine_worker_test.go b/server/engine_worker_test.go
new file mode 100644
index 0000000..132ee86
--- /dev/null
+++ b/server/engine_worker_test.go
@@ -0,0 +1,22 @@
+package server
+
+import (
+ "testing"
+
+ "github.com/Autumn-27/artex/db"
+)
+
+func TestWorkerCountForTaskUsesOverrideOrGlobalDefault(t *testing.T) {
+ if got := workerCountForTask(&Task{WorkerCount: 5}, 3); got != 5 {
+ t.Fatalf("override = %d, want 5", got)
+ }
+ if got := workerCountForTask(&Task{WorkerCount: 0}, 3); got != 3 {
+ t.Fatalf("zero override = %d, want global 3", got)
+ }
+ if got := workerCountForTask(&Task{WorkerCount: -1}, 3); got != 3 {
+ t.Fatalf("negative override = %d, want global 3", got)
+ }
+ if got := workerCountForTask(&Task{WorkerCount: db.MaxTaskWorkerCount + 1}, 3); got != db.MaxTaskWorkerCount {
+ t.Fatalf("oversized override = %d, want capped %d", got, db.MaxTaskWorkerCount)
+ }
+}
diff --git a/server/manager.go b/server/manager.go
index 09fdcd3..afea5ad 100644
--- a/server/manager.go
+++ b/server/manager.go
@@ -59,6 +59,7 @@ type Task struct {
TimeoutSeconds int `json:"timeout_seconds"`
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner 心跳触发间隔(秒)
CoverageEnabled bool `json:"coverage_enabled"` // 资产覆盖度功能开关(创建时定,默认开)
+ WorkerCount int `json:"worker_count"` // 本任务 Worker 数;0=用全局默认
FirstRunAt int64 `json:"first_run_at,omitempty"`
DeadlineAt int64 `json:"deadline_at,omitempty"`
Store *pgdb.ExplorationStore `json:"-"`
@@ -849,8 +850,8 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
SourceTaskIDs: append([]int64(nil), pt.SourceTaskIDs...),
CompanyIDs: append([]int64(nil), pt.CompanyIDs...),
TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds,
- CoverageEnabled: pt.CoverageEnabled,
- FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt),
+ CoverageEnabled: pt.CoverageEnabled, WorkerCount: pt.WorkerCount,
+ FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt),
Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1),
}
}
diff --git a/server/remarks.go b/server/remarks.go
index c33ecc6..0864eb9 100644
--- a/server/remarks.go
+++ b/server/remarks.go
@@ -10,7 +10,7 @@ import (
"github.com/Autumn-27/artex/db"
)
-const maxRemarkRunes = 2000
+const maxRemarkRunes = db.MaxTaskRemarkRunes
func normalizeRemark(value string) (string, error) {
value = strings.TrimSpace(value)
diff --git a/server/server.go b/server/server.go
index da6d755..d55ca77 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1541,6 +1541,8 @@ type createTaskReq struct {
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner 心跳触发间隔(秒);0/省略=默认600(10min);下限=默认=600,低于自动抬到600
SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // 创建时直接下发一条种子意图(内容=描述+目标),让 worker 免等首轮 planner 直接开跑;省略/null=默认关闭,走标准先规划再执行。显式传 true 才开(CTF 常一 work 解决时可省掉开跑前的 planner 轮)。
CoverageEnabled *bool `json:"coverage_enabled,omitempty"` // 资产覆盖度功能;省略/null=默认开(true)。false=关闭覆盖度计算/展示/自动累积范围+隐藏 add_task_scope/list_untested_assets。company 关联不受影响。
+ Remark string `json:"remark,omitempty"` // 任务备注(最多2000字);空=无
+ WorkerCount int `json:"worker_count,omitempty"` // 本任务 Worker 数;0/省略=用全局默认;上限50
// InterceptRules 任务级资产拦截/允许规则(创建时录入,存 task_intercept_rules,不进全局表)。
InterceptRules []taskInterceptRuleReq `json:"intercept_rules,omitempty"`
}
@@ -1604,12 +1606,27 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) {
}
schedule = &input
}
+ remark, err := normalizeRemark(req.Remark)
+ if err != nil {
+ writeErr(w, 400, err.Error())
+ return
+ }
+ if req.WorkerCount < 0 {
+ writeErr(w, 400, "Worker 运行数量不能为负数")
+ return
+ }
+ if req.WorkerCount > db.MaxTaskWorkerCount {
+ writeErr(w, 400, fmt.Sprintf("Worker 运行数量最多 %d", db.MaxTaskWorkerCount))
+ return
+ }
t, err := s.m.CreateTaskWithOptions(req.Description, req.Goal, db.TaskCreateOptions{
Schedule: schedule,
Name: strings.TrimSpace(req.Name), CategoryID: req.CategoryID,
SourceTaskIDs: sourceIDs, CompanyIDs: req.CompanyIDs, LLMProfileIDs: req.LLMProfileIDs,
TimeoutSeconds: req.TimeoutSeconds, PlanHeartbeatSeconds: req.PlanHeartbeatSeconds,
CoverageEnabled: req.CoverageEnabled,
+ Remark: remark,
+ WorkerCount: req.WorkerCount,
InterceptRules: interceptRules,
})
if err != nil {
diff --git a/server/task_templates.go b/server/task_templates.go
index 7125177..f94de47 100644
--- a/server/task_templates.go
+++ b/server/task_templates.go
@@ -22,8 +22,10 @@ type taskTemplateRequest struct {
Name *string `json:"name"`
Description *string `json:"description"`
Goal *string `json:"goal"`
+ Remark *string `json:"remark"`
CategoryID *int64 `json:"category_id"`
InterceptRules []taskInterceptRuleReq `json:"intercept_rules"`
+ WorkerCount *int `json:"worker_count"`
}
// decodeTaskTemplateRequest decodes the body into req and returns the set of
@@ -65,6 +67,7 @@ func validateTaskTemplateRequest(req taskTemplateRequest) error {
{name: "name", value: req.Name, limit: db.MaxTaskTemplateNameRunes},
{name: "description", value: req.Description, limit: db.MaxTaskTemplateTextRunes},
{name: "goal", value: req.Goal, limit: db.MaxTaskTemplateTextRunes},
+ {name: "remark", value: req.Remark, limit: db.MaxTaskRemarkRunes}, // 与任务备注 normalizeRemark 上限一致
}
for _, check := range checks {
if check.value == nil {
@@ -75,9 +78,20 @@ func validateTaskTemplateRequest(req taskTemplateRequest) error {
value = strings.Join(strings.Fields(value), " ")
}
if utf8.RuneCountInString(value) > check.limit {
+ if check.name == "remark" {
+ return fmt.Errorf("备注最多 %d 个字符", check.limit)
+ }
return fmt.Errorf("%s 最多 %d 个字符", check.name, check.limit)
}
}
+ if req.WorkerCount != nil {
+ if *req.WorkerCount < 0 {
+ return fmt.Errorf("Worker 运行数量不能为负数")
+ }
+ if *req.WorkerCount > db.MaxTaskWorkerCount {
+ return fmt.Errorf("Worker 运行数量最多 %d", db.MaxTaskWorkerCount)
+ }
+ }
return nil
}
@@ -125,12 +139,18 @@ func (s *Server) pgCreateTaskTemplate(w http.ResponseWriter, r *http.Request) {
writeErr(w, http.StatusBadRequest, "拦截/允许规则无效："+err.Error())
return
}
+ workerCount := 0
+ if req.WorkerCount != nil {
+ workerCount = *req.WorkerCount
+ }
template, err := pg.CreateTaskTemplate(db.TaskTemplateInput{
Name: stringValue(req.Name),
Description: stringValue(req.Description),
Goal: stringValue(req.Goal),
+ Remark: stringValue(req.Remark),
CategoryID: req.CategoryID,
InterceptRules: rules,
+ WorkerCount: workerCount,
})
if err != nil {
writeTaskTemplateErr(w, err)
@@ -160,11 +180,21 @@ func (s *Server) pgUpdateTaskTemplate(w http.ResponseWriter, r *http.Request) {
}
_, catPresent := present["category_id"]
_, rulesPresent := present["intercept_rules"]
- if req.Name == nil && req.Description == nil && req.Goal == nil && !catPresent && !rulesPresent {
- writeErr(w, http.StatusBadRequest, "至少需要提供 name、description、goal、category_id 或 intercept_rules")
+ _, remarkPresent := present["remark"]
+ _, workerPresent := present["worker_count"]
+ if req.Name == nil && req.Description == nil && req.Goal == nil && !catPresent && !rulesPresent && !remarkPresent && !workerPresent {
+ writeErr(w, http.StatusBadRequest, "至少需要提供 name、description、goal、remark、category_id、intercept_rules 或 worker_count")
return
}
patch := db.TaskTemplatePatch{Name: req.Name, Description: req.Description, Goal: req.Goal}
+ if remarkPresent {
+ if req.Remark == nil {
+ empty := ""
+ patch.Remark = &empty
+ } else {
+ patch.Remark = req.Remark
+ }
+ }
if catPresent {
patch.SetCategoryID = true
patch.CategoryID = req.CategoryID
@@ -178,6 +208,13 @@ func (s *Server) pgUpdateTaskTemplate(w http.ResponseWriter, r *http.Request) {
patch.SetInterceptRules = true
patch.InterceptRules = rules
}
+ if workerPresent {
+ n := 0
+ if req.WorkerCount != nil {
+ n = *req.WorkerCount
+ }
+ patch.WorkerCount = &n
+ }
template, err := pg.PatchTaskTemplate(id, patch)
if err != nil {
writeTaskTemplateErr(w, err)
diff --git a/server/tools_wire_test.go b/server/tools_wire_test.go
index dd4850e..705138a 100644
--- a/server/tools_wire_test.go
+++ b/server/tools_wire_test.go
@@ -5,9 +5,9 @@ import (
"encoding/json"
"testing"
- actool "github.com/Autumn-27/norma/tool"
"github.com/Autumn-27/artex/agent"
"github.com/Autumn-27/artex/db"
+ actool "github.com/Autumn-27/norma/tool"
)
func names(tools []actool.CoreTool) map[string]actool.CoreTool {
diff --git a/web/src/app/(main)/function/tasks/detail/page.tsx b/web/src/app/(main)/function/tasks/detail/page.tsx
index 6bf52f5..db99f6c 100644
--- a/web/src/app/(main)/function/tasks/detail/page.tsx
+++ b/web/src/app/(main)/function/tasks/detail/page.tsx
@@ -414,6 +414,11 @@ function TaskDetailInner() {
{task.goal}
+ {task.remark ? (
+
+ 备注：{task.remark}
+
+ ) : null}
{/* Tabs */}
diff --git a/web/src/app/(main)/function/tasks/page.tsx b/web/src/app/(main)/function/tasks/page.tsx
index bc9ae91..3737d6d 100644
--- a/web/src/app/(main)/function/tasks/page.tsx
+++ b/web/src/app/(main)/function/tasks/page.tsx
@@ -137,6 +137,7 @@ import type {
TaskCategory,
TaskStatus,
} from "@/lib/types";
+import { MAX_TASK_REMARK_RUNES, MAX_TASK_WORKER_COUNT, parseWorkerCountInput } from "@/lib/types";
import { cn } from "@/lib/utils";
// fmtBytes renders a human file size for the upload manifest.
@@ -3104,6 +3105,7 @@ function CreateTaskSheet({
}) {
const [open, setOpen] = React.useState(false);
const [name, setName] = React.useState("");
+ const [remark, setRemark] = React.useState("");
const [categoryID, setCategoryID] = React.useState(undefined);
const [description, setDescription] = React.useState("");
const [goal, setGoal] = React.useState("");
@@ -3118,6 +3120,7 @@ function CreateTaskSheet({
const [heartbeatMin, setHeartbeatMin] = React.useState("10"); // planner 心跳(分钟);默认10,下限10(与后端一致)
const [seedFirstIntent, setSeedFirstIntent] = React.useState(false); // 创建时下发种子意图,worker 免等首轮 planner 直接开跑;默认关闭,走标准先规划再执行
const [coverageEnabled, setCoverageEnabled] = React.useState(true); // 资产覆盖度功能;默认开。关闭=不计算/展示覆盖度、不累积范围、隐藏范围类工具(company 关联不受影响)
+ const [workerCount, setWorkerCount] = React.useState(""); // 本任务 Worker 数;空/0=用全局默认
const [interceptRules, setInterceptRules] = React.useState([]); // 任务级资产拦截规则(仅本任务生效,不进全局表)
const [scheduleEnabled, setScheduleEnabled] = React.useState(false);
const [schedule, setSchedule] = React.useState({
@@ -3181,12 +3184,18 @@ function CreateTaskSheet({
toast.error(`最多关联 ${MAX_SOURCE_TASKS} 个来源任务`);
return;
}
+ const workerCountNum = parseWorkerCountInput(workerCount);
+ if (workerCountNum == null) {
+ toast.error(`Worker 运行数量必须是 0-${MAX_TASK_WORKER_COUNT} 的整数`);
+ return;
+ }
setCreating(true);
try {
const timeoutSec = Math.max(0, Math.floor(Number(timeoutMin) || 0)) * 60;
const heartbeatSec = Math.max(10, Math.floor(Number(heartbeatMin) || 10)) * 60; // 下限 10min，与后端归一一致
await api.createTask({
name: name.trim(),
+ remark: remark.trim(),
categoryId: categoryID,
description: description.trim(),
goal: goal.trim(),
@@ -3197,6 +3206,7 @@ function CreateTaskSheet({
seedFirstIntent,
planHeartbeatSeconds: heartbeatSec,
coverageEnabled,
+ workerCount: workerCountNum,
schedule: scheduleEnabled ? { ...schedule, task_ids: [] } : undefined,
interceptRules: interceptRules
.map((r) => ({ ...r, pattern: r.pattern.trim() }))
@@ -3204,6 +3214,7 @@ function CreateTaskSheet({
});
toast.success("任务已创建");
setName("");
+ setRemark("");
setCategoryID(undefined);
setDescription("");
setGoal("");
@@ -3215,6 +3226,7 @@ function CreateTaskSheet({
setHeartbeatMin("10");
setSeedFirstIntent(false);
setCoverageEnabled(true);
+ setWorkerCount("");
setInterceptRules([]);
setScheduleEnabled(false);
setSchedule({
@@ -3263,15 +3275,19 @@ function CreateTaskSheet({
{
setDescription(template.description);
setGoal(template.goal);
+ setRemark(template.remark ?? "");
setCategoryID(template.category_id ?? undefined);
setInterceptRules(template.intercept_rules ?? []);
+ setWorkerCount(template.worker_count ? String(template.worker_count) : "");
setUploadCount(0);
}}
portalContainer={sheetContentRef}
@@ -3285,6 +3301,17 @@ function CreateTaskSheet({
onChange={(e) => setName(e.target.value)}
/>
+
+
+