diff --git a/db/schema.sql b/db/schema.sql index 2f70e13..6944971 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -457,11 +457,13 @@ CREATE TABLE IF NOT EXISTS agent_triggers ( on_goal_met BOOLEAN NOT NULL DEFAULT false, on_task_timeout BOOLEAN NOT NULL DEFAULT false, on_tool_call BOOLEAN NOT NULL DEFAULT false, + on_task_create BOOLEAN NOT NULL DEFAULT false, interval_message TEXT NOT NULL DEFAULT '', finding_message TEXT NOT NULL DEFAULT '', goal_message TEXT NOT NULL DEFAULT '', task_timeout_message TEXT NOT NULL DEFAULT '', tool_call_message TEXT NOT NULL DEFAULT '', + task_create_message TEXT NOT NULL DEFAULT '', tool_names TEXT NOT NULL DEFAULT '', last_fire TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), @@ -469,9 +471,11 @@ CREATE TABLE IF NOT EXISTS agent_triggers ( ); CREATE INDEX IF NOT EXISTS idx_agent_triggers_agent ON agent_triggers(agent_key); -- 加列迁移(已发版,旧库升级补列;新库 CREATE 已含这些列,ALTER 为 no-op)。幂等,每次启动可重复执行。 -ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS on_tool_call BOOLEAN NOT NULL DEFAULT false; -ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS tool_call_message TEXT NOT NULL DEFAULT ''; -ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS tool_names TEXT NOT NULL DEFAULT ''; +ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS on_tool_call BOOLEAN NOT NULL DEFAULT false; +ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS tool_call_message TEXT NOT NULL DEFAULT ''; +ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS tool_names TEXT NOT NULL DEFAULT ''; +ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS on_task_create BOOLEAN NOT NULL DEFAULT false; +ALTER TABLE agent_triggers ADD COLUMN IF NOT EXISTS task_create_message TEXT NOT NULL DEFAULT ''; DROP TRIGGER IF EXISTS trg_agent_triggers_upd ON agent_triggers; CREATE TRIGGER trg_agent_triggers_upd BEFORE UPDATE ON agent_triggers FOR EACH ROW EXECUTE FUNCTION set_updated_at(); diff --git a/db/triggers.go b/db/triggers.go index eb0cd04..c93e7a5 100644 --- a/db/triggers.go +++ b/db/triggers.go @@ -6,10 +6,11 @@ import ( "time" ) -// AgentTrigger is one P3 trigger attached to a custom agent. Five trigger +// AgentTrigger is one P3 trigger attached to a custom agent. Six trigger // conditions can be on at once: interval / on_finding / on_goal_met / -// on_task_timeout / on_tool_call. ToolNames scopes the tool-call trigger to a -// non-empty set of tool keys (empty is rejected at the API layer for on_tool_call). +// on_task_timeout / on_tool_call / on_task_create. ToolNames scopes the tool-call +// trigger to a non-empty set of tool keys (empty is rejected at the API layer for +// on_tool_call). type AgentTrigger struct { ID int64 `json:"id"` AgentKey string `json:"agent_key"` @@ -19,16 +20,18 @@ type AgentTrigger 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"` // 选中的工具 key 列表(DB 存 JSON 文本) LastFire *time.Time `json:"last_fire,omitempty"` } -const triggerCols = `id, agent_key, enabled, interval_sec, on_finding, on_goal_met, on_task_timeout, on_tool_call, interval_message, finding_message, goal_message, task_timeout_message, tool_call_message, tool_names, last_fire` +const triggerCols = `id, agent_key, enabled, interval_sec, on_finding, on_goal_met, on_task_timeout, on_tool_call, on_task_create, interval_message, finding_message, goal_message, task_timeout_message, tool_call_message, task_create_message, tool_names, last_fire` // marshalToolNames encodes the tool-key list as JSON text for the tool_names column. // A nil/empty list stores "" (not "null"/"[]") so the column default stays clean. @@ -47,8 +50,8 @@ func scanTrigger(sc interface{ Scan(...any) error }) (*AgentTrigger, error) { var t AgentTrigger var lf sql.NullTime var toolNames string - if err := sc.Scan(&t.ID, &t.AgentKey, &t.Enabled, &t.IntervalSec, &t.OnFinding, &t.OnGoalMet, &t.OnTaskTimeout, &t.OnToolCall, - &t.IntervalMessage, &t.FindingMessage, &t.GoalMessage, &t.TaskTimeoutMessage, &t.ToolCallMessage, &toolNames, &lf); err != nil { + if err := sc.Scan(&t.ID, &t.AgentKey, &t.Enabled, &t.IntervalSec, &t.OnFinding, &t.OnGoalMet, &t.OnTaskTimeout, &t.OnToolCall, &t.OnTaskCreate, + &t.IntervalMessage, &t.FindingMessage, &t.GoalMessage, &t.TaskTimeoutMessage, &t.ToolCallMessage, &t.TaskCreateMessage, &toolNames, &lf); err != nil { return nil, err } t.ToolNames = []string{} @@ -64,18 +67,18 @@ func scanTrigger(sc interface{ Scan(...any) error }) (*AgentTrigger, error) { // CreateTrigger inserts a trigger for agentKey and returns it. func (d *DB) CreateTrigger(t *AgentTrigger) (*AgentTrigger, error) { row := d.QueryRow(` -INSERT INTO agent_triggers(agent_key, enabled, interval_sec, on_finding, on_goal_met, on_task_timeout, on_tool_call, interval_message, finding_message, goal_message, task_timeout_message, tool_call_message, tool_names) -VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) RETURNING `+triggerCols, - t.AgentKey, t.Enabled, t.IntervalSec, t.OnFinding, t.OnGoalMet, t.OnTaskTimeout, t.OnToolCall, - t.IntervalMessage, t.FindingMessage, t.GoalMessage, t.TaskTimeoutMessage, t.ToolCallMessage, marshalToolNames(t.ToolNames)) +INSERT INTO agent_triggers(agent_key, enabled, interval_sec, on_finding, on_goal_met, on_task_timeout, on_tool_call, on_task_create, interval_message, finding_message, goal_message, task_timeout_message, tool_call_message, task_create_message, tool_names) +VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) RETURNING `+triggerCols, + t.AgentKey, t.Enabled, t.IntervalSec, t.OnFinding, t.OnGoalMet, t.OnTaskTimeout, t.OnToolCall, t.OnTaskCreate, + t.IntervalMessage, t.FindingMessage, t.GoalMessage, t.TaskTimeoutMessage, t.ToolCallMessage, t.TaskCreateMessage, marshalToolNames(t.ToolNames)) return scanTrigger(row) } // UpdateTrigger updates a trigger's fields (not last_fire). func (d *DB) UpdateTrigger(t *AgentTrigger) error { - _, err := d.Exec(`UPDATE agent_triggers SET enabled=$2, interval_sec=$3, on_finding=$4, on_goal_met=$5, on_task_timeout=$6, on_tool_call=$7, interval_message=$8, finding_message=$9, goal_message=$10, task_timeout_message=$11, tool_call_message=$12, tool_names=$13 WHERE id=$1`, - t.ID, t.Enabled, t.IntervalSec, t.OnFinding, t.OnGoalMet, t.OnTaskTimeout, t.OnToolCall, - t.IntervalMessage, t.FindingMessage, t.GoalMessage, t.TaskTimeoutMessage, t.ToolCallMessage, marshalToolNames(t.ToolNames)) + _, err := d.Exec(`UPDATE agent_triggers SET enabled=$2, interval_sec=$3, on_finding=$4, on_goal_met=$5, on_task_timeout=$6, on_tool_call=$7, on_task_create=$8, interval_message=$9, finding_message=$10, goal_message=$11, task_timeout_message=$12, tool_call_message=$13, task_create_message=$14, tool_names=$15 WHERE id=$1`, + t.ID, t.Enabled, t.IntervalSec, t.OnFinding, t.OnGoalMet, t.OnTaskTimeout, t.OnToolCall, t.OnTaskCreate, + t.IntervalMessage, t.FindingMessage, t.GoalMessage, t.TaskTimeoutMessage, t.ToolCallMessage, t.TaskCreateMessage, marshalToolNames(t.ToolNames)) return err } @@ -210,6 +213,31 @@ ORDER BY id`, lastID) return out, rows.Err() } +// NewTasksSince returns tasks created with id > lastID (excluding deleted), +// ordered by id (monotonic watermark → no double-fire across restarts). Triggered +// agent runs are conversations, not tasks, so this never fires on its own output. +func (d *DB) NewTasksSince(lastID int64) ([]TaskEvent, error) { + rows, err := d.Query(` +SELECT id, description, goal FROM tasks +WHERE deleted_at IS NULL AND id > $1 +ORDER BY id`, lastID) + if err != nil { + return nil, err + } + defer rows.Close() + out := []TaskEvent{} + for rows.Next() { + var e TaskEvent + if err := rows.Scan(&e.NodeID, &e.TaskDesc, &e.TaskGoal); err != nil { + return nil, err + } + e.TaskID = e.NodeID // task id doubles as NodeID for the watermark + e.Summary = e.TaskGoal + out = append(out, e) + } + return out, rows.Err() +} + // NewToolCallsSince returns completed tool calls (a tool_result row) with activity // id > lastID across all live tasks, ordered by id (monotonic watermark → no // double-fire). It is driven by tool_result rows (the tool finished, so both input diff --git a/server/scheduler.go b/server/scheduler.go index 3f83063..1aeeb17 100644 --- a/server/scheduler.go +++ b/server/scheduler.go @@ -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. diff --git a/server/triggers.go b/server/triggers.go index 43a981c..eae7bc2 100644 --- a/server/triggers.go +++ b/server/triggers.go @@ -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 diff --git a/web/src/components/agent-editor.tsx b/web/src/components/agent-editor.tsx index 84f5a85..f582a1d 100644 --- a/web/src/components/agent-editor.tsx +++ b/web/src/components/agent-editor.tsx @@ -721,11 +721,13 @@ function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent const [onGoalMet, setOnGoalMet] = React.useState(false); const [onTaskTimeout, setOnTaskTimeout] = React.useState(false); const [onToolCall, setOnToolCall] = React.useState(false); + const [onTaskCreate, setOnTaskCreate] = React.useState(false); const [intervalMsg, setIntervalMsg] = React.useState(""); const [findingMsg, setFindingMsg] = React.useState(""); const [goalMsg, setGoalMsg] = React.useState(""); const [taskTimeoutMsg, setTaskTimeoutMsg] = React.useState(""); const [toolCallMsg, setToolCallMsg] = React.useState(""); + const [taskCreateMsg, setTaskCreateMsg] = React.useState(""); const [toolNames, setToolNames] = React.useState([]); const [saving, setSaving] = React.useState(false); @@ -745,7 +747,7 @@ function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent async function create() { const n = onInterval ? Math.max(1, Math.floor(Number(intervalSec) || 0)) : 0; - if (n === 0 && !onFinding && !onGoalMet && !onTaskTimeout && !onToolCall) { + if (n === 0 && !onFinding && !onGoalMet && !onTaskTimeout && !onToolCall && !onTaskCreate) { toast.error("至少选择一种触发条件"); return; } @@ -762,11 +764,13 @@ function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent on_goal_met: onGoalMet, on_task_timeout: onTaskTimeout, on_tool_call: onToolCall, + on_task_create: onTaskCreate, interval_message: intervalMsg.trim(), finding_message: findingMsg.trim(), goal_message: goalMsg.trim(), task_timeout_message: taskTimeoutMsg.trim(), tool_call_message: toolCallMsg.trim(), + task_create_message: taskCreateMsg.trim(), tool_names: onToolCall ? toolNames : [], }); toast.success("已添加触发器"); @@ -776,11 +780,13 @@ function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent setOnGoalMet(false); setOnTaskTimeout(false); setOnToolCall(false); + setOnTaskCreate(false); setIntervalMsg(""); setFindingMsg(""); setGoalMsg(""); setTaskTimeoutMsg(""); setToolCallMsg(""); + setTaskCreateMsg(""); setToolNames([]); reload(); } catch (e) { @@ -798,11 +804,13 @@ function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent on_goal_met: t.on_goal_met, on_task_timeout: t.on_task_timeout, on_tool_call: t.on_tool_call, + on_task_create: t.on_task_create, interval_message: t.interval_message, finding_message: t.finding_message, goal_message: t.goal_message, task_timeout_message: t.task_timeout_message, tool_call_message: t.tool_call_message, + task_create_message: t.task_create_message, tool_names: t.tool_names, }); reload(); @@ -826,6 +834,7 @@ function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent if (t.on_goal_met) parts.push("目标达成"); if (t.on_task_timeout) parts.push("任务超时"); if (t.on_tool_call) parts.push(`工具调用(${t.tool_names.length})`); + if (t.on_task_create) parts.push("任务创建"); return parts.join(" · ") || "（无条件）"; } @@ -999,6 +1008,17 @@ function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent )} + {/* 任务创建 */} +
+ + {onTaskCreate && ( +