feat(trigger): 新增"任务创建"触发机制(on_task_create)

自定义 agent 触发条件从 5 种扩展到 6 种,新增任务被创建时触发。用 tasks.id 单调水位线 + scheduler_state 防重复,重启不补触发; 触发消息完整分行给出任务ID/描述/目标。DB 加列并带 ALTER 迁移。 Co-Authored-By: Claude Opus 4.8 <[email protected]>
2026-08-09 02:19:29 -04:00
1 parent 5962826f2d
commit 4cf622593a
@@ -26,10 +26,11 @@ type Scheduler struct {
 }

 const (
-	schedKeyLastFinding  = "last_finding_id"  // watermark: max finding node id fired for
-	schedKeyFiredGoals   = "fired_goals"      // JSON array of goal node ids already fired
-	schedKeyLastTimeout  = "last_timeout_id"  // watermark: max task id fired for task timeout
-	schedKeyLastToolCall = "last_toolcall_id" // watermark: max activity id fired for tool call
+	schedKeyLastFinding    = "last_finding_id"    // watermark: max finding node id fired for
+	schedKeyFiredGoals     = "fired_goals"        // JSON array of goal node ids already fired
+	schedKeyLastTimeout    = "last_timeout_id"    // watermark: max task id fired for task timeout
+	schedKeyLastToolCall   = "last_toolcall_id"   // watermark: max activity id fired for tool call
+	schedKeyLastTaskCreate = "last_taskcreate_id" // watermark: max task id fired for task create
 )

 func newScheduler(s *Server) *Scheduler {
@@ -99,6 +100,17 @@ func (sc *Scheduler) init() {
 		}
 		_ = sc.pg.SetSchedState(schedKeyLastToolCall, strconv.FormatInt(maxID, 10))
 	}
+	if sc.mustState(schedKeyLastTaskCreate) == "" {
+		var maxID int64
+		if evs, err := sc.pg.NewTasksSince(0); err == nil {
+			for _, e := range evs {
+				if e.NodeID > maxID {
+					maxID = e.NodeID
+				}
+			}
+		}
+		_ = sc.pg.SetSchedState(schedKeyLastTaskCreate, strconv.FormatInt(maxID, 10))
+	}
 }

 func (sc *Scheduler) step() {
@@ -114,6 +126,7 @@ func (sc *Scheduler) step() {
 	sc.fireGoals(triggers)
 	sc.fireTaskTimeouts(triggers)
 	sc.fireToolCalls(triggers)
+	sc.fireTaskCreates(triggers)
 }

 // fireIntervals fires triggers whose interval has elapsed since last_fire.
@@ -229,6 +242,37 @@ func (sc *Scheduler) fireTaskTimeouts(triggers []*db.AgentTrigger) {
 	_ = sc.pg.SetSchedState(schedKeyLastTimeout, strconv.FormatInt(maxID, 10))
 }

+// fireTaskCreates fires on_task_create triggers for tasks newly created above the
+// persisted watermark (task id → no double-fire across restarts).
+func (sc *Scheduler) fireTaskCreates(triggers []*db.AgentTrigger) {
+	var want []*db.AgentTrigger
+	for _, tr := range triggers {
+		if tr.OnTaskCreate {
+			want = append(want, tr)
+		}
+	}
+	if len(want) == 0 {
+		return
+	}
+	last, _ := strconv.ParseInt(sc.mustState(schedKeyLastTaskCreate), 10, 64)
+	events, err := sc.pg.NewTasksSince(last)
+	if err != nil || len(events) == 0 {
+		return
+	}
+	maxID := last
+	for _, e := range events {
+		if e.NodeID > maxID {
+			maxID = e.NodeID
+		}
+		msgCtx := fmt.Sprintf("\n\n【本次由任务创建触发】\n任务ID: #%d\n任务描述: %s\n任务目标: %s",
+			e.TaskID, e.TaskDesc, e.TaskGoal)
+		for _, tr := range want {
+			sc.s.StartTriggeredRun(tr.AgentKey, fmt.Sprintf("任务创建触发 · task#%d", e.TaskID), tr.TaskCreateMessage+msgCtx, e.TaskID, true)
+		}
+	}
+	_ = sc.pg.SetSchedState(schedKeyLastTaskCreate, strconv.FormatInt(maxID, 10))
+}
+
 // fireToolCalls fires on_tool_call triggers for tool calls (tool_result rows) above
 // the persisted watermark whose tool name is in the trigger's selected set. The fire
 // message carries the task id/desc/goal + tool name + (truncated) input & output.
@@ -28,19 +28,21 @@ type triggerReq struct {
 	OnGoalMet          bool     `json:"on_goal_met"`
 	OnTaskTimeout      bool     `json:"on_task_timeout"`
 	OnToolCall         bool     `json:"on_tool_call"`
+	OnTaskCreate       bool     `json:"on_task_create"`
 	IntervalMessage    string   `json:"interval_message"`
 	FindingMessage     string   `json:"finding_message"`
 	GoalMessage        string   `json:"goal_message"`
 	TaskTimeoutMessage string   `json:"task_timeout_message"`
 	ToolCallMessage    string   `json:"tool_call_message"`
+	TaskCreateMessage  string   `json:"task_create_message"`
 	ToolNames          []string `json:"tool_names"`
 }

 // validateTrigger enforces the shared trigger rules for create/update:
 // at least one condition, and on_tool_call requires a non-empty tool set.
 func validateTrigger(req *triggerReq) string {
-	if req.IntervalSec == 0 && !req.OnFinding && !req.OnGoalMet && !req.OnTaskTimeout && !req.OnToolCall {
-		return "至少选择一种触发条件(定时/发现finding/目标达成/任务超时/工具调用)"
+	if req.IntervalSec == 0 && !req.OnFinding && !req.OnGoalMet && !req.OnTaskTimeout && !req.OnToolCall && !req.OnTaskCreate {
+		return "至少选择一种触发条件(定时/发现finding/目标达成/任务超时/工具调用/任务创建)"
 	}
 	if req.OnToolCall && len(req.ToolNames) == 0 {
 		return "工具调用触发至少选择一个工具"
@@ -71,10 +73,10 @@ func (s *Server) pgCreateTrigger(w http.ResponseWriter, r *http.Request) {
 	}
 	tr, err := pg.CreateTrigger(&db.AgentTrigger{
 		AgentKey: a.Key, Enabled: req.Enabled, IntervalSec: req.IntervalSec,
-		OnFinding: req.OnFinding, OnGoalMet: req.OnGoalMet, OnTaskTimeout: req.OnTaskTimeout, OnToolCall: req.OnToolCall,
+		OnFinding: req.OnFinding, OnGoalMet: req.OnGoalMet, OnTaskTimeout: req.OnTaskTimeout, OnToolCall: req.OnToolCall, OnTaskCreate: req.OnTaskCreate,
 		IntervalMessage: req.IntervalMessage, FindingMessage: req.FindingMessage,
 		GoalMessage: req.GoalMessage, TaskTimeoutMessage: req.TaskTimeoutMessage,
-		ToolCallMessage: req.ToolCallMessage, ToolNames: req.ToolNames,
+		ToolCallMessage: req.ToolCallMessage, TaskCreateMessage: req.TaskCreateMessage, ToolNames: req.ToolNames,
 	})
 	if err != nil {
 		writeErr(w, 500, err.Error())
@@ -107,10 +109,10 @@ func (s *Server) pgUpdateTrigger(w http.ResponseWriter, r *http.Request) {
 	}
 	if err := pg.UpdateTrigger(&db.AgentTrigger{
 		ID: id, Enabled: req.Enabled, IntervalSec: req.IntervalSec,
-		OnFinding: req.OnFinding, OnGoalMet: req.OnGoalMet, OnTaskTimeout: req.OnTaskTimeout, OnToolCall: req.OnToolCall,
+		OnFinding: req.OnFinding, OnGoalMet: req.OnGoalMet, OnTaskTimeout: req.OnTaskTimeout, OnToolCall: req.OnToolCall, OnTaskCreate: req.OnTaskCreate,
 		IntervalMessage: req.IntervalMessage, FindingMessage: req.FindingMessage,
 		GoalMessage: req.GoalMessage, TaskTimeoutMessage: req.TaskTimeoutMessage,
-		ToolCallMessage: req.ToolCallMessage, ToolNames: req.ToolNames,
+		ToolCallMessage: req.ToolCallMessage, TaskCreateMessage: req.TaskCreateMessage, ToolNames: req.ToolNames,
 	}); err != nil {
 		writeErr(w, 500, err.Error())
 		return