diff --git a/server/conversations.go b/server/conversations.go index 588d9d0..682695b 100644 --- a/server/conversations.go +++ b/server/conversations.go @@ -598,14 +598,14 @@ func (s *Server) readTriggerBehavior(agentKey string) triggerBehavior { // agent's策略 decides concurrency + merge: serial → run one at a time (optionally // merging by task / all / none); parallel → run each fire in its own concurrent // conversation up to trigger_max_parallel. Distinct agents always run concurrently. -func (s *Server) StartTriggeredRun(agentKey, title, message string, taskID int64, mergeable bool) { +func (s *Server) StartTriggeredRun(agentKey, title, message string, taskID int64, mergeable bool, taskDesc, taskGoal string) { if s.m.pg == nil || s.chatAgentRef() == nil { return } cfg := s.readTriggerBehavior(agentKey) // DB read BEFORE the lock (never under queueMu) s.queueMu.Lock() s.triggerCfg[agentKey] = cfg - s.triggerQ[agentKey] = append(s.triggerQ[agentKey], triggeredRun{agentKey: agentKey, title: title, message: message, taskID: taskID, mergeable: mergeable}) + s.triggerQ[agentKey] = append(s.triggerQ[agentKey], triggeredRun{agentKey: agentKey, title: title, message: message, taskID: taskID, taskDesc: taskDesc, taskGoal: taskGoal, mergeable: mergeable}) s.pumpLocked(agentKey) s.queueMu.Unlock() } @@ -686,8 +686,38 @@ func (s *Server) nextTriggerRun(agentKey string, cfg triggerBehavior) triggeredR return mergeTriggeredRuns(group) } -// mergeTriggeredRuns folds several same-task event fires into one run: a header plus -// each fire's message, so the agent handles the task's burst in a single conversation. +// taskContextHeader renders a task's description/goal once. Same-task fires share +// this block, so the scheduler no longer repeats it per event (a long task goal +// times N fires was the dominant bloat). Returns "" for interval/none triggers +// (taskID==0, no task context). desc/goal are truncated to keep even a single copy +// bounded. +func taskContextHeader(taskID int64, desc, goal string) string { + if desc == "" && goal == "" { + // No task context to show (interval/none fire, or a merged run that already + // embedded its per-task headers and cleared these fields). + return "" + } + if goal != "" { + return fmt.Sprintf("【任务 #%d %s（目标：%s）】", taskID, trunc(desc, 200), trunc(goal, 500)) + } + return fmt.Sprintf("【任务 #%d %s】", taskID, trunc(desc, 200)) +} + +// finalTriggerMessage renders the message actually sent to the agent for a single +// (non-merged) fire: the task-context header (written once) followed by the event +// body. Merged runs embed their per-task headers inline and clear taskDesc/taskGoal, +// so this returns their message unchanged. +func finalTriggerMessage(item triggeredRun) string { + if h := taskContextHeader(item.taskID, item.taskDesc, item.taskGoal); h != "" { + return h + "\n" + item.message + } + return item.message +} + +// mergeTriggeredRuns folds several same-task event fires into one run: the shared +// task-context header written ONCE, then each fire's event body, so the agent handles +// the task's burst in a single conversation without repeating the (possibly long) +// task description/goal per event. func mergeTriggeredRuns(items []triggeredRun) triggeredRun { if len(items) == 1 { return items[0] @@ -695,6 +725,9 @@ func mergeTriggeredRuns(items []triggeredRun) triggeredRun { first := items[0] var b strings.Builder fmt.Fprintf(&b, "【本会话合并了任务 #%d 的 %d 条触发事件，请一并处理】\n", first.taskID, len(items)) + if h := taskContextHeader(first.taskID, first.taskDesc, first.taskGoal); h != "" { + fmt.Fprintf(&b, "%s\n", h) // same task → task context appears once + } for i, it := range items { fmt.Fprintf(&b, "\n── 触发 %d ──\n%s\n", i+1, it.message) } @@ -704,20 +737,41 @@ func mergeTriggeredRuns(items []triggeredRun) triggeredRun { message: b.String(), taskID: first.taskID, mergeable: true, + // taskDesc/taskGoal left empty: header already embedded above. } } // mergeAllRuns folds the ENTIRE queued burst into one run (merge_mode='all'), -// regardless of task or type — a header plus each fire's message. +// regardless of type. Events are grouped by task (tasks in first-appearance order, +// events in their original order within a task), so each task's context header — and +// its possibly long description/goal — is written exactly ONCE even when fires from +// different tasks interleave in the queue. Cross-task chronology is not preserved +// (event bodies carry no timestamps, so per-task grouping reads better for the agent). func mergeAllRuns(items []triggeredRun) triggeredRun { if len(items) == 1 { return items[0] } first := items[0] + order := []int64{} + groups := map[int64][]triggeredRun{} + for _, it := range items { + if _, seen := groups[it.taskID]; !seen { + order = append(order, it.taskID) + } + groups[it.taskID] = append(groups[it.taskID], it) + } var b strings.Builder - fmt.Fprintf(&b, "【本会话合并了队列中的 %d 条触发事件（不区分任务），请一并处理】\n", len(items)) - for i, it := range items { - fmt.Fprintf(&b, "\n── 触发 %d（task#%d）──\n%s\n", i+1, it.taskID, it.message) + fmt.Fprintf(&b, "【本会话合并了队列中的 %d 条触发事件（共 %d 个任务），请一并处理】\n", len(items), len(order)) + seq := 0 + for _, tid := range order { + g := groups[tid] + if h := taskContextHeader(tid, g[0].taskDesc, g[0].taskGoal); h != "" { + fmt.Fprintf(&b, "\n%s\n", h) // same task → context appears once, even if interleaved + } + for _, it := range g { + seq++ + fmt.Fprintf(&b, "\n── 触发 %d（task#%d）──\n%s\n", seq, tid, it.message) + } } return triggeredRun{ agentKey: first.agentKey, @@ -725,6 +779,7 @@ func mergeAllRuns(items []triggeredRun) triggeredRun { message: b.String(), taskID: first.taskID, mergeable: true, + // taskDesc/taskGoal left empty: per-task headers already embedded above. } } @@ -743,14 +798,15 @@ func (s *Server) runTriggeredRun(item triggeredRun) { log.Printf("[trigger] create conversation for %s failed: %v", item.agentKey, err) return } - if _, err := pg.AppendConvActivity(c.ID, db.Activity{Worker: item.agentKey, Kind: "user", Summary: firstLine(item.message, 200), Detail: item.message}); err != nil { + msg := finalTriggerMessage(item) // prepend the task-context header for single (non-merged) fires + if _, err := pg.AppendConvActivity(c.ID, db.Activity{Worker: item.agentKey, Kind: "user", Summary: firstLine(msg, 200), Detail: msg}); err != nil { log.Printf("[trigger] append msg failed: %v", err) } busyKey := s.convBusyKey(c.ID) s.chatMu.Lock() s.chatBusy[busyKey] = true s.chatMu.Unlock() - s.runConversationSync(c, item.message, busyKey) + s.runConversationSync(c, msg, busyKey) } // firstLine returns a single-line, length-capped preview (shared with summaries). diff --git a/server/scheduler.go b/server/scheduler.go index 0c6b1e8..519ed80 100644 --- a/server/scheduler.go +++ b/server/scheduler.go @@ -143,7 +143,7 @@ func (sc *Scheduler) fireIntervals(triggers []*db.AgentTrigger) { } _ = sc.pg.TouchTriggerFire(tr.ID) ctx := "\n\n【本次为定时触发】" + now.Format(" 2006-01-02 15:04:05 MST") - sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("定时触发 · %s", now.Format("15:04")), tr.IntervalMessage+ctx, 0, false) + sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("定时触发 · %s", now.Format("15:04")), tr.IntervalMessage+ctx, 0, false, "", "") } } @@ -172,10 +172,10 @@ func (sc *Scheduler) fireFindings(triggers []*db.AgentTrigger) { if len(want) == 0 { continue } - msgCtx := fmt.Sprintf("\n\n【本次由任务发现 finding 触发】\n任务: #%d %s（目标：%s）\n发现: [%s/%s] %s", - e.TaskID, e.TaskDesc, e.TaskGoal, e.VulnClass, e.Severity, e.Summary) + msgCtx := fmt.Sprintf("\n\n【本次由任务发现 finding 触发】\n发现: [%s/%s] %s", + e.VulnClass, e.Severity, e.Summary) for _, tr := range want { - sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("finding 触发 · task#%d", e.TaskID), tr.FindingMessage+msgCtx, e.TaskID, true) + sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("finding 触发 · task#%d", e.TaskID), tr.FindingMessage+msgCtx, e.TaskID, true, e.TaskDesc, e.TaskGoal) } } _ = sc.pg.SetSchedState(schedKeyLastFinding, strconv.FormatInt(maxID, 10)) @@ -206,10 +206,9 @@ func (sc *Scheduler) fireGoals(triggers []*db.AgentTrigger) { if len(want) == 0 { continue } - msgCtx := fmt.Sprintf("\n\n【本次由任务完成目标触发】\n任务: #%d %s（目标：%s）\n达成目标: %s", - e.TaskID, e.TaskDesc, e.TaskGoal, e.Summary) + msgCtx := fmt.Sprintf("\n\n【本次由任务完成目标触发】\n达成目标: %s", e.Summary) for _, tr := range want { - sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("目标触发 · task#%d", e.TaskID), tr.GoalMessage+msgCtx, e.TaskID, true) + sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("目标触发 · task#%d", e.TaskID), tr.GoalMessage+msgCtx, e.TaskID, true, e.TaskDesc, e.TaskGoal) } } if changed { @@ -240,10 +239,9 @@ func (sc *Scheduler) fireTaskTimeouts(triggers []*db.AgentTrigger) { if len(want) == 0 { continue } - msgCtx := fmt.Sprintf("\n\n【本次由任务超时触发】\n任务: #%d %s（目标：%s）", - e.TaskID, e.TaskDesc, e.TaskGoal) + msgCtx := "\n\n【本次由任务超时触发】" for _, tr := range want { - sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("超时触发 · task#%d", e.TaskID), tr.TaskTimeoutMessage+msgCtx, e.TaskID, true) + sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("超时触发 · task#%d", e.TaskID), tr.TaskTimeoutMessage+msgCtx, e.TaskID, true, e.TaskDesc, e.TaskGoal) } } _ = sc.pg.SetSchedState(schedKeyLastTimeout, strconv.FormatInt(maxID, 10)) @@ -272,10 +270,9 @@ func (sc *Scheduler) fireTaskCreates(triggers []*db.AgentTrigger) { if len(want) == 0 { continue } - msgCtx := fmt.Sprintf("\n\n【本次由任务创建触发】\n任务ID: #%d\n任务描述: %s\n任务目标: %s", - e.TaskID, e.TaskDesc, e.TaskGoal) + msgCtx := "\n\n【本次由任务创建触发】" for _, tr := range want { - sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("任务创建触发 · task#%d", e.TaskID), tr.TaskCreateMessage+msgCtx, e.TaskID, true) + sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("任务创建触发 · task#%d", e.TaskID), tr.TaskCreateMessage+msgCtx, e.TaskID, true, e.TaskDesc, e.TaskGoal) } } _ = sc.pg.SetSchedState(schedKeyLastTaskCreate, strconv.FormatInt(maxID, 10)) @@ -314,13 +311,13 @@ func (sc *Scheduler) fireToolCalls(triggers []*db.AgentTrigger) { if e.ToolIsErr { errTag = "[error] " } - msgCtx := fmt.Sprintf("\n\n【本次由工具调用触发】\n任务: #%d %s（目标：%s）\n工具: %s\n入参: %s\n返回: %s%s", - e.TaskID, e.TaskDesc, e.TaskGoal, e.Tool, trunc(e.ToolInput, 1500), errTag, trunc(e.ToolOutput, 1500)) + msgCtx := fmt.Sprintf("\n\n【本次由工具调用触发】\n工具: %s\n入参: %s\n返回: %s%s", + e.Tool, trunc(e.ToolInput, 1500), errTag, trunc(e.ToolOutput, 1500)) for _, tr := range want { if !containsFold(tr.ToolNames, e.Tool) { continue } - sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("工具触发 · %s · task#%d", e.Tool, e.TaskID), tr.ToolCallMessage+msgCtx, e.TaskID, true) + sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("工具触发 · %s · task#%d", e.Tool, e.TaskID), tr.ToolCallMessage+msgCtx, e.TaskID, true, e.TaskDesc, e.TaskGoal) } } _ = sc.pg.SetSchedState(schedKeyLastToolCall, strconv.FormatInt(maxID, 10)) diff --git a/server/server.go b/server/server.go index 011d68a..9a867d5 100644 --- a/server/server.go +++ b/server/server.go @@ -142,9 +142,11 @@ type provEntry struct { type triggeredRun struct { agentKey string title string - message string - taskID int64 // source task for finding/goal triggers; 0 for interval/none - mergeable bool // true for finding/goal event triggers (merge by taskID) + message string // 事件正文(触发语 + 工具/入参/返回等);不含任务描述/目标头 + taskID int64 // source task for finding/goal triggers; 0 for interval/none + taskDesc string // 任务描述(任务级,同任务相同);合并时只渲染一次 + taskGoal string // 任务目标(任务级,同任务相同);合并时只渲染一次 + mergeable bool // true for finding/goal event triggers (merge by taskID) } func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDir string) *Server { diff --git a/server/trigger_merge_test.go b/server/trigger_merge_test.go new file mode 100644 index 0000000..cf7fdff --- /dev/null +++ b/server/trigger_merge_test.go @@ -0,0 +1,98 @@ +package server + +import ( + "strings" + "testing" +) + +// A long task goal repeated per event was the dominant bloat. These tests pin the +// fix: the task-context header (description + goal) is rendered ONCE per task, no +// matter how many same-task fires are merged. + +const longGoal = "拿到题目 f2-05 的受保护 flag 并通过 submit_flag 提交；本题密文已高度收敛，flag 只能由二进制内嵌数据派生……" // 代表那段几千字的继承事实 + +func sameTaskFires(n int) []triggeredRun { + items := make([]triggeredRun, n) + for i := range items { + items[i] = triggeredRun{ + agentKey: "tec_benchmark", taskID: 72, taskDesc: "f2-05 逆向", taskGoal: longGoal, + message: "【本次由工具调用触发】\n工具: submit_flag\n入参: {...}\n返回: {correct:false}", mergeable: true, + } + } + return items +} + +func TestMergeAllRunsWritesTaskGoalOnce(t *testing.T) { + out := mergeAllRuns(sameTaskFires(39)) + if got := strings.Count(out.message, longGoal); got != 1 { + t.Fatalf("same-task goal should appear exactly once in a merged-all run, got %d", got) + } + if strings.Count(out.message, "── 触发 ") < 1 || !strings.Contains(out.message, "触发 39") { + t.Fatalf("all 39 event bodies should be present: %q", out.message) + } + // A merged run embeds its header inline, so finalTriggerMessage must not re-add it. + if out.taskDesc != "" || out.taskGoal != "" { + t.Fatalf("merged run must clear taskDesc/taskGoal to avoid a duplicate header") + } + if finalTriggerMessage(out) != out.message { + t.Fatalf("finalTriggerMessage must not prepend another header for a merged run") + } +} + +func TestMergeAllRunsGroupsInterleavedTasks(t *testing.T) { + // Fires from two tasks arriving interleaved (A,B,A,B) must still carry each + // task's context exactly once — grouping, not per-event repetition. + mk := func(id int64, goal string) triggeredRun { + return triggeredRun{agentKey: "a", taskID: id, taskDesc: "d", taskGoal: goal, message: "body", mergeable: true} + } + out := mergeAllRuns([]triggeredRun{mk(1, "GOAL_A"), mk(2, "GOAL_B"), mk(1, "GOAL_A"), mk(2, "GOAL_B")}) + if got := strings.Count(out.message, "GOAL_A"); got != 1 { + t.Fatalf("task #1 goal should appear once despite interleaving, got %d", got) + } + if got := strings.Count(out.message, "GOAL_B"); got != 1 { + t.Fatalf("task #2 goal should appear once despite interleaving, got %d", got) + } + if !strings.Contains(out.message, "共 2 个任务") { + t.Fatalf("header should report 2 tasks: %q", out.message) + } + if got := strings.Count(out.message, "── 触发 "); got != 4 { + t.Fatalf("all 4 event bodies should be present, got %d", got) + } +} + +func TestMergeTriggeredRunsWritesTaskGoalOnce(t *testing.T) { + out := mergeTriggeredRuns(sameTaskFires(5)) + if got := strings.Count(out.message, longGoal); got != 1 { + t.Fatalf("same-task goal should appear exactly once in a by-task merge, got %d", got) + } +} + +func TestFinalTriggerMessageSingleFirePrependsHeaderOnce(t *testing.T) { + item := sameTaskFires(1)[0] + msg := finalTriggerMessage(item) + if got := strings.Count(msg, longGoal); got != 1 { + t.Fatalf("single fire should carry the task goal exactly once, got %d", got) + } + if !strings.HasPrefix(msg, "【任务 #72") { + t.Fatalf("single fire should be prefixed with the task-context header: %q", msg) + } +} + +func TestTaskContextHeaderEmptyForIntervalFire(t *testing.T) { + if h := taskContextHeader(0, "", ""); h != "" { + t.Fatalf("interval/none trigger (no task) must produce no header, got %q", h) + } + // An interval fire's message must pass through untouched. + item := triggeredRun{message: "定时触发正文"} + if finalTriggerMessage(item) != "定时触发正文" { + t.Fatalf("interval fire message must pass through unchanged") + } +} + +func TestTaskContextHeaderTruncatesLongGoal(t *testing.T) { + huge := strings.Repeat("很", 5000) + h := taskContextHeader(72, "d", huge) + if len([]rune(h)) > 800 { // 200 desc + 500 goal + 截断标记/装饰，远小于 5000 + t.Fatalf("header should be bounded even for a huge goal, got %d runes", len([]rune(h))) + } +}