diff --git a/db/config.go b/db/config.go index 3505a48..0322d83 100644 --- a/db/config.go +++ b/db/config.go @@ -132,13 +132,20 @@ type Agent struct { // 任务级超时收尾词(与 per-run 两套;仅 worker/planner 用);空/0=用代码内置默认。 TaskTimeoutWrapupPrompt string `json:"task_timeout_wrapup_prompt"` TaskTimeoutWrapupMaxTurns int `json:"task_timeout_wrapup_max_turns"` + // P3 触发后处理策略(仅自定义 agent 有意义): + // TriggerRunMode serial|parallel — 串行排队 / 每次触发各自并发一个会话 + // TriggerMergeMode by_task|all|none — 仅 serial 用:同任务合并 / 全部合并 / 不合并 + // TriggerMaxParallel — 仅 parallel 用的每 agent 并发上限;0=不限 + TriggerRunMode string `json:"trigger_run_mode"` + TriggerMergeMode string `json:"trigger_merge_mode"` + TriggerMaxParallel int `json:"trigger_max_parallel"` } -const agentCols = `id,key,name,COALESCE(description,''),role,builtin,enabled,COALESCE(max_turns,0),COALESCE(run_seconds,600),COALESCE(web_search,false),COALESCE(interactive_shell,false),COALESCE(wrapup_prompt,''),COALESCE(wrapup_max_turns,0),COALESCE(task_timeout_wrapup_prompt,''),COALESCE(task_timeout_wrapup_max_turns,0)` +const agentCols = `id,key,name,COALESCE(description,''),role,builtin,enabled,COALESCE(max_turns,0),COALESCE(run_seconds,600),COALESCE(web_search,false),COALESCE(interactive_shell,false),COALESCE(wrapup_prompt,''),COALESCE(wrapup_max_turns,0),COALESCE(task_timeout_wrapup_prompt,''),COALESCE(task_timeout_wrapup_max_turns,0),COALESCE(trigger_run_mode,'serial'),COALESCE(trigger_merge_mode,'by_task'),COALESCE(trigger_max_parallel,5)` func scanAgent(sc interface{ Scan(...any) error }) (*Agent, error) { var a Agent - err := sc.Scan(&a.ID, &a.Key, &a.Name, &a.Description, &a.Role, &a.Builtin, &a.Enabled, &a.MaxTurns, &a.RunSecs, &a.WebSearch, &a.InteractiveShell, &a.WrapupPrompt, &a.WrapupMaxTurns, &a.TaskTimeoutWrapupPrompt, &a.TaskTimeoutWrapupMaxTurns) + err := sc.Scan(&a.ID, &a.Key, &a.Name, &a.Description, &a.Role, &a.Builtin, &a.Enabled, &a.MaxTurns, &a.RunSecs, &a.WebSearch, &a.InteractiveShell, &a.WrapupPrompt, &a.WrapupMaxTurns, &a.TaskTimeoutWrapupPrompt, &a.TaskTimeoutWrapupMaxTurns, &a.TriggerRunMode, &a.TriggerMergeMode, &a.TriggerMaxParallel) return &a, err } @@ -300,6 +307,28 @@ func (d *DB) SetAgentRunSeconds(key string, runSecs int) error { return err } +// SetAgentTriggerBehavior stores an agent's P3 trigger post-processing策略: +// runMode(serial|parallel) / mergeMode(by_task|all|none) / maxParallel(parallel 用,0=不限)。 +// 枚举做白名单校验,非法值回落默认,避免脏数据把调度 pump 带偏。 +func (d *DB) SetAgentTriggerBehavior(key, runMode, mergeMode string, maxParallel int) error { + switch runMode { + case "serial", "parallel": + default: + runMode = "serial" + } + switch mergeMode { + case "by_task", "all", "none": + default: + mergeMode = "by_task" + } + if maxParallel < 0 { + maxParallel = 0 + } + _, err := d.Exec(`UPDATE agents SET trigger_run_mode=$1, trigger_merge_mode=$2, trigger_max_parallel=$3 WHERE key=$4`, + runMode, mergeMode, maxParallel, key) + return err +} + type PromptVar struct { Name string `json:"name"` Description string `json:"description"` diff --git a/db/schema.sql b/db/schema.sql index fbc5e5d..578315f 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -269,10 +269,17 @@ CREATE TABLE IF NOT EXISTS agents ( wrapup_max_turns INTEGER NOT NULL DEFAULT 0, task_timeout_wrapup_prompt TEXT NOT NULL DEFAULT '', task_timeout_wrapup_max_turns INTEGER NOT NULL DEFAULT 0, + trigger_run_mode TEXT NOT NULL DEFAULT 'serial' CHECK (trigger_run_mode IN ('serial','parallel')), + trigger_merge_mode TEXT NOT NULL DEFAULT 'by_task' CHECK (trigger_merge_mode IN ('by_task','all','none')), + trigger_max_parallel INTEGER NOT NULL DEFAULT 5, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), CONSTRAINT agents_role_ck CHECK (role IN ('goals','main','planner','worker','assistant')) ); +-- 加列迁移(已发版,旧库升级补列;新库 CREATE 已含。迁移不带 CHECK:旧库存量安全 + 后端写入白名单兜底)。 +ALTER TABLE agents ADD COLUMN IF NOT EXISTS trigger_run_mode TEXT NOT NULL DEFAULT 'serial'; +ALTER TABLE agents ADD COLUMN IF NOT EXISTS trigger_merge_mode TEXT NOT NULL DEFAULT 'by_task'; +ALTER TABLE agents ADD COLUMN IF NOT EXISTS trigger_max_parallel INTEGER NOT NULL DEFAULT 5; DROP TRIGGER IF EXISTS trg_agents_upd ON agents; CREATE TRIGGER trg_agents_upd BEFORE UPDATE ON agents FOR EACH ROW EXECUTE FUNCTION set_updated_at(); diff --git a/server/conversations.go b/server/conversations.go index dc2dba4..dc9fa9e 100644 --- a/server/conversations.go +++ b/server/conversations.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "math" "net/http" "strconv" "strings" @@ -351,51 +352,111 @@ func (s *Server) runConversationSync(c *db.Conversation, msg, busyKey string) { _ = pg.TouchConversation(c.ID) } -// StartTriggeredRun enqueues a P3 trigger fire for agentKey. Fires for the SAME -// agent queue up and run one at a time (FIFO) — the next only starts after the -// previous conversation finishes. Distinct agents still run concurrently. taskID + -// mergeable let the drainer coalesce same-task event triggers before a run starts. +// triggerBehavior is an agent's cached P3 trigger post-processing策略 (see the +// agents table trigger_* columns). Read once per fire in StartTriggeredRun so the +// pump never touches the DB while holding queueMu. +type triggerBehavior struct { + runMode string // serial | parallel + mergeMode string // by_task | all | none (serial 用;parallel 忽略,每条各自一个会话) + maxParallel int // parallel 用的每 agent 并发上限;<=0=不限 +} + +// readTriggerBehavior loads an agent's策略, falling back to safe defaults +// (serial / by_task / 5) on any error or unknown enum value. +func (s *Server) readTriggerBehavior(agentKey string) triggerBehavior { + b := triggerBehavior{runMode: "serial", mergeMode: "by_task", maxParallel: 5} + if s.m.pg == nil { + return b + } + a, err := s.m.pg.GetAgentByKey(agentKey) + if err != nil || a == nil { + return b + } + if a.TriggerRunMode == "parallel" { + b.runMode = "parallel" + } + switch a.TriggerMergeMode { + case "by_task", "all", "none": + b.mergeMode = a.TriggerMergeMode + } + b.maxParallel = a.TriggerMaxParallel + return b +} + +// StartTriggeredRun enqueues a P3 trigger fire for agentKey and pumps the queue. The +// 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) { 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}) - depth := len(s.triggerQ[agentKey]) - if !s.triggerRun[agentKey] { - s.triggerRun[agentKey] = true - go s.drainTriggerQueue(agentKey) + s.pumpLocked(agentKey) + s.queueMu.Unlock() +} + +// pumpLocked launches queued fires for one agent up to its concurrency limit, then +// returns. Serial → limit 1; parallel → limit = maxParallel (<=0 → unlimited). Each +// launched run re-pumps on completion (runAndPump) to fill the freed slot. Caller +// holds queueMu. +func (s *Server) pumpLocked(agentKey string) { + if s.ctx.Err() != nil { + return + } + cfg := s.triggerCfg[agentKey] + limit := 1 + if cfg.runMode == "parallel" { + if cfg.maxParallel <= 0 { + limit = math.MaxInt + } else { + limit = cfg.maxParallel + } + } + for s.triggerActive[agentKey] < limit && len(s.triggerQ[agentKey]) > 0 { + item := s.nextTriggerRun(agentKey, cfg) + s.triggerActive[agentKey]++ + go s.runAndPump(agentKey, item) + } +} + +// runAndPump runs one fire to completion, then decrements the active counter and +// re-pumps to fill the freed slot (cleaning up the agent's maps when fully idle). +func (s *Server) runAndPump(agentKey string, item triggeredRun) { + s.runTriggeredRun(item) + s.queueMu.Lock() + s.triggerActive[agentKey]-- + if s.triggerActive[agentKey] <= 0 && len(s.triggerQ[agentKey]) == 0 { + delete(s.triggerActive, agentKey) + delete(s.triggerQ, agentKey) + delete(s.triggerCfg, agentKey) } else { - log.Printf("[trigger] %s busy → queued (depth %d)", agentKey, depth) + s.pumpLocked(agentKey) } s.queueMu.Unlock() } -// drainTriggerQueue runs one agent's queued trigger fires serially until the queue -// empties, then clears the active flag. Before each run it coalesces all queued -// finding/goal fires from the SAME task into one conversation (nextTriggerRun), so a -// burst of a task's events becomes a single session. A panic is contained. -func (s *Server) drainTriggerQueue(agentKey string) { - for { - s.queueMu.Lock() - if len(s.triggerQ[agentKey]) == 0 || s.ctx.Err() != nil { - delete(s.triggerQ, agentKey) - delete(s.triggerRun, agentKey) - s.queueMu.Unlock() - return - } - item := s.nextTriggerRun(agentKey) - s.queueMu.Unlock() - - s.runTriggeredRun(item) - } -} - -// nextTriggerRun pops the head of the agent's queue; if the head is a mergeable -// (finding/goal) fire, it also pulls every other queued mergeable fire from the SAME -// task and merges them into one run. Caller holds queueMu. -func (s *Server) nextTriggerRun(agentKey string) triggeredRun { +// nextTriggerRun pops the next run from the agent's queue per the merge mode: +// +// parallel / none → one head item, no merge +// by_task → head + all queued same-task mergeable fires merged into one +// all → the ENTIRE queue merged into one run (regardless of task/type) +// +// Caller holds queueMu. +func (s *Server) nextTriggerRun(agentKey string, cfg triggerBehavior) triggeredRun { q := s.triggerQ[agentKey] + if cfg.runMode == "parallel" || cfg.mergeMode == "none" { + s.triggerQ[agentKey] = q[1:] + return q[0] + } + if cfg.mergeMode == "all" { + s.triggerQ[agentKey] = q[:0:0] + return mergeAllRuns(q) + } + // by_task: head + same-task mergeable fires. head := q[0] if !head.mergeable || head.taskID == 0 { s.triggerQ[agentKey] = q[1:] @@ -435,6 +496,27 @@ func mergeTriggeredRuns(items []triggeredRun) triggeredRun { } } +// mergeAllRuns folds the ENTIRE queued burst into one run (merge_mode='all'), +// regardless of task or type — a header plus each fire's message. +func mergeAllRuns(items []triggeredRun) triggeredRun { + if len(items) == 1 { + return items[0] + } + first := items[0] + 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) + } + return triggeredRun{ + agentKey: first.agentKey, + title: fmt.Sprintf("合并触发 · 全部 · %d 条", len(items)), + message: b.String(), + taskID: first.taskID, + mergeable: true, + } +} + // runTriggeredRun creates the conversation for one queued fire, records the human // turn, and runs it synchronously (blocks until the run ends). recover keeps a // panic from killing the drain loop and wedging the agent's queue. diff --git a/server/dto.go b/server/dto.go index 28595fe..44385e4 100644 --- a/server/dto.go +++ b/server/dto.go @@ -304,6 +304,10 @@ type AgentDTO struct { RunSecs int `json:"run_seconds"` WebSearch bool `json:"web_search"` InteractiveShell bool `json:"interactive_shell"` + // P3 触发后处理策略(仅自定义 agent 有意义)。 + TriggerRunMode string `json:"trigger_run_mode"` + TriggerMergeMode string `json:"trigger_merge_mode"` + TriggerMaxParallel int `json:"trigger_max_parallel"` // binding counts (populated only by the list endpoint) — shown on agent cards. McpCount int `json:"mcp_count"` SkillCount int `json:"skill_count"` @@ -312,17 +316,20 @@ type AgentDTO struct { func agentDTO(a *db.Agent) AgentDTO { return AgentDTO{ - ID: i64s(a.ID), - Key: a.Key, - Name: a.Name, - Description: a.Description, - Role: a.Role, - Builtin: a.Builtin, - Enabled: a.Enabled, - MaxTurns: a.MaxTurns, - RunSecs: a.RunSecs, - WebSearch: a.WebSearch, - InteractiveShell: a.InteractiveShell, + ID: i64s(a.ID), + Key: a.Key, + Name: a.Name, + Description: a.Description, + Role: a.Role, + Builtin: a.Builtin, + Enabled: a.Enabled, + MaxTurns: a.MaxTurns, + RunSecs: a.RunSecs, + WebSearch: a.WebSearch, + InteractiveShell: a.InteractiveShell, + TriggerRunMode: a.TriggerRunMode, + TriggerMergeMode: a.TriggerMergeMode, + TriggerMaxParallel: a.TriggerMaxParallel, } } diff --git a/server/server.go b/server/server.go index 52f5389..edd8a7f 100644 --- a/server/server.go +++ b/server/server.go @@ -17,14 +17,14 @@ import ( "sync" "time" - "github.com/Autumn-27/norma/llm" - "github.com/Autumn-27/norma/memory" - actool "github.com/Autumn-27/norma/tool" - "github.com/Autumn-27/norma/transcript" "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/db" "github.com/Autumn-27/artex/llmrec" "github.com/Autumn-27/artex/report" + "github.com/Autumn-27/norma/llm" + "github.com/Autumn-27/norma/memory" + actool "github.com/Autumn-27/norma/tool" + "github.com/Autumn-27/norma/transcript" ) // Server exposes the ARTEX backend over a JSON HTTP API for the shadcn/ui @@ -42,7 +42,7 @@ type Server struct { chatAgent *agent.ChatAgent // conversational runner for the chat page; nil w/o LLM llmCfg agent.Config // current LLM config (key not exposed) llmOn bool - llmProf string // active LLM profile name (for llmrec tagging) + llmProf string // active LLM profile name (for llmrec tagging) // chatBusy guards the per-task main-agent run: the chat handler launches the // agent on the server's background ctx (not the request ctx) and returns @@ -57,15 +57,18 @@ type Server struct { // trigger queue — the drain goroutine simply proceeds to the next queued fire. chatCancel map[string]context.CancelFunc - // triggerQ serializes P3 trigger fires PER AGENT: multiple fires for the same - // custom agent queue up (FIFO) and run one at a time — the next conversation - // only starts after the previous one finishes. triggerRun marks that a drain - // goroutine is already active for that agent key so we don't spawn two. Distinct - // agents still run concurrently. Queue is in-memory (matches chatBusy); a restart - // drops pending fires — the scheduler re-fires on its next tick from watermarks. - queueMu sync.Mutex - triggerQ map[string][]triggeredRun - triggerRun map[string]bool + // triggerQ buffers P3 trigger fires PER AGENT. A per-agent "pump" launches runs up + // to a concurrency limit derived from the agent's策略: serial → limit 1 (+ optional + // merge); parallel → limit = trigger_max_parallel (0=∞), no merge. triggerActive + // counts in-flight runs per agent (replaces a boolean drain flag); a run's + // completion decrements it and re-pumps to fill the freed slot. triggerCfg caches + // the agent's last-read策略 so the pump never queries the DB while holding queueMu. + // Distinct agents always run concurrently. Queue is in-memory (matches chatBusy); a + // restart drops pending fires — the scheduler re-fires from watermarks next tick. + queueMu sync.Mutex + triggerQ map[string][]triggeredRun + triggerActive map[string]int + triggerCfg map[string]triggerBehavior // profAgents caches the dedicated planner/worker built for a specific (non-active) // LLM profile, keyed by profile id, so tasks pinned to that profile share one @@ -100,7 +103,8 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string) *Serv log.Fatalf("[auth] JWT key: %v", err) } s := &Server{m: m, engine: NewEngine(m), ctx: ctx, skillDir: skillDir, jwtKey: key, chatBusy: map[string]bool{}, - chatCancel: map[string]context.CancelFunc{}, triggerQ: map[string][]triggeredRun{}, triggerRun: map[string]bool{}, + chatCancel: map[string]context.CancelFunc{}, triggerQ: map[string][]triggeredRun{}, + triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{}, profAgents: map[int64]*profBundle{}, profChatAgents: map[int64]*agent.ChatAgent{}} // per-task LLM: a task pinned to a specific profile runs on that profile's // dedicated planner/worker; unpinned tasks fall back to the global active pair. @@ -157,11 +161,11 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string) *Serv } wireAgentAugment(m.pg, s.skillDir, s.hostTools) // 可见 skills/MCP + 流量/编排 host 工具装配进 agent 工具集 domainReg := buildDomainReg(m.Assets()) - wireTools(m.pg, domainReg) // 内置工具表：按 agent 过滤 + 覆盖描述/schema + 注入默认值 - seedPrompts(m.pg) // 内置 agent 默认提示词正文播种进 agent_prompts(仅空时) - s.seedOrchestrationTools() // P2 跨任务编排工具 seed 进 tools 表(可按 agent 绑定) - s.seedPythonInterpreter() // 自定义脚本工具:开机检测 python 解释器入库(仅空时) - go newScheduler(s).Run(s.ctx) // P3 触发器调度(定时/finding/目标事件),仅自定义 agent + wireTools(m.pg, domainReg) // 内置工具表：按 agent 过滤 + 覆盖描述/schema + 注入默认值 + seedPrompts(m.pg) // 内置 agent 默认提示词正文播种进 agent_prompts(仅空时) + s.seedOrchestrationTools() // P2 跨任务编排工具 seed 进 tools 表(可按 agent 绑定) + s.seedPythonInterpreter() // 自定义脚本工具:开机检测 python 解释器入库(仅空时) + go newScheduler(s).Run(s.ctx) // P3 触发器调度(定时/finding/目标事件),仅自定义 agent // Fill the tool cache for any enabled MCP that has none yet (notably the // seeded browser MCP on first run). Async so it never blocks startup. go s.discoverEmptyMCPsOnStartup() @@ -721,8 +725,8 @@ func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) { active = t.ID } list := s.m.List() - toks, _ := s.m.PG().TokenTotalsAll() // whole-task token totals, one query for all tasks - lastAct, _ := s.m.PG().LastActivityAll() // persisted last-activity per task, one query + toks, _ := s.m.PG().TokenTotalsAll() // whole-task token totals, one query for all tasks + lastAct, _ := s.m.PG().LastActivityAll() // persisted last-activity per task, one query goalCounts, _ := s.m.PG().GoalCountsAll() // goal progress per exploration, one query dtos := make([]TaskDTO, 0, len(list)) for _, t := range list { @@ -1080,8 +1084,6 @@ func (s *Server) getTask(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, t) } - - func (s *Server) frontier(w http.ResponseWriter, r *http.Request) { t := s.m.ResolveTask(r.URL.Query().Get("task")) if t == nil { @@ -1402,7 +1404,6 @@ func (s *Server) activityDetail(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"detail": d}) } - func (s *Server) getTraffic(w http.ResponseWriter, r *http.Request) { tr := s.m.Traffic() if tr == nil { @@ -1457,15 +1458,15 @@ func (s *Server) settingsPayload() map[string]any { on, backend, braveKey, tavilyKey, proxy := s.m.WebSearch() pyStored, _, _ := s.m.pg.GetSetting(settingPythonInterp) return map[string]any{ - "traffic_capture": s.m.TrafficEnabled(), - "llm_record": s.m.LLMRecordEnabled(), - "web_search_enabled": on, - "web_search_backend": backend, - "brave_key_set": strings.TrimSpace(braveKey) != "", - "tavily_key_set": strings.TrimSpace(tavilyKey) != "", - "web_search_proxy": proxy, // 独立出口代理(http/https/socks5)，空=直连 - "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) - "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 + "traffic_capture": s.m.TrafficEnabled(), + "llm_record": s.m.LLMRecordEnabled(), + "web_search_enabled": on, + "web_search_backend": backend, + "brave_key_set": strings.TrimSpace(braveKey) != "", + "tavily_key_set": strings.TrimSpace(tavilyKey) != "", + "web_search_proxy": proxy, // 独立出口代理(http/https/socks5)，空=直连 + "python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) + "workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效 } } diff --git a/server/server_mgmt.go b/server/server_mgmt.go index 470f134..e54d6d1 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -20,9 +20,9 @@ import ( "text/template/parse" "time" - "github.com/Autumn-27/norma/skill" "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/db" + "github.com/Autumn-27/norma/skill" ) // reSkillName is the base character-set pattern; validSkillName enforces @@ -212,24 +212,32 @@ func (s *Server) pgSaveAgentConfig(w http.ResponseWriter, r *http.Request) { if !ok { return } - // run_seconds is optional (pointer) so omitting it leaves the stored value - // untouched rather than resetting it to 0/unlimited. + // All fields optional (pointers) so a partial patch (e.g. the triggers tab sending + // only trigger_* fields) leaves the untouched settings alone instead of resetting + // max_turns/run_seconds to 0. var req struct { - MaxTurns int `json:"max_turns"` + MaxTurns *int `json:"max_turns"` RunSeconds *int `json:"run_seconds"` WebSearch *bool `json:"web_search"` InteractiveShell *bool `json:"interactive_shell"` + // P3 触发后处理策略(三者一起可选,提供任一即整体写入;未提供则不动)。 + TriggerRunMode *string `json:"trigger_run_mode"` + TriggerMergeMode *string `json:"trigger_merge_mode"` + TriggerMaxParallel *int `json:"trigger_max_parallel"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeErr(w, 400, err.Error()) return } - if req.MaxTurns < 0 { - req.MaxTurns = 0 - } - if err := pg.SetAgentMaxTurns(a.Key, req.MaxTurns); err != nil { - writeErr(w, 500, err.Error()) - return + if req.MaxTurns != nil { + mt := *req.MaxTurns + if mt < 0 { + mt = 0 + } + if err := pg.SetAgentMaxTurns(a.Key, mt); err != nil { + writeErr(w, 500, err.Error()) + return + } } if req.RunSeconds != nil { rs := *req.RunSeconds @@ -253,6 +261,24 @@ func (s *Server) pgSaveAgentConfig(w http.ResponseWriter, r *http.Request) { return } } + // P3 触发策略:三者作为一组写入(SetAgentTriggerBehavior 一次写三列),缺的字段用 + // 当前存量值回填,避免只传一个把另两个覆盖成默认。 + if req.TriggerRunMode != nil || req.TriggerMergeMode != nil || req.TriggerMaxParallel != nil { + runMode, mergeMode, maxPar := a.TriggerRunMode, a.TriggerMergeMode, a.TriggerMaxParallel + if req.TriggerRunMode != nil { + runMode = *req.TriggerRunMode + } + if req.TriggerMergeMode != nil { + mergeMode = *req.TriggerMergeMode + } + if req.TriggerMaxParallel != nil { + maxPar = *req.TriggerMaxParallel + } + if err := pg.SetAgentTriggerBehavior(a.Key, runMode, mergeMode, maxPar); err != nil { + writeErr(w, 500, err.Error()) + return + } + } // rebuild the live agents so the new max_turns/run_seconds apply without a restart. s.cfgMu.Lock() cfg, on := s.llmCfg, s.llmOn @@ -260,7 +286,10 @@ func (s *Server) pgSaveAgentConfig(w http.ResponseWriter, r *http.Request) { if on { _ = s.applyLLM(cfg) } - resp := map[string]any{"ok": true, "max_turns": req.MaxTurns} + resp := map[string]any{"ok": true} + if req.MaxTurns != nil { + resp["max_turns"] = *req.MaxTurns + } if req.RunSeconds != nil { resp["run_seconds"] = *req.RunSeconds } @@ -1544,7 +1573,7 @@ func (s *Server) pgActivateProfile(w http.ResponseWriter, r *http.Request) { // the OpenAI path, it falls back to the OpenAI endpoint at the stripped root. func (s *Server) pgListModels(w http.ResponseWriter, r *http.Request) { var req struct { - Provider string `json:"provider"` // "openai" | "anthropic" + Provider string `json:"provider"` // "openai" | "anthropic" BaseURL string `json:"base_url"` APIKey string `json:"api_key"` Proxy string `json:"proxy"` diff --git a/web/src/app/(main)/system/agents/page.tsx b/web/src/app/(main)/system/agents/page.tsx index 683c6ca..7add813 100644 --- a/web/src/app/(main)/system/agents/page.tsx +++ b/web/src/app/(main)/system/agents/page.tsx @@ -258,6 +258,15 @@ export default function AgentsPage() { { + const t = e.detail.originalEvent.target as Element | null; + if (t?.closest?.("[data-radix-popper-content-wrapper],[data-radix-select-content],[role='listbox']")) { + e.preventDefault(); + } + }} > {editing && ( <> diff --git a/web/src/components/agent-editor.tsx b/web/src/components/agent-editor.tsx index 4173f07..84f5a85 100644 --- a/web/src/components/agent-editor.tsx +++ b/web/src/components/agent-editor.tsx @@ -12,6 +12,7 @@ import { Label } from "@/components/ui/label"; import { Checkbox } from "@/components/ui/checkbox"; import { Switch } from "@/components/ui/switch"; import { Separator } from "@/components/ui/separator"; +import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { Dialog, @@ -24,7 +25,7 @@ import { import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip"; import { api } from "@/lib/api"; import { cn } from "@/lib/utils"; -import type { AgentDetail, AgentTrigger, MCPServer, PromptVar, PromptVersion, Settings, SkillItem, Tool } from "@/lib/types"; +import type { Agent, AgentDetail, AgentTrigger, MCPServer, PromptVar, PromptVersion, Settings, SkillItem, Tool } from "@/lib/types"; // Traffic tools are host tools gated by the global 流量捕获 switch: bindable, but // only usable when capture is on. Keep this list in sync with traffic.SeedToolMetas. @@ -626,7 +627,7 @@ export function AgentEditor({ agentKey, onSaved }: { agentKey: string; onSaved?: {/* 触发(P3, 仅自定义 agent) */} {isCustom && ( - + )} @@ -690,9 +691,30 @@ function DiffView({ oldText, newText }: { oldText: string; newText: string }) { // AgentTriggersTab manages a custom agent's P3 triggers: list + add + delete. // Each trigger fires (定时/发现finding/目标达成/任务超时/工具调用，可多选) → a new conversation runs // in parallel with the base user message + auto context appended by the backend. -function AgentTriggersTab({ agentKey }: { agentKey: string }) { +function AgentTriggersTab({ agentKey, agent }: { agentKey: string; agent?: Agent }) { const [triggers, setTriggers] = React.useState([]); const [tools, setTools] = React.useState([]); + // 触发后处理策略(每 agent);初值来自 agent detail,改动即保存。 + const [runMode, setRunMode] = React.useState<"serial" | "parallel">(agent?.trigger_run_mode ?? "serial"); + const [mergeMode, setMergeMode] = React.useState<"by_task" | "all" | "none">(agent?.trigger_merge_mode ?? "by_task"); + const [maxParallel, setMaxParallel] = React.useState(String(agent?.trigger_max_parallel ?? 5)); + React.useEffect(() => { + setRunMode(agent?.trigger_run_mode ?? "serial"); + setMergeMode(agent?.trigger_merge_mode ?? "by_task"); + setMaxParallel(String(agent?.trigger_max_parallel ?? 5)); + }, [agent?.trigger_run_mode, agent?.trigger_merge_mode, agent?.trigger_max_parallel]); + + async function saveBehavior(patch: { + trigger_run_mode?: "serial" | "parallel"; + trigger_merge_mode?: "by_task" | "all" | "none"; + trigger_max_parallel?: number; + }) { + try { + await api.saveAgentConfig(agentKey, patch); + } catch (e) { + toast.error("保存策略失败：" + (e as Error).message); + } + } const [onInterval, setOnInterval] = React.useState(false); const [intervalSec, setIntervalSec] = React.useState("60"); const [onFinding, setOnFinding] = React.useState(false); @@ -810,10 +832,86 @@ function AgentTriggersTab({ agentKey }: { agentKey: string }) { return (

- 触发器让这个自定义 Agent 自动运行：每次触发都新建一个会话并行运行（在「对话」页可见）。 + 触发器让这个自定义 Agent 自动运行：每次触发都新建一个会话运行（在「对话」页可见）。 可多选触发条件；系统会把「本次为何触发 + 相关任务/finding/目标」自动附加到你写的基础消息后面。

+ {/* 触发后处理策略 */} +
+ +
+
+ + +
+ +
+ + +
+ + {runMode === "parallel" && ( +
+ + setMaxParallel(e.target.value)} + onBlur={() => { + const n = Math.max(0, Math.floor(Number(maxParallel) || 0)); + setMaxParallel(String(n)); + saveBehavior({ trigger_max_parallel: n }); + }} + /> +
+ )} +
+

+ {runMode === "parallel" + ? "并行：每次触发立即各开一个会话并发运行，不合并；超过最大并发的触发排队等空位。" + : mergeMode === "by_task" + ? "串行·按任务合并：同 agent 一次跑一个；排队中同一任务的事件触发合并成一条会话。" + : mergeMode === "all" + ? "串行·全部合并：同 agent 一次跑一个；取队列时把当前排队的所有触发合并成一条会话。" + : "串行·不合并：同 agent 一次跑一个；每条触发各自一个会话。"} +

+
+ {/* 新增触发器 */}
diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 334ed8d..cb27de5 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -315,8 +315,18 @@ export const api = { updateTrigger: (id: number, t: Omit) => patch<{ ok: boolean }>(`/triggers/${id}`, t), deleteTrigger: (id: number) => del<{ deleted: number }>(`/triggers/${id}`), - saveAgentConfig: (key: string, patch: { max_turns: number; run_seconds?: number; web_search?: boolean; interactive_shell?: boolean }) => - put<{ ok: boolean }>(`/agents/${key}/config`, patch), + saveAgentConfig: ( + key: string, + patch: { + max_turns?: number; + run_seconds?: number; + web_search?: boolean; + interactive_shell?: boolean; + trigger_run_mode?: "serial" | "parallel"; + trigger_merge_mode?: "by_task" | "all" | "none"; + trigger_max_parallel?: number; + }, + ) => put<{ ok: boolean }>(`/agents/${key}/config`, patch), agentPromptVersions: (key: string) => get<{ versions: PromptVersion[] }>(`/agents/${key}/prompts`).then((r) => arr(r.versions)), agentVariables: (key: string) => get<{ variables: PromptVar[] }>(`/agents/${key}/variables`).then((r) => arr(r.variables)), previewAgentPrompt: (key: string, template: string, sample?: Record) => diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index d99e123..525724f 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -387,6 +387,10 @@ export interface Agent { run_seconds?: number; // worker 单次运行墙钟上限(秒)；0 = 无限制 web_search?: boolean; // 是否启用网络搜索(受系统全局开关门控) interactive_shell?: boolean; // 是否启用交互式 shell(持久 PTY 会话工具族) + // P3 触发后处理策略(仅自定义 agent 有意义) + trigger_run_mode?: "serial" | "parallel"; // 串行排队 / 每次触发各自并发一个会话 + trigger_merge_mode?: "by_task" | "all" | "none"; // 仅 serial：同任务合并 / 全部合并 / 不合并 + trigger_max_parallel?: number; // 仅 parallel：每 agent 并发上限；0=不限 // 绑定数量(仅列表接口返回)：可见 MCP / 可见 Skill / 绑定工具 mcp_count?: number; skill_count?: number;