From 47ba58bda64a17e2f33314dfd5b3b3ce9d9cc050 Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:18:07 -0400 Subject: [PATCH] =?UTF-8?q?feat(planner):=20finding=20=E5=BD=93=E5=9C=BA?= =?UTF-8?q?=E5=94=A4=E9=86=92=20+=20planner=20N=20=E7=A7=92=E5=BF=83?= =?UTF-8?q?=E8=B7=B3=E8=A7=A6=E5=8F=91(=E4=BB=BB=E5=8A=A1=E5=8F=82?= =?UTF-8?q?=E6=95=B0,=E9=BB=98=E8=AE=A45min)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 触发机制补两条(见 docs/planner-trigger-impl-plan.md)： 1) finding 当场唤醒：worker 飞行中 report_finding 落库即唤醒本任务 planner， 不再等 worker 整趟收工才反应。 - worker.Execute 加 notify 入参并 SetNotify(补上 worker toolset 之前缺失的接线) - addFinding 写 KindFinding 成功后调 notify(只对 finding,不对 fact) - engine 两处 worker.Execute 调用传 t.Notify 2) planner N 秒心跳：距上轮规划结束/任务开始满 N 秒且期间无触发 → 触发一轮， 兜底卡死 + 唤醒去监督飞行中的 worker(steer/kill) + 周期复查。 - plannerLoop 重构为 runRound 闭包 + 三臂 select(ctx/notify/heartbeat) - 心跳在 loop 入口臂 = 从任务 start 计时(覆盖 seed 跳过首轮)，每轮结束重臂 - 新增任务参数 plan_heartbeat_seconds，下限=默认=300(5min),低于自动抬到300 - DB 迁移：schema.sql 加列 + ALTER ADD COLUMN IF NOT EXISTS 补旧库 - 全链路接线：db.CreateTask/manager/server HTTP/orchestration spawn_task - 前端：api.createTask 传参 + 建任务表单新增"planner 心跳(分钟)"字段 --- agent/tools.go | 3 + agent/worker.go | 5 +- db/schema.sql | 3 + db/tasks.go | 27 +++- db/tasks_test.go | 2 +- server/engine.go | 164 +++++++++++++-------- server/manager.go | 14 +- server/orchestration.go | 4 +- server/server.go | 5 +- web/src/app/(main)/function/tasks/page.tsx | 20 ++- web/src/lib/api.ts | 3 +- 11 files changed, 168 insertions(+), 82 deletions(-) diff --git a/agent/tools.go b/agent/tools.go index b78db01..7a940cf 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -647,6 +647,9 @@ func (t *ToolSet) addFinding() actool.CoreTool { _ = t.ts.Link(intent, db.RelYields, id) // chain: intent -> finding } _, _ = t.ts.AddStandaloneFinding(t.taskID, id, a.VulnClass, a.Severity, a.Summary, a.Evidence, t.worker, anchors) + if t.notify != nil { + t.notify() // 确证漏洞落库 → 当场唤醒本任务 planner（不等 worker 收工，debounce 合并） + } } else { // conversation context: no exploration store available, cannot record finding return actool.Errorf("report_finding 需要任务上下文（exploration store 未初始化）"), nil diff --git a/agent/worker.go b/agent/worker.go index 44f431e..82d7e27 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -234,11 +234,13 @@ func renderWorkerGraphOverview(data map[string]any) string { // Execute runs one intent. hooks (the per-task Guard) gates every tool call; may // be nil. emit, if non-nil, receives one ActivityRecord per execution step. +// notify, if non-nil, is called when this worker writes a finding (report_finding) +// so the task's planner wakes mid-flight instead of waiting for the worker to finish. // Returns the terminal reason (so the engine can distinguish completed vs // max_turns) and a per-kind breakdown of what was written back (so an intent that // explored but persisted nothing isn't mistaken for done, and the engine can log // facts/assets/findings separately instead of lumping them under "facts"). -func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger) (harness.TerminalReason, WriteCounts, error) { +func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger, notify func()) (harness.TerminalReason, WriteCounts, error) { tsx := NewToolSet(ts, name) tsx.SetTaskID(taskID) if as != nil { @@ -246,6 +248,7 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db. } tsx.SetOwnerNode(intent.ID) // assets this worker discovers anchor to its intent → visible to the task tsx.SetEnrich(enr) // async DNS/HTTP auto-completion for assets this worker writes + tsx.SetNotify(notify) // finding 写回时当场唤醒本任务 planner（report_finding 触发） // base = built-in worker tools ∪ host tools (traffic) ∪ default tools (incl. Bash); // then augment with the agent's visible skills/MCP. During the SDK settlement // phase, Bash is hidden via Settlement.DisabledTools (no local gating needed). diff --git a/db/schema.sql b/db/schema.sql index 2c3ba40..8a529c7 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -238,6 +238,7 @@ CREATE TABLE IF NOT EXISTS tasks ( company_id BIGINT REFERENCES companies(id) ON DELETE SET NULL, parent_ref TEXT, timeout_seconds INTEGER NOT NULL DEFAULT 0, + plan_heartbeat_seconds INTEGER NOT NULL DEFAULT 300, first_run_at TIMESTAMPTZ, deadline_at TIMESTAMPTZ, deleted_at TIMESTAMPTZ, @@ -250,6 +251,8 @@ CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status) WHERE dele DROP TRIGGER IF EXISTS trg_tasks_upd ON tasks; 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; -- 任务测试范围（资产覆盖度的分母 + 授权边界）。 -- 自动填(source='auto')：insertAssets 顶层按 worker 显式插入的资产类型加保守范围 diff --git a/db/tasks.go b/db/tasks.go index 48cb19b..940d4f1 100644 --- a/db/tasks.go +++ b/db/tasks.go @@ -21,6 +21,9 @@ type Task struct { TimeoutSeconds int `json:"timeout_seconds"` // 0=不限时 FirstRunAt *time.Time `json:"first_run_at,omitempty"` // 首次真正开始运行的时刻(非 created_at);nil=尚未运行 DeadlineAt *time.Time `json:"deadline_at,omitempty"` // = first_run_at + timeout_seconds;nil=不限或未运行 + // planner 心跳触发间隔(秒)：距上轮 plan 结束/任务开始满该值且期间无触发 → 触发一轮。 + // 下限=默认=300(5min)，低于一律抬到 300(在 CreateTask 归一)。见 docs/planner-trigger-impl-plan.md + PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` } // IsTerminal reports whether a task status is a terminal (finished) state. @@ -32,7 +35,18 @@ func IsTerminal(status string) bool { // CreateTask creates an exploration + task in one transaction and returns the task. // timeoutSeconds is the task-level wall-clock budget (0 = 不限时); deadline_at is // stamped later at first real run (see engine), not here. -func (d *DB) CreateTask(description, goal string, llmProfileID *int64, timeoutSeconds int) (*Task, error) { +// MinPlanHeartbeatSeconds 是 planner 心跳间隔的下限 = 默认 = 5min。 +// 低于它(含缺省 0 / 负值 / 误配的小值)一律抬到 5min，防止把 planner 打爆。 +const MinPlanHeartbeatSeconds = 300 + +func normalizeHeartbeat(sec int) int { + if sec < MinPlanHeartbeatSeconds { + return MinPlanHeartbeatSeconds + } + return sec +} + +func (d *DB) CreateTask(description, goal string, llmProfileID *int64, timeoutSeconds, planHeartbeatSeconds int) (*Task, error) { tx, err := d.Begin() if err != nil { return nil, err @@ -60,20 +74,21 @@ VALUES ($1, 'fact', $2, 0, 'origin', 'system')`, expID, string(originPayload)); if timeoutSeconds < 0 { timeoutSeconds = 0 } - t := &Task{Description: description, Goal: goal, ExplorationID: expID, LLMProfileID: llmProfileID, TimeoutSeconds: timeoutSeconds} + planHeartbeatSeconds = normalizeHeartbeat(planHeartbeatSeconds) // <=0→300，<30→30 + t := &Task{Description: description, Goal: goal, ExplorationID: expID, LLMProfileID: llmProfileID, TimeoutSeconds: timeoutSeconds, PlanHeartbeatSeconds: planHeartbeatSeconds} if err := tx.QueryRow(` -INSERT INTO tasks(description, goal, exploration_id, llm_profile_id, timeout_seconds) VALUES ($1,$2,$3,$4,$5) -RETURNING id, status, paused, created_at`, description, goal, expID, llmProfileID, timeoutSeconds).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil { +INSERT INTO tasks(description, goal, exploration_id, llm_profile_id, timeout_seconds, plan_heartbeat_seconds) VALUES ($1,$2,$3,$4,$5,$6) +RETURNING id, status, paused, created_at`, description, goal, expID, llmProfileID, timeoutSeconds, planHeartbeatSeconds).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil { return nil, err } 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), first_run_at, deadline_at` +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` 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.FirstRunAt, &t.DeadlineAt); err != nil { + 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 { return nil, err } return &t, nil diff --git a/db/tasks_test.go b/db/tasks_test.go index 6139091..f2e9def 100644 --- a/db/tasks_test.go +++ b/db/tasks_test.go @@ -9,7 +9,7 @@ func TestTaskLifecycleAndDeleteCascade(t *testing.T) { } defer d.Close() - tk, err := d.CreateTask("迁移测试", "目标X", nil, 0) + tk, err := d.CreateTask("迁移测试", "目标X", nil, 0, 0) if err != nil { t.Fatal(err) } diff --git a/server/engine.go b/server/engine.go index 1bcbe79..3c5a5fa 100644 --- a/server/engine.go +++ b/server/engine.go @@ -441,74 +441,112 @@ func (e *Engine) Run(ctx context.Context, t *Task) { } } +// plannerHeartbeatInterval 解析任务的 planner 心跳间隔。db.CreateTask 已归一 +// (低于 300 一律抬到 300);这里再兜一次底,防内存态异常值。 +func plannerHeartbeatInterval(t *Task) time.Duration { + sec := t.PlanHeartbeatSeconds + if sec < db.MinPlanHeartbeatSeconds { // 下限=默认=300(5min) + sec = db.MinPlanHeartbeatSeconds + } + return time.Duration(sec) * time.Second +} + +// resetPlannerTimer 安全重臂一个可能已触发的 Timer(标准 Stop→drain→Reset 模式)。 +func resetPlannerTimer(timer *time.Timer, d time.Duration) { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(d) +} + func (e *Engine) plannerLoop(ctx context.Context, t *Task) { + interval := plannerHeartbeatInterval(t) + // 心跳定时器在 loop 入口臂 = 从任务 start 计时:即使是跳过首轮 planner 的 seed 任务 + // (Run 里 frontier 非空不 kick 首轮)、这里一直阻塞,心跳也会在「任务 start + interval」 + // 触发第一轮规划。之后每次唤醒(边沿/心跳)都重臂 = 距上次任意规划触发的时长。 + heartbeat := time.NewTimer(interval) + defer heartbeat.Stop() + + // runRound 跑一轮规划(含 debounce 合并 + 各 guard)。src 仅用于日志区分触发来源。 + runRound := func(src string) { + // debounce: coalesce a burst of changes into one planning round + timer := time.NewTimer(e.debounce) + drain: + for { + select { + case <-t.notify: + case <-timer.C: + break drain + } + } + planner, _ := e.snapshotFor(t) + if planner == nil { + return // idle until LLM configured + } + if e.IsPaused(t.ID) { + return // user-paused: don't plan + } + // terminal task (goals all met → done, or failed): the run is over. A + // resume/nudge — e.g. auto-resume of the active task on restart — must NOT + // re-plan (it would burn an LLM round and re-confirm a settled result). + if isTerminalStatus(t.Status) { + return + } + // 任务级超时收尾中:丢弃普通唤醒——worker 收尾写回、Resume 的 Notify 都不再 + // 触发常规规划轮;终局那一轮由协调器(settleTask)直接驱动,不走这里。 + if e.isSettling(t.ID) { + return + } + e.stampFirstRun(t) // 首次真正规划 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务) + e.touch(t.ID) + emit := func(r db.Activity) { e.emitActivity(t, r) } + ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline + log.Printf("[planner] task %s 规划中…(%s 触发)", t.ID, src) + // round marker: each Plan() is one planner round; emit a boundary so the + // UI can separate rounds in the transcript (kind='round'). + e.emitActivity(t, db.Activity{Worker: "planner", Kind: "round", + Summary: fmt.Sprintf("第 %d 轮规划", e.nextPlannerRound(t.ID))}) + // which intents' completion triggered this round (may be several — debounce + // coalesces a burst; empty for non-completion wakes like a new hint / heartbeat). + doneIntents := t.drainDone() + e.incInflight(t.ID) + taskIDInt, _ := strconv.ParseInt(t.ID, 10, 64) + met, reason, err := planner.Plan(ectx, taskIDInt, e.m.assets, t.Store, t.Goal, doneIntents, emit) + e.decInflight(t.ID) + switch { + case err != nil && ectx.Err() == nil: + log.Printf("[planner] task %s 规划出错: %v", t.ID, err) + case met: + log.Printf("[planner] task %s 判定目标达成: %s", t.ID, reason) + // 所有目标达成 → 持久化任务状态为 done（前端 DTO 会优先展示该终态）。 + if err := e.m.SetTaskStatus(t.ID, "done"); err != nil { + log.Printf("[planner] task %s 标记完成落库失败: %v", t.ID, err) + } + // 任务已判完成 → 立刻取消在跑的 worker：它们手头的意图跑出来也没意义了。 + // 下一轮 worker 循环撞终态门就不再领新意图;被取消的这批走下方"任务已完成"分支 + // 归为 stopped(而非 blocked)。 + e.cancelExec(t.ID) + default: + log.Printf("[planner] task %s 规划完成", t.ID) + } + e.touch(t.ID) + } + for { select { case <-ctx.Done(): return case <-t.notify: - // debounce: coalesce a burst of changes into one planning round - timer := time.NewTimer(e.debounce) - drain: - for { - select { - case <-t.notify: - case <-timer.C: - break drain - } - } - planner, _ := e.snapshotFor(t) - if planner == nil { - continue // idle until LLM configured - } - if e.IsPaused(t.ID) { - continue // user-paused: don't plan - } - // terminal task (goals all met → done, or failed): the run is over. A - // resume/nudge — e.g. auto-resume of the active task on restart — must NOT - // re-plan (it would burn an LLM round and re-confirm a settled result). - if isTerminalStatus(t.Status) { - continue - } - // 任务级超时收尾中:丢弃普通唤醒——worker 收尾写回、Resume 的 Notify 都不再 - // 触发常规规划轮;终局那一轮由协调器(settleTask)直接驱动,不走这里。 - if e.isSettling(t.ID) { - continue - } - e.stampFirstRun(t) // 首次真正规划 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务) - e.touch(t.ID) - emit := func(r db.Activity) { e.emitActivity(t, r) } - ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline - log.Printf("[planner] task %s 规划中…", t.ID) - // round marker: each Plan() is one planner round; emit a boundary so the - // UI can separate rounds in the transcript (kind='round'). - e.emitActivity(t, db.Activity{Worker: "planner", Kind: "round", - Summary: fmt.Sprintf("第 %d 轮规划", e.nextPlannerRound(t.ID))}) - // which intents' completion triggered this round (may be several — debounce - // coalesces a burst; empty for non-completion wakes like a new hint). - doneIntents := t.drainDone() - e.incInflight(t.ID) - taskIDInt, _ := strconv.ParseInt(t.ID, 10, 64) - met, reason, err := planner.Plan(ectx, taskIDInt, e.m.assets, t.Store, t.Goal, doneIntents, emit) - e.decInflight(t.ID) - switch { - case err != nil && ectx.Err() == nil: - log.Printf("[planner] task %s 规划出错: %v", t.ID, err) - case met: - log.Printf("[planner] task %s 判定目标达成: %s", t.ID, reason) - // 所有目标达成 → 持久化任务状态为 done（前端 DTO 会优先展示该终态）。 - if err := e.m.SetTaskStatus(t.ID, "done"); err != nil { - log.Printf("[planner] task %s 标记完成落库失败: %v", t.ID, err) - } - // 任务已判完成 → 立刻取消在跑的 worker：它们手头的意图跑出来也没意义了。 - // 下一轮 worker 循环撞终态门就不再领新意图;被取消的这批走下方"任务已完成"分支 - // 归为 stopped(而非 blocked)。 - e.cancelExec(t.ID) - default: - log.Printf("[planner] task %s 规划完成", t.ID) - } - e.touch(t.ID) + runRound("edge") // worker 结束 / finding / kill / resume / seed 首轮 + case <-heartbeat.C: + // 周期兜底:死锁兜底 + 唤醒去监督飞行中的 worker(steer/kill) + 周期复查。 + runRound("heartbeat") } + // 每次唤醒(边沿或心跳)后重臂心跳:任意规划触发都重算这段静置计时。 + resetPlannerTimer(heartbeat, interval) } } @@ -571,7 +609,7 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { hooks := steerHooks{inner: t.Guard.Hooks(), drain: func() (string, bool) { return e.drainSteer(iid) }} wTaskID, _ := strconv.ParseInt(t.ID, 10, 64) e.incInflight(t.ID) // 计入在跑,供收尾时序 drain 等待 - reason, wrote, err := worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich) + reason, wrote, err := worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.Notify) // model_error 收场 → 额外重跑几次（退避后再试）。仅在意图仍属本 work、任务 // 未暂停/未终止/未取消【且未进入收尾】时重试；否则让位给对应分支处理(收尾期不 // 再重试,避免退避挤占其他 worker 的优雅收尾窗口)。 @@ -583,7 +621,7 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { if sleepCtx(workCtx, modelErrorRetryBackoff) { break // 退避期间被取消（终止/暂停）→ 交给下方分支处理 } - reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich) + reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.Notify) } e.decInflight(t.ID) // CAPTURE kill state BEFORE unregisterWork cancels workCtx. kill = this work's diff --git a/server/manager.go b/server/manager.go index 52ea0a8..98a7169 100644 --- a/server/manager.go +++ b/server/manager.go @@ -36,9 +36,10 @@ type Task struct { LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // 指定运行本任务 planner/worker 的 LLM 配置;nil=用全局激活配置 Status string `json:"status"` // persisted lifecycle status (done/failed/timeout 为终态；空/其它则由运行态推导) // 任务级超时(见 docs/任务级超时与收尾设计.md)。DeadlineAt/FirstRunAt 为 unix 秒,0=未设/未运行。 - TimeoutSeconds int `json:"timeout_seconds"` - FirstRunAt int64 `json:"first_run_at,omitempty"` - DeadlineAt int64 `json:"deadline_at,omitempty"` + TimeoutSeconds int `json:"timeout_seconds"` + PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner 心跳触发间隔(秒) + FirstRunAt int64 `json:"first_run_at,omitempty"` + DeadlineAt int64 `json:"deadline_at,omitempty"` Store *pgdb.ExplorationStore `json:"-"` Guard *guard.Guard `json:"-"` notify chan struct{} @@ -459,15 +460,16 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef, LLMProfileID: pt.LLMProfileID, - TimeoutSeconds: pt.TimeoutSeconds, FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt), + TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds, + FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt), Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1), } } // CreateTask creates a task + its exploration and makes it active. // timeoutSeconds is the task-level wall-clock budget (0 = 不限时). -func (m *Manager) CreateTask(description, goal string, llmProfileID *int64, timeoutSeconds int) (*Task, error) { - pt, err := m.pg.CreateTask(description, goal, llmProfileID, timeoutSeconds) +func (m *Manager) CreateTask(description, goal string, llmProfileID *int64, timeoutSeconds, planHeartbeatSeconds int) (*Task, error) { + pt, err := m.pg.CreateTask(description, goal, llmProfileID, timeoutSeconds, planHeartbeatSeconds) if err != nil { return nil, err } diff --git a/server/orchestration.go b/server/orchestration.go index 57eac71..c69ec45 100644 --- a/server/orchestration.go +++ b/server/orchestration.go @@ -239,6 +239,7 @@ func (s *Server) toolSpawnTask() actool.CoreTool { "parent_ref": strParam("可选：父任务 id(做父子关联)"), "llm_profile_id": map[string]any{"type": "integer", "description": "可选：指定本子任务 planner/worker 用的 LLM 配置 id(见 list_llm_profiles)；留空则继承父任务、再回退全局激活配置"}, "timeout_seconds": map[string]any{"type": "integer", "description": "可选：任务级超时(秒)。到点后触发优雅收尾并进入 timeout 终态；留空或 0 = 不限时"}, + "plan_heartbeat_seconds": map[string]any{"type": "integer", "description": "可选：planner 心跳触发间隔(秒)。距上轮规划结束/任务开始满该值且期间无触发 → 触发一轮规划(兜底死锁 + 唤醒去监督飞行中的 worker)。留空或 0 = 默认 300(5min)；"}, "seed_first_intent": map[string]any{"type": "boolean", "description": "可选：默认 true。true=创建即把「描述+目标」作为一条种子意图下发，worker 免等首轮 planner 直接开跑(CTF 等常一个 work 解决的场景推荐)；false=走标准的先规划再执行"}, }, "description", "goal"), func(_ context.Context, in json.RawMessage) (actool.Result, error) { @@ -246,6 +247,7 @@ func (s *Server) toolSpawnTask() actool.CoreTool { Description, Goal, ParentRef string LLMProfileID json.RawMessage `json:"llm_profile_id"` TimeoutSeconds int `json:"timeout_seconds"` + PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` SeedFirstIntent *bool `json:"seed_first_intent"` } _ = json.Unmarshal(in, &a) @@ -270,7 +272,7 @@ func (s *Server) toolSpawnTask() actool.CoreTool { pin = pt.LLMProfileID } } - t, err := s.m.CreateTask(a.Description, a.Goal, pin, a.TimeoutSeconds) + t, err := s.m.CreateTask(a.Description, a.Goal, pin, a.TimeoutSeconds, a.PlanHeartbeatSeconds) if err != nil { return actool.Errorf(err.Error()), nil } diff --git a/server/server.go b/server/server.go index 635ab8b..521a749 100644 --- a/server/server.go +++ b/server/server.go @@ -1016,7 +1016,8 @@ type createTaskReq struct { Description string `json:"description"` Goal string `json:"goal"` LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // 指定运行本任务的 LLM 配置;省略/null=用激活配置 - TimeoutSeconds int `json:"timeout_seconds"` // 任务级超时(秒);0/省略=不限时 + TimeoutSeconds int `json:"timeout_seconds"` // 任务级超时(秒);0/省略=不限时 + PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner 心跳触发间隔(秒);0/省略=默认300(5min);下限=默认=300,低于自动抬到300 SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // 创建时直接下发一条种子意图(内容=描述+目标),让 worker 免等首轮 planner 直接开跑;省略/null=默认开启。CTF 常一 work 解决,省掉开跑前的 planner 轮。 } @@ -1040,7 +1041,7 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) { if req.TimeoutSeconds < 0 { req.TimeoutSeconds = 0 } - t, err := s.m.CreateTask(req.Description, req.Goal, req.LLMProfileID, req.TimeoutSeconds) + t, err := s.m.CreateTask(req.Description, req.Goal, req.LLMProfileID, req.TimeoutSeconds, req.PlanHeartbeatSeconds) if err != nil { writeErr(w, 500, err.Error()) return diff --git a/web/src/app/(main)/function/tasks/page.tsx b/web/src/app/(main)/function/tasks/page.tsx index 858ff50..cd5b8bc 100644 --- a/web/src/app/(main)/function/tasks/page.tsx +++ b/web/src/app/(main)/function/tasks/page.tsx @@ -111,6 +111,7 @@ export default function TasksPage() { const [profiles, setProfiles] = React.useState([]); const [llmProfile, setLlmProfile] = React.useState(ACTIVE_PROFILE); // sentinel = active const [timeoutMin, setTimeoutMin] = React.useState(""); // 任务级超时(分钟);空/0 = 不限时 + const [heartbeatMin, setHeartbeatMin] = React.useState("5"); // planner 心跳(分钟);默认5,下限5(与后端一致) const [seedFirstIntent, setSeedFirstIntent] = React.useState(true); // 创建时下发种子意图,worker 免等首轮 planner 直接开跑 const [open, setOpen] = React.useState(false); const [query, setQuery] = React.useState(""); @@ -173,12 +174,14 @@ export default function TasksPage() { try { const pid = llmProfile === ACTIVE_PROFILE ? undefined : Number(llmProfile); const timeoutSec = Math.max(0, Math.floor(Number(timeoutMin) || 0)) * 60; - await api.createTask(description.trim(), goal.trim(), pid, timeoutSec, seedFirstIntent); + const heartbeatSec = Math.max(5, Math.floor(Number(heartbeatMin) || 5)) * 60; // 下限 5min，与后端归一一致 + await api.createTask(description.trim(), goal.trim(), pid, timeoutSec, seedFirstIntent, heartbeatSec); toast.success("任务已创建"); setDescription(""); setGoal(""); setLlmProfile(ACTIVE_PROFILE); setTimeoutMin(""); + setHeartbeatMin("5"); setSeedFirstIntent(true); setOpen(false); load(); @@ -303,6 +306,21 @@ export default function TasksPage() { 到点后触发优雅收尾（各 agent 写回 + planner 终局判定），任务进入 timeout 终态。

+
+ + setHeartbeatMin(e.target.value)} + /> +

+ 距上轮规划结束/任务开始满该时长且期间无触发，自动触发一轮规划（兜底卡死 + 唤醒去监督在跑的 worker）。下限 5 分钟。 +

+