From 678025f850da9ea9e08feee9585d9e861ea733fb Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:18:06 -0400 Subject: [PATCH] =?UTF-8?q?feat(planner):=20=E8=A7=A6=E5=8F=91=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E6=96=87=E5=B8=A6=E5=AE=9E=E9=99=85=E5=8F=98=E5=8A=A8?= =?UTF-8?q?=20+=20=E5=BF=83=E8=B7=B3=E8=A7=A6=E5=8F=91=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E3=80=8C=E5=AE=9A=E6=97=B6=E5=B7=A1=E6=A3=80=E3=80=8D=E6=8E=AA?= =?UTF-8?q?=E8=BE=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把 planner 每轮的触发信号从「固定一句"图发生了变化"」升级为具体变动： - 统一 TriggerEvent{Kind,IntentID,Detail} 累加器(替换原 doneIntents): worker 结束→{done}, report_finding→{finding,摘要}。Task.NotifyDone/新增 NotifyFinding 记录, drainTriggers 消费, Plan 参数改为 []TriggerEvent。 - renderTriggers 渲染:worker 结束→哪个意图(摘要)+worker 输出结论(取 result/text 全文,截断800字);finding→哪个意图+finding 摘要。 - worker 回调 notify()→notifyFinding(intentID,summary),report_finding 落库带上下文。 - 时间/心跳触发(triggers 为空)不再谎称"图发生了变化",开场白改为「定时巡检」并 提示复查在跑意图(steer_work 纠偏/kill_work 止损),对齐心跳的飞行中监督职责。 --- agent/planner.go | 108 +++++++++++++++++++++++++++++++++------ agent/tools.go | 19 ++++++- agent/worker.go | 13 ++--- server/engine.go | 12 ++--- server/engine_timeout.go | 2 +- server/manager.go | 47 ++++++++++------- 6 files changed, 151 insertions(+), 50 deletions(-) diff --git a/agent/planner.go b/agent/planner.go index af105bc..4dcde86 100644 --- a/agent/planner.go +++ b/agent/planner.go @@ -91,18 +91,88 @@ func renderPlannerTodos(items []actool.Todo) string { return b.String() } -// renderDoneIntents surfaces the intent(s) whose completion triggered this round -// so the planner looks first at their fresh yields (in the态势 below) before -// deciding whether a new direction opened up. Empty for non-completion wakes. -func renderDoneIntents(ids []int64) string { - if len(ids) == 0 { +// TriggerEvent describes what concretely caused this planning round to fire, so +// the planner looks first at the actual change instead of re-scanning the whole +// overview. Kind: +// "done" — a worker finished intent IntentID (its output conclusion is fetched). +// "finding" — a worker reported a finding on intent IntentID (Detail = 摘要). +type TriggerEvent struct { + Kind string + IntentID int64 + Detail string +} + +// renderTriggers spells out the change(s) that fired this round: for a finished +// worker — which intent + its output conclusion; for a finding — which intent + +// what was found. Empty for time/heartbeat wakes. Reads the store (best-effort; +// a blank field never blocks the round). +func renderTriggers(ts *db.ExplorationStore, evs []TriggerEvent) string { + if len(evs) == 0 || ts == nil { return "" } - parts := make([]string, len(ids)) - for i, id := range ids { - parts[i] = fmt.Sprintf("#%d", id) + var b strings.Builder + b.WriteString("\n\n【本次触发本轮的实际变动（先看这里，再决定是否补方向）】：") + for _, ev := range evs { + sum := intentSummary(ts, ev.IntentID) + switch ev.Kind { + case "finding": + b.WriteString(fmt.Sprintf("\n- 意图 #%d（%s）的 worker 报告了一个 finding：%s", ev.IntentID, sum, ev.Detail)) + default: // "done" + b.WriteString(fmt.Sprintf("\n- 意图 #%d（%s）的 worker 结束，输出结论：%s", ev.IntentID, sum, workerOutput(ts, ev.IntentID))) + } } - return "\n\n【本次触发本轮的、刚完成的意图】：" + strings.Join(parts, "、") + b.WriteString("\n（完整细节可 node_detail / get_worker_output / list_findings 再查。）") + return b.String() +} + +// intentSummary reads an intent node's one-line summary (best-effort, "?" on miss). +func intentSummary(ts *db.ExplorationStore, id int64) string { + n, err := ts.GetNode(id) + if err != nil || n == nil { + return "?" + } + var p map[string]any + if json.Unmarshal(n.Payload, &p) == nil { + if s, ok := p["summary"].(string); ok && s != "" { + return s + } + } + return "?" +} + +// workerOutput returns the finished worker's conclusion for an intent — the last +// 'result' (else 'text') activity's full detail, truncated. Same source get_worker_output uses. +func workerOutput(ts *db.ExplorationStore, id int64) string { + acts, _, err := ts.ActivityList(&id, 0, 1000) + if err != nil { + return "(取输出失败)" + } + var pick *db.Activity + for i := range acts { + if acts[i].Kind == "result" { + pick = &acts[i] + } else if acts[i].Kind == "text" && pick == nil { + pick = &acts[i] + } + } + if pick == nil { + return "(该 work 尚无输出记录)" + } + out, _ := ts.ActivityDetail(pick.ID) + if out == "" { + out = pick.Summary + } + return truncOutput(out, 800) +} + +// truncOutput caps a worker-output blob so the trigger context doesn't bloat the +// system prompt every round; full text is one get_worker_output call away. +func truncOutput(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n]) + " …（已截断，完整见 get_worker_output）" } // renderGraphOverview folds the pre-computed graph_overview snapshot into the @@ -167,11 +237,11 @@ func plannerSystem(goal, workDir string) string { // steps (so users can see how it reads the situation and judges goals — the // planner is the intent generator and was previously a black box). Returns whether // the planner judged the goal met. -// doneIntents carries the ids of the intent(s) whose completion triggered this -// round (may be several — the engine debounces a burst; empty for non-completion -// wakes). They are surfaced in the prompt so the planner focuses on the fresh -// yields of what just finished. -func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, goal string, doneIntents []int64, emit func(db.Activity)) (met bool, reason string, err error) { +// triggers carries the concrete change(s) that fired this round — worker(s) done +// and/or finding(s) reported (may be several — the engine debounces a burst; empty +// for time/heartbeat wakes). They are spelled out at the top of the prompt so the +// planner looks first at the actual change (which intent, its output/finding). +func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, goal string, triggers []TriggerEvent, emit func(db.Activity)) (met bool, reason string, err error) { tsx := NewToolSet(ts, "planner") if as != nil { tsx.SetAssetStore(as, as.Companies()) @@ -189,7 +259,7 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts // 关键态势（刚完成的意图 + 预取的完整图）放进 system prompt，而不是 user 输入： // system 段永不被 compaction 压缩，长回合里也保证这份态势始终在场；轮内固定，还能吃到 // system 缓存。它落在指令之后、deferred 工具块之前（deferredSystem 保证块在最后）。 - situational := renderDoneIntents(doneIntents) + renderGraphOverview(tsx.graphOverviewData()) + situational := renderTriggers(ts, triggers) + renderGraphOverview(tsx.graphOverviewData()) // 任务级 deadline / 终局模式(经 ctx 注入,见 taskclock.go)。终局那一轮把任务超时 // planner 收尾词作为【本轮操作指令】直接注入 system,让它只做最后目标判定、不产新意图。 tc := taskClockFrom(ctx) @@ -243,7 +313,13 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts } // 态势（刚完成的意图 + 完整图）已在上面拼进 system prompt（抗压缩、轮内常驻）。这里 // 的 user 输入只留指令 + 跨唤醒待办（todo 是模型自己的规划便签，可再生，放 user 即可）。 - input := "图发生了变化，请规划下一步：读取上方 system 里的态势，判定目标。**本轮若无未被覆盖的新方向，直接结束即可（生成 0 个意图是正常且常见的，尤其刚派完意图在等 worker 产出时）。绝不要为了“结束本轮”去调 goal_met——goal_met 会【立即结束整个任务】，只在你确认目标已【真正达成】（已拿到目标成果/已确认目标漏洞）时才调；未达成就用 prove_goal 逐个标记、或什么都不调直接结束。**" + + // 开场白按「本轮有无具体变动」分两种：有变动 → 指向上方【实际变动】块；无变动 + // (心跳定时巡检 / hint / 恢复等) → 别谎称"图发生了变化",转而提示顺带复查在跑意图。 + lead := "刚有具体变动（见上方 system 的【本次触发本轮的实际变动】），据此规划下一步：" + if len(triggers) == 0 { + lead = "本轮是**定时巡检（心跳到点）/无具体变动信号**的唤醒——图不一定有新变动。顺带复查在跑意图：长时间无进展或跑偏的用 steer_work 纠偏、方向整个错的用 kill_work 止损；再判定目标、决定是否补方向：" + } + input := lead + "读取上方 system 里的态势，判定目标。**本轮若无未被覆盖的新方向，直接结束即可（生成 0 个意图是正常且常见的，尤其刚派完意图在等 worker 产出时）。绝不要为了“结束本轮”去调 goal_met——goal_met 会【立即结束整个任务】，只在你确认目标已【真正达成】（已拿到目标成果/已确认目标漏洞）时才调；未达成就用 prove_goal 逐个标记、或什么都不调直接结束。**" + renderPlannerTodos(opts.Todos.List()) // 有 deadline 夹逼时加硬 ctx 兜底(软预算 + grace),防单轮卡死绕过轮边界软超时。 runCtx := ctx diff --git a/agent/tools.go b/agent/tools.go index 7a940cf..6a9d10c 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -73,12 +73,19 @@ type ToolSet struct { // re-planned promptly (currently: a new hint). nil = no wake (the hint is still // stored and read on the next round triggered by other events). debounced. notify func() + // notifyFinding, if set, wakes the task's planner when this run reports a finding, + // carrying (intentID, summary) so the round can spell out which intent found what. + // Wired for workers; nil elsewhere → falls back to notify (bare wake). + notifyFinding func(intentID int64, summary string) } // SetNotify wires the planner-wake callback (see ToolSet.notify). Set by callers // that hold the task handle (main-agent chat, cross-task orchestration). func (t *ToolSet) SetNotify(fn func()) { t.notify = fn } +// SetNotifyFinding wires the finding-wake callback (see ToolSet.notifyFinding). +func (t *ToolSet) SetNotifyFinding(fn func(int64, string)) { t.notifyFinding = fn } + // EnrichTrigger is the enrichment engine seen from the tool layer (see package // enrich). Kept as an interface here to avoid coupling agent → enrich. type EnrichTrigger interface { @@ -647,8 +654,16 @@ 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 合并） + // 确证漏洞落库 → 当场唤醒本任务 planner（不等 worker 收工，debounce 合并）。 + // 优先带上下文(哪个意图+finding摘要);intent 用工具参数,缺省回退到 owner 意图。 + if t.notifyFinding != nil { + iid := pid(a.IntentID) + if iid <= 0 { + iid = t.ownerNode + } + t.notifyFinding(iid, a.Summary) + } else if t.notify != nil { + t.notify() } } else { // conversation context: no exploration store available, cannot record finding diff --git a/agent/worker.go b/agent/worker.go index 82d7e27..f71eede 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -234,21 +234,22 @@ 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. +// notifyFinding, if non-nil, is called (intentID, summary) when this worker writes +// a finding (report_finding) so the task's planner wakes mid-flight — with context +// on which intent found what — 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, notify func()) (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, notifyFinding func(int64, string)) (harness.TerminalReason, WriteCounts, error) { tsx := NewToolSet(ts, name) tsx.SetTaskID(taskID) if as != nil { tsx.SetAssetStore(as, as.Companies()) } - 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 触发） + 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.SetNotifyFinding(notifyFinding) // report_finding 落库时当场唤醒 planner，带上「哪个意图+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/server/engine.go b/server/engine.go index c64c5d3..78eb2aa 100644 --- a/server/engine.go +++ b/server/engine.go @@ -511,12 +511,12 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) { // 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() + // what fired this round (worker done / finding; may be several — debounce + // coalesces a burst; empty for time/heartbeat wakes). + triggers := t.drainTriggers() 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) + met, reason, err := planner.Plan(ectx, taskIDInt, e.m.assets, t.Store, t.Goal, triggers, emit) e.decInflight(t.ID) switch { case err != nil && ectx.Err() == nil: @@ -611,7 +611,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, t.Notify) + reason, wrote, err := worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding) // model_error 收场 → 额外重跑几次（退避后再试）。仅在意图仍属本 work、任务 // 未暂停/未终止/未取消【且未进入收尾】时重试；否则让位给对应分支处理(收尾期不 // 再重试,避免退避挤占其他 worker 的优雅收尾窗口)。 @@ -623,7 +623,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, t.Notify) + reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding) } e.decInflight(t.ID) // CAPTURE kill state BEFORE unregisterWork cancels workCtx. kill = this work's diff --git a/server/engine_timeout.go b/server/engine_timeout.go index ccfbe35..14fa3fe 100644 --- a/server/engine_timeout.go +++ b/server/engine_timeout.go @@ -209,7 +209,7 @@ func (e *Engine) runFinalPlannerRound(ctx context.Context, t *Task) (met bool) { fctx := e.clockCtx(ctx, t, true) e.incInflight(t.ID) tTaskID, _ := strconv.ParseInt(t.ID, 10, 64) - met, reason, err := planner.Plan(fctx, tTaskID, e.m.assets, t.Store, t.Goal, t.drainDone(), emit) + met, reason, err := planner.Plan(fctx, tTaskID, e.m.assets, t.Store, t.Goal, t.drainTriggers(), emit) e.decInflight(t.ID) if err != nil { log.Printf("[deadline] task %s 终局规划出错: %v", t.ID, err) diff --git a/server/manager.go b/server/manager.go index 98a7169..e37b73b 100644 --- a/server/manager.go +++ b/server/manager.go @@ -44,11 +44,11 @@ type Task struct { Guard *guard.Guard `json:"-"` notify chan struct{} - // doneIntents accumulates the ids of intents that completed since the last - // planning round consumed them. The debounce coalesces a burst of completions - // into one round, so several ids may pile up before drainDone() clears them. - doneMu sync.Mutex - doneIntents []int64 + // pendingTriggers accumulates the concrete changes (worker done / finding) that + // fired planning rounds since the last one consumed them. The debounce coalesces + // a burst into one round, so several may pile up before drainTriggers() clears them. + trigMu sync.Mutex + pendingTriggers []agent.TriggerEvent } // Manager owns the PostgreSQL data source (asset graph + every task's exploration @@ -696,24 +696,33 @@ func (t *Task) Notify() { } } -// NotifyDone is Notify plus a hint: intentID just finished and is what triggered -// this wake-up. The planner reads the accumulated ids next round so it knows -// which intents' fresh yields to focus on. Ids pile up (debounce) until the -// round drains them via drainDone. +// NotifyDone is Notify plus a hint: a worker just finished intentID and that is +// what triggered this wake-up. The planner reads the accumulated triggers next +// round so it can spell out which intent finished (+ its output). Events pile up +// (debounce) until the round drains them via drainTriggers. func (t *Task) NotifyDone(intentID int64) { if intentID > 0 { - t.doneMu.Lock() - t.doneIntents = append(t.doneIntents, intentID) - t.doneMu.Unlock() + t.trigMu.Lock() + t.pendingTriggers = append(t.pendingTriggers, agent.TriggerEvent{Kind: "done", IntentID: intentID}) + t.trigMu.Unlock() } t.Notify() } -// drainDone returns and clears the intent ids completed since the last round. -func (t *Task) drainDone() []int64 { - t.doneMu.Lock() - defer t.doneMu.Unlock() - ids := t.doneIntents - t.doneIntents = nil - return ids +// NotifyFinding records that a worker reported a finding on intentID (summary), +// then wakes the planner — so the round spells out which intent found what. +func (t *Task) NotifyFinding(intentID int64, summary string) { + t.trigMu.Lock() + t.pendingTriggers = append(t.pendingTriggers, agent.TriggerEvent{Kind: "finding", IntentID: intentID, Detail: summary}) + t.trigMu.Unlock() + t.Notify() +} + +// drainTriggers returns and clears the trigger events accumulated since the last round. +func (t *Task) drainTriggers() []agent.TriggerEvent { + t.trigMu.Lock() + defer t.trigMu.Unlock() + ev := t.pendingTriggers + t.pendingTriggers = nil + return ev }