feat(agent): 触发后处理可配置(运行模式/合并模式/并发上限) + 修复抽屉内下拉误关

每个自定义 agent 可配置触发后如何调度运行:
- 运行模式 serial|parallel:串行排队一次一个 / 每次触发各自并发一个会话
- 合并模式 by_task|all|none(仅 serial):同任务合并 / 全部合并成一条 / 不合并
- parallel 带每 agent 最大并发上限(0=不限),超限排队等空位

实现:
- DB: agents 加 trigger_run_mode/trigger_merge_mode/trigger_max_parallel 三列
  (schema + 幂等 ALTER 迁移,已发版补旧库);Agent 结构/scan/SetAgentTriggerBehavior
- 调度重构: triggerRun bool → triggerActive 计数 + triggerCfg 缓存;
  StartTriggeredRun 走 pumpLocked/runAndPump 统一泵模型,nextTriggerRun 按合并模式
  取(新增 mergeAllRuns);并发安全(session 按 conv-id 隔离,已确认)
- config 接口暴露/接收三字段;max_turns 改可选指针,避免触发 tab 部分 patch 清零
- 前端: 触发 tab 顶部加运行/合并模式下拉 + 最大并发输入,改动即存

修复: 抽屉(Sheet)内 Select 下拉 portal 到抽屉外,点选项/点别处被当成点击外部
误关抽屉。SelectContent 改 position=popper,并给 SheetContent 加 onInteractOutside
守卫(按 e.detail.originalEvent.target 判定弹层内点击则不关)。

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Autumn-27andClaude Opus 4.8 committed 2026-08-08 04:23:09 -04:00
1 parent f9716269ea
commit 6d8fa5bc48
10 files changed
+374 -98

No files matched your search

+115 -33
View File
@@ -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.
+18 -11
View File
@@ -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,
}
}
+35 -34
View File
@@ -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);对之后启动的任务生效
}
}
+41 -12
View File
@@ -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"`