fix(trigger): 合并触发会话按任务去重任务描述/目标,避免长目标重复撑爆消息

on_tool_call 等事件触发在 merge 时,每条事件都重复整段任务描述+目标;
当任务目标很长(如续跑继承事实)且短时间多次触发时,合并消息体积爆炸
(实测 39 条 submit_flag 连击 → 十几万字)。

- triggeredRun 拆出 taskDesc/taskGoal 任务级字段,message 只留事件正文
- by_task 合并:同任务头只写一次
- all 合并:按任务分组(首次出现排序、任务内保序),每任务头只写一次,
  交错触发也不重复
- 单条触发由 finalTriggerMessage 补一次任务头
- 任务描述截 200 字、目标截 500 字,单份也有界
- 新增 trigger_merge_test.go 覆盖去重/交错/截断/定时无头
This commit is contained in:
Autumn-27 committed 2026-09-12 09:26:38 -04:00
1 parent e4c744a318
commit 44194a9ebc
4 files changed
+182 -29

No files matched your search

+66 -10
View File
@@ -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).
+13 -16
View File
@@ -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))
+5 -3
View File
@@ -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 {
+98
View File
@@ -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)))
}
}