Register plugin workflows and drop them on uninstall
Why: Intent routing needs a registry of workflow definitions, and uninstall was leaving intents and workflows in place. Evidence: go test ./internal/workflow/ ./internal/plugins/ ./internal/app/ passed. Validation rejects duplicate ids, unknown kinds, denied tools, and both step caps. Unregister clears the registry and the intent gate. Impact: Script steps must stay inside the plugin directory. The server holds the registry for later routing. Refs: feature/agent-harness-p4 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Vendored
+1
-1
Submodule deps/agent-system-core updated: 9b8cf79c96...f93f473a0e
@@ -46,6 +46,7 @@ import (
|
||||
"agent-system-backend/internal/squadadapter"
|
||||
"agent-system-backend/internal/store"
|
||||
"agent-system-backend/internal/tools"
|
||||
"agent-system-backend/internal/workflow"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
@@ -205,6 +206,11 @@ func Run(opt Options) error {
|
||||
go func() { _ = http.Serve(hostLn, hostSrv.Handler()) }()
|
||||
pluginHost.SetHostAPI("http://"+hostLn.Addr().String(), hostSrv.Token)
|
||||
pluginHost.OnIntents = func(overlay []json.RawMessage) error { return gate.Rebuild(context.Background(), overlay) }
|
||||
wfReg := workflow.NewRegistry()
|
||||
pluginHost.OnWorkflows = func(defs []core.WorkflowDef, scripts map[string]string) error {
|
||||
wfReg.Set(defs, scripts)
|
||||
return nil
|
||||
}
|
||||
for id, perr := range pluginHost.LoadDir() {
|
||||
if perr != nil {
|
||||
log.Printf("[plugins] %s: %s: %v", id, errcode.SysUnavailable, perr) // fail-closed:单个失败不阻断启动
|
||||
@@ -263,6 +269,7 @@ func Run(opt Options) error {
|
||||
}
|
||||
|
||||
srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost)
|
||||
srv.Workflows = wfReg
|
||||
srv.Cards = cardSvc
|
||||
srv.Bus = bus
|
||||
srv.Gate = gate
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"agent-system-backend/internal/settings"
|
||||
"agent-system-backend/internal/store"
|
||||
"agent-system-backend/internal/tools"
|
||||
"agent-system-backend/internal/workflow"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
@@ -26,6 +27,7 @@ type Loaded struct {
|
||||
Panels []core.PluginPanel
|
||||
Cards map[string][]core.UINode // kind → 模板节点
|
||||
Workflows []core.WorkflowDef
|
||||
Scripts map[string]string // workflow.ScriptKey → 插件目录内的脚本正文
|
||||
Intents []json.RawMessage // agent-intent overlay 形态(目录侧消费)
|
||||
Commands []json.RawMessage // Agent-Command spec overlay(cmdbus 注册)
|
||||
Workspaces []core.PluginWorkspace // 工作区弹层声明(OverlayHost 渲染)
|
||||
@@ -57,6 +59,8 @@ type Host struct {
|
||||
OnIntents func(overlay []json.RawMessage) error
|
||||
// OnCommands 在装载集合变化后回调全部命令 spec overlay(宿主接命令总线;nil = 不接)。
|
||||
OnCommands func(overlay []json.RawMessage) error
|
||||
// OnWorkflows 在装载集合变化后回调全部工作流与脚本正文(nil = 不接)。
|
||||
OnWorkflows func(defs []core.WorkflowDef, scripts map[string]string) error
|
||||
}
|
||||
|
||||
func New(st *store.Store, reg *tools.Registry, s *settings.Service, dir string) *Host {
|
||||
@@ -96,7 +100,8 @@ func (h *Host) LoadDir() map[string]error {
|
||||
}
|
||||
overlay := h.collectIntentsLocked()
|
||||
cmdOverlay := h.collectCommandsLocked()
|
||||
onIntents, onCommands := h.OnIntents, h.OnCommands
|
||||
defs, scripts := h.collectWorkflowsLocked()
|
||||
onIntents, onCommands, onWorkflows := h.OnIntents, h.OnCommands, h.OnWorkflows
|
||||
h.mu.Unlock()
|
||||
if onCommands != nil {
|
||||
if err := onCommands(cmdOverlay); err != nil {
|
||||
@@ -108,6 +113,11 @@ func (h *Host) LoadDir() map[string]error {
|
||||
results["<intents>"] = err
|
||||
}
|
||||
}
|
||||
if onWorkflows != nil {
|
||||
if err := onWorkflows(defs, scripts); err != nil {
|
||||
results["<workflows>"] = err
|
||||
}
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
@@ -132,7 +142,7 @@ func (h *Host) loadOne(dir string) error {
|
||||
return fmt.Errorf("plugins: %s: apiVersion %s 不在兼容窗口(%s, %s)", m.ID, m.APIVersion, core.PluginAPIVersion, core.PluginAPIVersionExperimental)
|
||||
}
|
||||
|
||||
ld := &Loaded{Manifest: m, Dir: dir, Cards: map[string][]core.UINode{}, Prompts: map[string]string{}, Defaults: map[string]any{}}
|
||||
ld := &Loaded{Manifest: m, Dir: dir, Cards: map[string][]core.UINode{}, Prompts: map[string]string{}, Defaults: map[string]any{}, Scripts: map[string]string{}}
|
||||
|
||||
// 同 ID 重载:先注销旧注册(含旧版本工具),再装载新资产。
|
||||
h.registry.UnregisterBySource(core.SourcePlugin, m.ID)
|
||||
@@ -178,12 +188,28 @@ func (h *Host) loadOne(dir string) error {
|
||||
ld.Cards[card.Kind] = card.Data
|
||||
}
|
||||
|
||||
// workflows/*.json
|
||||
// workflows/*.json。script 步骤的文件必须留在插件目录内,正文随定义一起交给注册表。
|
||||
for _, name := range m.Provides.Workflows {
|
||||
var wf core.WorkflowDef
|
||||
if err := readAsset(dir, "workflows", name, &wf); err != nil {
|
||||
return fmt.Errorf("plugins: %s: workflow %s: %w", m.ID, name, err)
|
||||
}
|
||||
if err := workflow.Validate(wf, workflow.DeclarativeLimits(), func(tool string) bool {
|
||||
_, _, ok := h.registry.Get(tool)
|
||||
return ok
|
||||
}); err != nil {
|
||||
return fmt.Errorf("plugins: %s: workflow %s: %w", m.ID, name, err)
|
||||
}
|
||||
for _, step := range wf.Steps {
|
||||
if step.Kind != "script" {
|
||||
continue
|
||||
}
|
||||
body, err := readPluginFile(dir, step.Script)
|
||||
if err != nil {
|
||||
return fmt.Errorf("plugins: %s: workflow %s script: %w", m.ID, name, err)
|
||||
}
|
||||
ld.Scripts[workflow.ScriptKey(wf.ID, step.Script)] = string(body)
|
||||
}
|
||||
ld.Workflows = append(ld.Workflows, wf)
|
||||
}
|
||||
|
||||
@@ -245,7 +271,8 @@ func (h *Host) loadOne(dir string) error {
|
||||
h.loaded[m.ID] = ld
|
||||
overlay := h.collectIntentsLocked()
|
||||
cmdOverlay := h.collectCommandsLocked()
|
||||
onIntents, onCommands := h.OnIntents, h.OnCommands
|
||||
defs, scripts := h.collectWorkflowsLocked()
|
||||
onIntents, onCommands, onWorkflows := h.OnIntents, h.OnCommands, h.OnWorkflows
|
||||
h.mu.Unlock()
|
||||
_ = h.st.UpsertPlugin(m.ID, dir, m)
|
||||
if onCommands != nil {
|
||||
@@ -258,6 +285,11 @@ func (h *Host) loadOne(dir string) error {
|
||||
return fmt.Errorf("plugins: %s: intents rejected by host gate (kept previous): %w", m.ID, err)
|
||||
}
|
||||
}
|
||||
if onWorkflows != nil {
|
||||
if err := onWorkflows(defs, scripts); err != nil {
|
||||
return fmt.Errorf("plugins: %s: workflows rejected: %w", m.ID, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -290,6 +322,37 @@ func (h *Host) collectCommandsLocked() []json.RawMessage {
|
||||
return out
|
||||
}
|
||||
|
||||
func (h *Host) collectWorkflowsLocked() ([]core.WorkflowDef, map[string]string) {
|
||||
ids := make([]string, 0, len(h.loaded))
|
||||
for id := range h.loaded {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
sort.Strings(ids)
|
||||
var defs []core.WorkflowDef
|
||||
scripts := map[string]string{}
|
||||
for _, id := range ids {
|
||||
ld := h.loaded[id]
|
||||
defs = append(defs, ld.Workflows...)
|
||||
for k, v := range ld.Scripts {
|
||||
scripts[k] = v
|
||||
}
|
||||
}
|
||||
return defs, scripts
|
||||
}
|
||||
|
||||
func readPluginFile(root, rel string) ([]byte, error) {
|
||||
cleaned := filepath.Clean(rel)
|
||||
if cleaned == "." || filepath.IsAbs(cleaned) {
|
||||
return nil, fmt.Errorf("path %s escapes plugin dir", rel)
|
||||
}
|
||||
full := filepath.Join(root, cleaned)
|
||||
rooted, err := filepath.Rel(root, full)
|
||||
if err != nil || rooted == ".." || strings.HasPrefix(rooted, ".."+string(filepath.Separator)) {
|
||||
return nil, fmt.Errorf("path %s escapes plugin dir", rel)
|
||||
}
|
||||
return os.ReadFile(full)
|
||||
}
|
||||
|
||||
func readAsset(dir, sub, name string, v any) error {
|
||||
p := filepath.Join(dir, sub, name+".json")
|
||||
if sub == "" {
|
||||
@@ -345,12 +408,23 @@ func (h *Host) Unregister(id string) {
|
||||
delete(h.loaded, id)
|
||||
cmd := h.sidecars[id]
|
||||
delete(h.sidecars, id)
|
||||
overlay := h.collectIntentsLocked()
|
||||
cmdOverlay := h.collectCommandsLocked()
|
||||
defs, scripts := h.collectWorkflowsLocked()
|
||||
onIntents, onCommands, onWorkflows := h.OnIntents, h.OnCommands, h.OnWorkflows
|
||||
h.mu.Unlock()
|
||||
if cmd != nil && cmd.Process != nil {
|
||||
_ = cmd.Process.Kill()
|
||||
}
|
||||
h.registry.UnregisterBySource(core.SourcePlugin, id)
|
||||
_ = h.st.RemovePlugin(id)
|
||||
if onCommands != nil {
|
||||
_ = onCommands(cmdOverlay)
|
||||
}
|
||||
if onIntents != nil {
|
||||
_ = onIntents(overlay)
|
||||
}
|
||||
if onWorkflows != nil {
|
||||
_ = onWorkflows(defs, scripts)
|
||||
}
|
||||
}
|
||||
|
||||
var _ = strings.TrimSpace
|
||||
|
||||
@@ -1,13 +1,17 @@
|
||||
package plugins
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"agent-system-backend/internal/intentgate"
|
||||
"agent-system-backend/internal/settings"
|
||||
"agent-system-backend/internal/store"
|
||||
"agent-system-backend/internal/tools"
|
||||
"agent-system-backend/internal/workflow"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
@@ -155,3 +159,52 @@ func TestWorkspacesOverlay(t *testing.T) {
|
||||
t.Fatalf("workspaces: %+v", ws)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnregisterDropsWorkflowsAndIntents(t *testing.T) {
|
||||
host, _, dir := newHost(t)
|
||||
root := filepath.Join(dir, "example")
|
||||
os.MkdirAll(filepath.Join(root, "workflows", "scripts"), 0o755)
|
||||
os.MkdirAll(filepath.Join(root, "intents"), 0o755)
|
||||
os.WriteFile(filepath.Join(root, "plugin.json"), []byte(`{"id":"example","name":"示例","version":"0.1.0","apiVersion":"1.0","provides":{"workflows":["demo-brief"],"intents":["demo"]}}`), 0o644)
|
||||
os.WriteFile(filepath.Join(root, "workflows", "demo-brief.json"), []byte(`{"id":"demo.brief","title":"简报","intentId":"demo.brief","steps":[{"id":"run","title":"脚本","kind":"script","script":"workflows/scripts/demo.py"}]}`), 0o644)
|
||||
os.WriteFile(filepath.Join(root, "workflows", "scripts", "demo.py"), []byte("print('ok')\n"), 0o644)
|
||||
os.WriteFile(filepath.Join(root, "intents", "demo.json"), []byte(`[{"id":"demo.brief","domain":"demo","version":1,"matcher":{"includes_any":["简报"]}}]`), 0o644)
|
||||
|
||||
reg := workflow.NewRegistry()
|
||||
gate := intentgate.New()
|
||||
host.OnWorkflows = func(defs []core.WorkflowDef, scripts map[string]string) error {
|
||||
reg.Set(defs, scripts)
|
||||
return nil
|
||||
}
|
||||
host.OnIntents = func(overlay []json.RawMessage) error {
|
||||
return gate.Rebuild(context.Background(), overlay)
|
||||
}
|
||||
if err := host.LoadDir()["example"]; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := reg.ByIntent("demo.brief"); !ok || !gate.Enabled() {
|
||||
t.Fatal("workflow and intent should be registered")
|
||||
}
|
||||
body, ok := reg.Script("demo.brief", "workflows/scripts/demo.py")
|
||||
if !ok || body != "print('ok')\n" {
|
||||
t.Fatalf("script body %q", body)
|
||||
}
|
||||
host.Unregister("example")
|
||||
if _, ok := reg.ByIntent("demo.brief"); ok {
|
||||
t.Fatal("unregister left the workflow")
|
||||
}
|
||||
if gate.Enabled() {
|
||||
t.Fatal("unregister left the intent")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowScriptCannotEscape(t *testing.T) {
|
||||
host, _, dir := newHost(t)
|
||||
root := filepath.Join(dir, "example")
|
||||
os.MkdirAll(filepath.Join(root, "workflows"), 0o755)
|
||||
os.WriteFile(filepath.Join(root, "plugin.json"), []byte(`{"id":"example","name":"示例","version":"0.1.0","apiVersion":"1.0","provides":{"workflows":["bad"]}}`), 0o644)
|
||||
os.WriteFile(filepath.Join(root, "workflows", "bad.json"), []byte(`{"id":"bad","title":"坏","steps":[{"id":"run","title":"逃","kind":"script","script":"../secret.py"}]}`), 0o644)
|
||||
if err := host.LoadDir()["example"]; err == nil {
|
||||
t.Fatal("script path outside the plugin dir should be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -30,6 +30,7 @@ import (
|
||||
"agent-system-backend/internal/settings"
|
||||
"agent-system-backend/internal/store"
|
||||
"agent-system-backend/internal/tools"
|
||||
"agent-system-backend/internal/workflow"
|
||||
"agent-system-backend/web"
|
||||
core "agent-system-core"
|
||||
)
|
||||
@@ -42,10 +43,11 @@ type Server struct {
|
||||
Limiter ratelimit.Limiter // 租户限流(nil = 不限;组合根注入)
|
||||
LimitLookup func() int // 当前限流值(次/分;<=0 不限;组合根缓存注入)
|
||||
mgr *runmgr.Manager
|
||||
Cards *cardsvc.Service // 收件箱任务卡(nil = 不启用;组合根注入)
|
||||
Bus *cmdbus.Bus // 命令总线(nil = 不启用;组合根注入;四入口同源 D27)
|
||||
Gate NLIntentGate // 意图闸门(chat 入口同源判定;nil = 不接)
|
||||
Prompts *prompts.Service // 提示词库(nil = 不启用 API 与技能命令;执行器另有回退)
|
||||
Cards *cardsvc.Service // 收件箱任务卡(nil = 不启用;组合根注入)
|
||||
Bus *cmdbus.Bus // 命令总线(nil = 不启用;组合根注入;四入口同源 D27)
|
||||
Gate NLIntentGate // 意图闸门(chat 入口同源判定;nil = 不接)
|
||||
Workflows *workflow.Registry // 已装载工作流(nil = 不接)
|
||||
Prompts *prompts.Service // 提示词库(nil = 不启用 API 与技能命令;执行器另有回退)
|
||||
reg *tools.Registry
|
||||
prof *profiles.Service
|
||||
set *settings.Service
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
// Package workflow 保存已装载的工作流定义。它只依赖 core,执行器在 runtime。
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sync"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
// ScriptKey 是脚本正文的索引。同一个相对路径可以出现在不同工作流里。
|
||||
func ScriptKey(workflowID, path string) string {
|
||||
return workflowID + "\n" + path
|
||||
}
|
||||
|
||||
// Registry 是当前装载集合的快照。Set 整体替换,不增量合并。
|
||||
type Registry struct {
|
||||
mu sync.RWMutex
|
||||
byID map[string]core.WorkflowDef
|
||||
byIntent map[string]string
|
||||
scripts map[string]string
|
||||
}
|
||||
|
||||
func NewRegistry() *Registry {
|
||||
return &Registry{
|
||||
byID: map[string]core.WorkflowDef{},
|
||||
byIntent: map[string]string{},
|
||||
scripts: map[string]string{},
|
||||
}
|
||||
}
|
||||
|
||||
// Set 用这一批定义替换注册表。同 ID 或同意图以后出现的为准。
|
||||
// scripts 的键是 ScriptKey。定义会被拷贝,调用方之后的修改不影响已注册的副本。
|
||||
func (r *Registry) Set(defs []core.WorkflowDef, scripts map[string]string) {
|
||||
byID := map[string]core.WorkflowDef{}
|
||||
byIntent := map[string]string{}
|
||||
for _, def := range defs {
|
||||
cloned := cloneDef(def)
|
||||
byID[cloned.ID] = cloned
|
||||
if cloned.IntentID != "" {
|
||||
byIntent[cloned.IntentID] = cloned.ID
|
||||
}
|
||||
}
|
||||
copied := map[string]string{}
|
||||
for k, v := range scripts {
|
||||
copied[k] = v
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.byID, r.byIntent, r.scripts = byID, byIntent, copied
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *Registry) Get(id string) (core.WorkflowDef, bool) {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
def, ok := r.byID[id]
|
||||
if !ok {
|
||||
return core.WorkflowDef{}, false
|
||||
}
|
||||
return cloneDef(def), true
|
||||
}
|
||||
|
||||
func (r *Registry) ByIntent(intentID string) (core.WorkflowDef, bool) {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
id, ok := r.byIntent[intentID]
|
||||
if !ok {
|
||||
return core.WorkflowDef{}, false
|
||||
}
|
||||
return cloneDef(r.byID[id]), true
|
||||
}
|
||||
|
||||
func (r *Registry) Script(workflowID, path string) (string, bool) {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
body, ok := r.scripts[ScriptKey(workflowID, path)]
|
||||
return body, ok
|
||||
}
|
||||
|
||||
func cloneDef(def core.WorkflowDef) core.WorkflowDef {
|
||||
raw, err := json.Marshal(def)
|
||||
if err != nil {
|
||||
return def
|
||||
}
|
||||
var out core.WorkflowDef
|
||||
if err := json.Unmarshal(raw, &out); err != nil {
|
||||
return def
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"regexp"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
// Vars 是一次执行里可以填进模板的值。
|
||||
type Vars struct {
|
||||
Input string
|
||||
Slots map[string]string
|
||||
Steps map[string]string
|
||||
}
|
||||
|
||||
var placeholder = regexp.MustCompile(`\{\{(input|slots\.[A-Za-z_][A-Za-z0-9_]*|steps\.[A-Za-z_][A-Za-z0-9_]*\.output)\}\}`)
|
||||
|
||||
// Render 替换 {{input}}、{{slots.<key>}}、{{steps.<id>.output}}。缺值替换成空串。
|
||||
func Render(s string, v Vars) string {
|
||||
return placeholder.ReplaceAllStringFunc(s, func(token string) string {
|
||||
body := token[2 : len(token)-2]
|
||||
switch {
|
||||
case body == "input":
|
||||
return v.Input
|
||||
case len(body) > 6 && body[:6] == "slots.":
|
||||
return v.Slots[body[6:]]
|
||||
case len(body) > 6 && body[:6] == "steps.":
|
||||
id := body[6 : len(body)-len(".output")]
|
||||
return v.Steps[id]
|
||||
default:
|
||||
return token
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// RenderRaw 在 JSON 字符串内部替换占位符,结果仍是合法 JSON。
|
||||
func RenderRaw(raw json.RawMessage, v Vars) (json.RawMessage, error) {
|
||||
if len(raw) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
var node any
|
||||
if err := json.Unmarshal(raw, &node); err != nil {
|
||||
return nil, fmt.Errorf("workflow: 参数不是合法 JSON")
|
||||
}
|
||||
out, err := json.Marshal(walk(node, v))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func walk(node any, v Vars) any {
|
||||
switch x := node.(type) {
|
||||
case string:
|
||||
return Render(x, v)
|
||||
case []any:
|
||||
for i := range x {
|
||||
x[i] = walk(x[i], v)
|
||||
}
|
||||
return x
|
||||
case map[string]any:
|
||||
for k, val := range x {
|
||||
x[k] = walk(val, v)
|
||||
}
|
||||
return x
|
||||
default:
|
||||
return node
|
||||
}
|
||||
}
|
||||
|
||||
// RenderStep 渲染这一步会交给执行器的文本和参数。
|
||||
func RenderStep(step core.WorkflowStep, v Vars) (core.WorkflowStep, error) {
|
||||
step.Prompt = Render(step.Prompt, v)
|
||||
step.Artifact = Render(step.Artifact, v)
|
||||
if step.Hitl != nil {
|
||||
step.Hitl.Question = Render(step.Hitl.Question, v)
|
||||
}
|
||||
if step.Card != nil {
|
||||
step.Card.Title = Render(step.Card.Title, v)
|
||||
}
|
||||
args, err := RenderRaw(step.Args, v)
|
||||
if err != nil {
|
||||
return step, err
|
||||
}
|
||||
step.Args = args
|
||||
return step, nil
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
// 声明式工作流与模型生成的工作流用两套上限。
|
||||
const (
|
||||
MaxDeclarativeSteps = 20
|
||||
MaxDynamicSteps = 8
|
||||
MaxStepRetries = 3
|
||||
MaxWorkflowRetries = 6
|
||||
)
|
||||
|
||||
// DeniedTools 是工作流步骤和脚本回调都不能调用的工具。
|
||||
var DeniedTools = map[string]struct{}{
|
||||
"run_python": {},
|
||||
"run_shell": {},
|
||||
"run_workflow": {},
|
||||
"delegate_task": {},
|
||||
"squad_orchestrate": {},
|
||||
"review_findings": {},
|
||||
}
|
||||
|
||||
// Limits 区分声明式和动态工作流。
|
||||
type Limits struct {
|
||||
MaxSteps int
|
||||
AllowScript bool
|
||||
}
|
||||
|
||||
func DeclarativeLimits() Limits {
|
||||
return Limits{MaxSteps: MaxDeclarativeSteps, AllowScript: true}
|
||||
}
|
||||
|
||||
func DynamicLimits() Limits {
|
||||
return Limits{MaxSteps: MaxDynamicSteps, AllowScript: false}
|
||||
}
|
||||
|
||||
var stepKinds = map[string]struct{}{
|
||||
"prompt": {}, "tool": {}, "hitl": {}, "artifact": {}, "card": {}, "script": {},
|
||||
}
|
||||
|
||||
// Validate 检查定义是否能执行。known 报告工具是否已注册;nil 时只检查禁用名单。
|
||||
func Validate(def core.WorkflowDef, lim Limits, known func(name string) bool) error {
|
||||
if strings.TrimSpace(def.ID) == "" {
|
||||
return fmt.Errorf("workflow: id 为空")
|
||||
}
|
||||
if lim.MaxSteps <= 0 {
|
||||
return fmt.Errorf("workflow: %s: 未设置步骤上限", def.ID)
|
||||
}
|
||||
if n := len(def.Steps); n == 0 || n > lim.MaxSteps {
|
||||
return fmt.Errorf("workflow: %s: 步骤数 %d 超出上限 %d", def.ID, n, lim.MaxSteps)
|
||||
}
|
||||
if def.MaxRetries > MaxWorkflowRetries {
|
||||
return fmt.Errorf("workflow: %s: maxRetries %d 超过 %d", def.ID, def.MaxRetries, MaxWorkflowRetries)
|
||||
}
|
||||
seen := map[string]struct{}{}
|
||||
for _, step := range def.Steps {
|
||||
if strings.TrimSpace(step.ID) == "" {
|
||||
return fmt.Errorf("workflow: %s: 步骤 id 为空", def.ID)
|
||||
}
|
||||
if _, ok := seen[step.ID]; ok {
|
||||
return fmt.Errorf("workflow: %s: 步骤 id %s 重复", def.ID, step.ID)
|
||||
}
|
||||
seen[step.ID] = struct{}{}
|
||||
if _, ok := stepKinds[step.Kind]; !ok {
|
||||
return fmt.Errorf("workflow: %s: 步骤 %s 的 kind %q 不合法", def.ID, step.ID, step.Kind)
|
||||
}
|
||||
if step.Critic != nil && step.Critic.MaxRetries > MaxStepRetries {
|
||||
return fmt.Errorf("workflow: %s: 步骤 %s 的重试上限 %d 超过 %d", def.ID, step.ID, step.Critic.MaxRetries, MaxStepRetries)
|
||||
}
|
||||
switch step.Kind {
|
||||
case "tool":
|
||||
if err := checkTool(def.ID, step, known); err != nil {
|
||||
return err
|
||||
}
|
||||
case "script":
|
||||
if !lim.AllowScript {
|
||||
return fmt.Errorf("workflow: %s: 动态工作流不能包含 script 步骤 %s", def.ID, step.ID)
|
||||
}
|
||||
if strings.TrimSpace(step.Script) == "" || !scriptPathOK(step.Script) {
|
||||
return fmt.Errorf("workflow: %s: 步骤 %s 的脚本路径不合法", def.ID, step.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkTool(workflowID string, step core.WorkflowStep, known func(name string) bool) error {
|
||||
name := strings.TrimSpace(step.Tool)
|
||||
if name == "" {
|
||||
return fmt.Errorf("workflow: %s: 步骤 %s 缺少工具名", workflowID, step.ID)
|
||||
}
|
||||
if _, denied := DeniedTools[name]; denied {
|
||||
return fmt.Errorf("workflow: %s: 步骤 %s 不能调用 %s", workflowID, step.ID, name)
|
||||
}
|
||||
if known != nil && !known(name) {
|
||||
return fmt.Errorf("workflow: %s: 步骤 %s 的工具 %s 未注册", workflowID, step.ID, name)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func scriptPathOK(path string) bool {
|
||||
if path == "" || strings.HasPrefix(path, "/") || strings.Contains(path, "\\") {
|
||||
return false
|
||||
}
|
||||
if path == ".." || strings.HasPrefix(path, "../") || strings.Contains(path, "/../") {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
func known(names ...string) func(string) bool {
|
||||
set := map[string]struct{}{}
|
||||
for _, n := range names {
|
||||
set[n] = struct{}{}
|
||||
}
|
||||
return func(name string) bool {
|
||||
_, ok := set[name]
|
||||
return ok
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateRejectsBadDefinitions(t *testing.T) {
|
||||
okTool := known("time_now")
|
||||
cases := []struct {
|
||||
name string
|
||||
def core.WorkflowDef
|
||||
lim Limits
|
||||
want string
|
||||
}{
|
||||
{"dup", core.WorkflowDef{ID: "a", Steps: []core.WorkflowStep{{ID: "s", Kind: "hitl"}, {ID: "s", Kind: "card"}}}, DeclarativeLimits(), "重复"},
|
||||
{"kind", core.WorkflowDef{ID: "a", Steps: []core.WorkflowStep{{ID: "s", Kind: "nope"}}}, DeclarativeLimits(), "不合法"},
|
||||
{"denied", core.WorkflowDef{ID: "a", Steps: []core.WorkflowStep{{ID: "s", Kind: "tool", Tool: "run_python"}}}, DeclarativeLimits(), "不能调用"},
|
||||
{"missing", core.WorkflowDef{ID: "a", Steps: []core.WorkflowStep{{ID: "s", Kind: "tool", Tool: "no_such"}}}, DeclarativeLimits(), "未注册"},
|
||||
{"over", core.WorkflowDef{ID: "a", Steps: manySteps(21)}, DeclarativeLimits(), "超出上限"},
|
||||
{"dynamic", core.WorkflowDef{ID: "a", Steps: manySteps(9)}, DynamicLimits(), "超出上限"},
|
||||
{"script", core.WorkflowDef{ID: "a", Steps: []core.WorkflowStep{{ID: "s", Kind: "script", Script: "scripts/demo.py"}}}, DynamicLimits(), "不能包含 script"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
err := Validate(tc.def, tc.lim, okTool)
|
||||
if err == nil || !strings.Contains(err.Error(), tc.want) {
|
||||
t.Fatalf("got %v, want substring %q", err, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
good := core.WorkflowDef{ID: "a", MaxRetries: 6, Steps: []core.WorkflowStep{
|
||||
{ID: "s", Kind: "tool", Tool: "time_now"},
|
||||
{ID: "p", Kind: "script", Script: "scripts/demo.py", Critic: &core.StepCritic{Criteria: "有结果", MaxRetries: 3}},
|
||||
}}
|
||||
if err := Validate(good, DeclarativeLimits(), okTool); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func manySteps(n int) []core.WorkflowStep {
|
||||
out := make([]core.WorkflowStep, n)
|
||||
for i := range out {
|
||||
out[i] = core.WorkflowStep{ID: "s" + strconv.Itoa(i), Kind: "hitl"}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestRenderKeepsArgsJSON(t *testing.T) {
|
||||
raw := json.RawMessage(`{"q":"关于 {{slots.topic}}","note":"{{input}}"}`)
|
||||
got, err := RenderRaw(raw, Vars{Input: `他说 "好"`, Slots: map[string]string{"topic": "主题"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m map[string]string
|
||||
if err := json.Unmarshal(got, &m); err != nil {
|
||||
t.Fatalf("not json: %s (%v)", got, err)
|
||||
}
|
||||
if m["q"] != "关于 主题" || m["note"] != `他说 "好"` {
|
||||
t.Fatalf("%s", got)
|
||||
}
|
||||
text := Render("{{steps.draft.output}} / {{slots.missing}}", Vars{Steps: map[string]string{"draft": "正文"}})
|
||||
if text != "正文 / " {
|
||||
t.Fatalf("%q", text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestByIntentAndReplace(t *testing.T) {
|
||||
reg := NewRegistry()
|
||||
def := core.WorkflowDef{ID: "demo.brief", IntentID: "demo.brief", Steps: []core.WorkflowStep{{ID: "s", Kind: "hitl"}}}
|
||||
reg.Set([]core.WorkflowDef{def}, map[string]string{ScriptKey("demo.brief", "scripts/demo.py"): "print(1)"})
|
||||
got, ok := reg.ByIntent("demo.brief")
|
||||
if !ok || got.ID != "demo.brief" {
|
||||
t.Fatalf("%+v %v", got, ok)
|
||||
}
|
||||
body, ok := reg.Script("demo.brief", "scripts/demo.py")
|
||||
if !ok || body != "print(1)" {
|
||||
t.Fatalf("script %q %v", body, ok)
|
||||
}
|
||||
reg.Set(nil, nil)
|
||||
if _, ok := reg.ByIntent("demo.brief"); ok {
|
||||
t.Fatal("cleared registry still has the workflow")
|
||||
}
|
||||
if _, ok := reg.Get("demo.brief"); ok {
|
||||
t.Fatal("cleared registry still has the id")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user