diff --git a/deps/agent-system-core b/deps/agent-system-core index 9b8cf79..f93f473 160000 --- a/deps/agent-system-core +++ b/deps/agent-system-core @@ -1 +1 @@ -Subproject commit 9b8cf79c96668628926d32046cd650f0cfbd285b +Subproject commit f93f473a0ea898785921e3c4fb16114549fd2783 diff --git a/internal/app/app.go b/internal/app/app.go index 3a9429b..11daac3 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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,8 @@ func Run(opt Options) error { } srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost) + srv.Workflows = wfReg + mgr.SetWorkflows(wfReg) srv.Cards = cardSvc srv.Bus = bus srv.Gate = gate diff --git a/internal/intentgate/gate.go b/internal/intentgate/gate.go index a54dec2..b7c7141 100644 --- a/internal/intentgate/gate.go +++ b/internal/intentgate/gate.go @@ -12,11 +12,26 @@ import ( agentintent "agent-intent/intent" ) +// SlotQuestion 是命中意图后仍缺的必填槽位,以及建议的提问。 +type SlotQuestion struct { + Key string + Question string +} + +// MatchResult 是一次识别。Branch 与 Match 相同。 +type MatchResult struct { + Branch string + ID string + Slots map[string]string + Missing []SlotQuestion +} + type Gate struct { - mu sync.RWMutex - reg *agentintent.Registry - rec *agentintent.FunnelRecognizer - specs int + mu sync.RWMutex + reg *agentintent.Registry + rec *agentintent.FunnelRecognizer + specs int + specsByID map[string]agentintent.IntentSpec } func New() *Gate { @@ -27,7 +42,7 @@ func New() *Gate { func (g *Gate) Rebuild(ctx context.Context, overlay []json.RawMessage) error { if len(overlay) == 0 { g.mu.Lock() - g.reg, g.specs = nil, 0 + g.reg, g.specs, g.specsByID = nil, 0, nil g.mu.Unlock() return nil } @@ -47,8 +62,12 @@ func (g *Gate) Rebuild(ctx context.Context, overlay []json.RawMessage) error { if err != nil { return fmt.Errorf("intentgate: build: %w", err) } + byID := make(map[string]agentintent.IntentSpec, len(specs)) + for _, s := range specs { + byID[s.ID] = s + } g.mu.Lock() - g.reg, g.specs = reg, len(specs) + g.reg, g.specs, g.specsByID = reg, len(specs), byID g.mu.Unlock() return nil } @@ -56,28 +75,50 @@ func (g *Gate) Rebuild(ctx context.Context, overlay []json.RawMessage) error { // Match 识别自由输入;返回 (branch, topID)。 // branch:matched | ambiguous | fallback | ""(未配置/系统级错误 → 回退通道)。 func (g *Gate) Match(ctx context.Context, input string) (string, string) { + res := g.Recognize(ctx, input) + return res.Branch, res.ID +} + +// Recognize 在 Match 的结果上带上槽位和缺失提问。 +func (g *Gate) Recognize(ctx context.Context, input string) MatchResult { g.mu.RLock() - reg, rec := g.reg, g.rec + reg, rec, byID := g.reg, g.rec, g.specsByID g.mu.RUnlock() if reg == nil { - return "", "" + return MatchResult{} } res, err := rec.Recognize(ctx, input, reg.Snapshot()) if err != nil { - return "", "" - } - topID := "" - if res.Top != nil { - topID = res.Top.ID + return MatchResult{} } + out := MatchResult{Branch: "fallback"} switch res.Branch { case agentintent.BranchMatched: - return "matched", topID + out.Branch = "matched" case agentintent.BranchAmbiguous: - return "ambiguous", topID - default: - return "fallback", topID + out.Branch = "ambiguous" } + if res.Top == nil { + return out + } + out.ID = res.Top.ID + if len(res.Top.Slots) > 0 { + out.Slots = map[string]string{} + for k, v := range res.Top.Slots { + out.Slots[k] = v + } + } + spec := byID[out.ID] + for _, key := range res.Top.MissingRequiredSlots { + q := key + for _, sl := range spec.Slots { + if sl.Key == key && sl.Question != "" { + q = sl.Question + } + } + out.Missing = append(out.Missing, SlotQuestion{Key: key, Question: q}) + } + return out } // Enabled 报告是否装载了意图(诊断/设置展示用)。 diff --git a/internal/intentgate/gate_test.go b/internal/intentgate/gate_test.go index 0c0c917..813f41b 100644 --- a/internal/intentgate/gate_test.go +++ b/internal/intentgate/gate_test.go @@ -65,3 +65,29 @@ func TestGateFailClosed(t *testing.T) { t.Error("empty rebuild should disable gate") } } + +func TestRecognizeKeepsSlots(t *testing.T) { + g := New() + spec := raw(t, map[string]any{ + "id": "demo.brief", "domain": "demo", "version": 1, + "matcher": map[string]any{"includes_any": []string{"简报"}}, + "slots": []map[string]any{{ + "key": "topic", "required": true, "pattern": `主题[::]\s*(\S+)`, + "question": "主题是什么?", + }}, + }) + if err := g.Rebuild(context.Background(), []json.RawMessage{spec}); err != nil { + t.Fatal(err) + } + miss := g.Recognize(context.Background(), "写一份简报") + if miss.Branch != "matched" || miss.ID != "demo.brief" || len(miss.Missing) != 1 || miss.Missing[0].Question != "主题是什么?" { + t.Fatalf("%+v", miss) + } + hit := g.Recognize(context.Background(), "主题:核能 写一份简报") + if hit.Slots["topic"] != "核能" || len(hit.Missing) != 0 { + t.Fatalf("%+v", hit) + } + if branch, id := g.Match(context.Background(), "写一份简报"); branch != "matched" || id != "demo.brief" { + t.Fatalf("Match wrapper = %s %s", branch, id) + } +} diff --git a/internal/plugins/host.go b/internal/plugins/host.go index a59b646..99f55a3 100644 --- a/internal/plugins/host.go +++ b/internal/plugins/host.go @@ -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[""] = err } } + if onWorkflows != nil { + if err := onWorkflows(defs, scripts); err != nil { + results[""] = 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 diff --git a/internal/plugins/host_test.go b/internal/plugins/host_test.go index 9af296e..aed4787 100644 --- a/internal/plugins/host_test.go +++ b/internal/plugins/host_test.go @@ -1,13 +1,18 @@ package plugins import ( + "context" + "encoding/json" "os" "path/filepath" + "strings" "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 +160,95 @@ 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") + } +} + +func TestRepoExamplePluginLoads(t *testing.T) { + root := filepath.Join("..", "..", "..", "plugins") + if _, err := os.Stat(filepath.Join(root, "example", "workflows", "demo-brief.json")); err != nil { + t.Skip("example plugin is not beside this module") + } + dir := t.TempDir() + st, err := store.OpenSQLite(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + regTools := tools.NewRegistry() + regTools.RegisterBuiltin() + host := New(st, regTools, settings.NewService(st), root) + gate := intentgate.New() + wf := workflow.NewRegistry() + host.OnIntents = func(overlay []json.RawMessage) error { + return gate.Rebuild(context.Background(), overlay) + } + host.OnWorkflows = func(defs []core.WorkflowDef, scripts map[string]string) error { + wf.Set(defs, scripts) + return nil + } + if err := host.LoadDir()["example"]; err != nil { + t.Fatal(err) + } + if _, ok := wf.ByIntent("demo.brief"); !ok { + t.Fatal("demo.brief is not bound") + } + body, ok := wf.Script("demo.script", "workflows/scripts/demo.py") + if !ok || !strings.Contains(body, `wf.tool("time_now"`) { + t.Fatalf("script %q", body) + } + got := gate.Recognize(context.Background(), "写一份简报") + if got.ID != "demo.brief" || len(got.Missing) != 1 || got.Missing[0].Question != "主题是什么?" { + t.Fatalf("%+v", got) + } + filled := gate.Recognize(context.Background(), "主题:核能 写一份简报") + if filled.Slots["topic"] != "核能" || len(filled.Missing) != 0 { + t.Fatalf("%+v", filled) + } +} diff --git a/internal/prompts/library/agent-sub-workflow-plan.md b/internal/prompts/library/agent-sub-workflow-plan.md new file mode 100644 index 0000000..c7b7cd9 --- /dev/null +++ b/internal/prompts/library/agent-sub-workflow-plan.md @@ -0,0 +1,29 @@ +--- +id: agent.sub.workflow-plan +layer: agent +version: 1 +title: 工作流规划子代理 +description: 把目标写成可校验的工作流 JSON,不执行步骤 +tools: knowledge_search, time_now +model: lightweight +maxTurns: 4 +--- +你是工作流规划子代理。用户给你的是待规划的目标,不是要你现在完成的任务。不要调用工具,不要回答目标本身。只输出一个 JSON 对象,不要解释,不要写脚本。 + +对象字段: +- id:短标识,形如 plan.brief +- title:给人看的标题 +- steps:数组,最多 8 步 + +每步字段: +- id:步内唯一 +- title:这一步做什么 +- kind:只能是 prompt、tool、hitl、artifact、card +- prompt:prompt 或 artifact 步骤的文本,可用 {{input}}、{{slots.名称}}、{{steps.步骤id.output}} +- agent:prompt 步骤的子代理,缺省 general,只能是 general、explore、research、plan、verify、writer +- tool 与 args:tool 步骤。不能使用 run_python、run_shell、run_workflow、delegate_task、squad_orchestrate、review_findings +- hitl:hitl 步骤的提问,含 question、kind、choices +- artifact:artifact 步骤的文件名 +- card:card 步骤的 kind 与 title + +禁止 script 步骤。拿不准的步骤用 hitl 问用户。 diff --git a/internal/prompts/prompts_test.go b/internal/prompts/prompts_test.go index 1cc90af..e5d7db7 100644 --- a/internal/prompts/prompts_test.go +++ b/internal/prompts/prompts_test.go @@ -442,6 +442,7 @@ func TestSubagentOverrideRejected(t *testing.T) { {ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Tools: []string{"delegate_task"}}, {ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Tools: []string{"squad_orchestrate"}}, {ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Tools: []string{"review_findings"}}, + {ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Tools: []string{"run_workflow"}}, } for _, m := range cases { if err := svc.Save("alice", m); err == nil { diff --git a/internal/prompts/service.go b/internal/prompts/service.go index c931af8..abbeeae 100644 --- a/internal/prompts/service.go +++ b/internal/prompts/service.go @@ -183,7 +183,7 @@ func (s *Service) Get(owner, id string) (core.PromptModule, bool, error) { } var dispatchTools = map[string]struct{}{ - "delegate_task": {}, "squad_orchestrate": {}, "review_findings": {}, + "delegate_task": {}, "squad_orchestrate": {}, "review_findings": {}, "run_workflow": {}, } // validateSubagentOverride 只允许 agent.sub.* 携带 tools/model/maxTurns, diff --git a/internal/runmgr/entry_test.go b/internal/runmgr/entry_test.go index 2da2857..33e52c5 100644 --- a/internal/runmgr/entry_test.go +++ b/internal/runmgr/entry_test.go @@ -4,6 +4,7 @@ import ( "testing" "agent-system-backend/internal/runtime" + "agent-system-backend/internal/workflow" core "agent-system-core" ) @@ -13,16 +14,19 @@ func (n nameRunner) Name() string { return n.name } func (n nameRunner) Execute(rc runtime.RunCtx) error { return nil } func TestSelectConversationRunner(t *testing.T) { - m := &Manager{runner: nameRunner{name: "direct"}, mock: nameRunner{name: "mock"}, squad: nameRunner{name: "squad"}} - if got := selectConversationRunner(m, core.EffectiveSettings{}); got.Name() != "direct" { + m := &Manager{runner: nameRunner{name: "direct"}, mock: nameRunner{name: "mock"}, squad: nameRunner{name: "squad"}, wfRunner: nameRunner{name: "workflow"}} + if got := selectConversationRunner(m, core.EffectiveSettings{}, nil); got.Name() != "direct" { t.Fatalf("default = %s", got.Name()) } squadEff := core.EffectiveSettings{"runtime": map[string]any{"entry": "squad"}} - if got := selectConversationRunner(m, squadEff); got.Name() != "squad" { + if got := selectConversationRunner(m, squadEff, nil); got.Name() != "squad" { t.Fatalf("squad entry = %s", got.Name()) } mockEff := core.EffectiveSettings{"runtime": map[string]any{"entry": "squad", "mock": true}} - if got := selectConversationRunner(m, mockEff); got.Name() != "mock" { + if got := selectConversationRunner(m, mockEff, nil); got.Name() != "mock" { t.Fatalf("mock wins = %s", got.Name()) } + if got := selectConversationRunner(m, mockEff, &workflow.Launch{}); got.Name() != "workflow" { + t.Fatalf("workflow wins = %s", got.Name()) + } } diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 93aca22..f4cdaeb 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -20,6 +20,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" ) @@ -29,11 +30,13 @@ type Manager struct { registry *tools.Registry profiles *profiles.Service settings *settings.Service - runner runtime.Runner // conversation runner(mock 或 LLM) - subRunner runtime.Runner // subagent runner(LLM 轻任务) - mock runtime.Runner // 会话脚本回放(演示/测试与 mock 模式) - mockSub runtime.Runner // 子代理脚本回放 - squad runtime.Runner // runtime.entry=squad 时的对话入口 + runner runtime.Runner // conversation runner(mock 或 LLM) + subRunner runtime.Runner // subagent runner(LLM 轻任务) + mock runtime.Runner // 会话脚本回放(演示/测试与 mock 模式) + mockSub runtime.Runner // 子代理脚本回放 + squad runtime.Runner // runtime.entry=squad 时的对话入口 + wfRunner runtime.Runner // 意图或工具启动的工作流;nil 时用 runtime.WorkflowRunner + workflows *workflow.Registry // run_workflow 按 id 查找已注册定义 gate IntentGate cards *cardsvc.Service // 可选意图闸门(nil = 未启用,行为回退) // OnUsage 在 run 结束时报告 token(日预算记账;nil = 不记)。 @@ -147,6 +150,31 @@ type runCtx struct { subTask string subHint string subAgent string + launch *workflow.Launch +} + +// Workflow 返回这次 run 启动时拷贝的工作流。普通回合没有。 +func (c *runCtx) Workflow() (workflow.Launch, bool) { + if c.launch == nil { + return workflow.Launch{}, false + } + return *c.launch, true +} + +// WorkflowByID 按 id 取当前注册表里的定义。运行中的副本不受之后卸载的影响。 +func (c *runCtx) WorkflowByID(id string) (core.WorkflowDef, bool) { + if c.mgr.workflows == nil { + return core.WorkflowDef{}, false + } + return c.mgr.workflows.Get(id) +} + +// WorkflowScript 读取插件脚本的当前正文。启动工作流时会另外拷进 Launch。 +func (c *runCtx) WorkflowScript(id, path string) (string, bool) { + if c.mgr.workflows == nil { + return "", false + } + return c.mgr.workflows.Script(id, path) } func (c *runCtx) Context() context.Context { return c.ctx } @@ -693,8 +721,21 @@ func (t *toolCtx) PeerConversationMessages(conversationID string, limit int) ([] // SetSquadRunner 注入 runtime.entry=squad 时使用的对话入口。 func (m *Manager) SetSquadRunner(r runtime.Runner) { m.squad = r } -// selectConversationRunner 按设置选择入口。mock 优先;squad 仅在注入了入口时生效。 -func selectConversationRunner(m *Manager, eff core.EffectiveSettings) runtime.Runner { +// SetWorkflowRunner 注入工作流执行器。测试用来替换默认实现。 +func (m *Manager) SetWorkflowRunner(r runtime.Runner) { m.wfRunner = r } + +// SetWorkflows 注入已装载工作流,供 run_workflow 按 id 查找。 +func (m *Manager) SetWorkflows(r *workflow.Registry) { m.workflows = r } + +// selectConversationRunner 按设置选择入口。带工作流定义时走工作流执行器; +// 否则 mock 优先,squad 仅在注入了入口时生效。 +func selectConversationRunner(m *Manager, eff core.EffectiveSettings, launch *workflow.Launch) runtime.Runner { + if launch != nil { + if m.wfRunner != nil { + return m.wfRunner + } + return runtime.WorkflowRunner{} + } if eff.GetBool("runtime.mock", false) { return m.mock } @@ -711,7 +752,42 @@ func (m *Manager) StartConversationRun(conv core.Conversation, userMessage core. if err := m.claimConversation(conv.ID); err != nil { return core.Run{}, err } - return m.startClaimedRun(conv, userMessage, attachments, inputs) + return m.startClaimedRun(conv, userMessage, attachments, inputs, nil) +} + +// StartWorkflowRun 启动一条绑定工作流的会话 run。定义在这里拷贝, +// 之后插件卸载不会改到已经在跑的这份。 +func (m *Manager) StartWorkflowRun(conv core.Conversation, userMessage core.Message, def core.WorkflowDef, slots map[string]string, missing []workflow.SlotQuestion) (core.Run, error) { + if err := m.claimConversation(conv.ID); err != nil { + return core.Run{}, err + } + copiedSlots := map[string]string{} + for k, v := range slots { + copiedSlots[k] = v + } + launch := &workflow.Launch{ + Def: workflow.Clone(def), + Slots: copiedSlots, + Missing: append([]workflow.SlotQuestion(nil), missing...), + Scripts: workflowScriptBodies(m.workflows, def), + } + return m.startClaimedRun(conv, userMessage, nil, nil, launch) +} + +func workflowScriptBodies(reg *workflow.Registry, def core.WorkflowDef) map[string]string { + out := map[string]string{} + if reg == nil { + return out + } + for _, step := range def.Steps { + if step.Kind != "script" || strings.TrimSpace(step.Script) == "" { + continue + } + if body, ok := reg.Script(def.ID, step.Script); ok { + out[step.Script] = body + } + } + return out } func (m *Manager) claimConversation(conversationID string) error { @@ -727,7 +803,7 @@ func (m *Manager) claimConversation(conversationID string) error { return nil } -func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile) (core.Run, error) { +func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile, launch *workflow.Launch) (core.Run, error) { prof, _ := m.profiles.Get(conv.Owner, conv.ProfileID) eff := m.settings.Effective(conv.Owner, conv.ProjectID) run := core.Run{ @@ -744,9 +820,9 @@ func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Messa m.active[run.ID] = h m.mu.Unlock() - runner := selectConversationRunner(m, eff) + runner := selectConversationRunner(m, eff, launch) inputPaths, _ := m.stageInputs(conv, eff, inputs) - rc := &runCtx{mgr: m, run: run, prof: prof, eff: eff, ctx: ctx, h: h, userMessage: userMessage, attachments: attachments, inputPaths: inputPaths} + rc := &runCtx{mgr: m, run: run, prof: prof, eff: eff, ctx: ctx, h: h, userMessage: userMessage, attachments: attachments, inputPaths: inputPaths, launch: launch} go m.execute(rc, runner) return run, nil } @@ -1235,7 +1311,7 @@ func (m *Manager) startNoticeRun(convID string, items []bgNotice) error { m.releaseConversation(convID) return err } - if _, err := m.startClaimedRun(conv, msg, nil, nil); err != nil { + if _, err := m.startClaimedRun(conv, msg, nil, nil, nil); err != nil { m.releaseConversation(convID) return err } diff --git a/internal/runmgr/workflow_route_test.go b/internal/runmgr/workflow_route_test.go new file mode 100644 index 0000000..bdc3634 --- /dev/null +++ b/internal/runmgr/workflow_route_test.go @@ -0,0 +1,132 @@ +package runmgr + +import ( + "path/filepath" + "testing" + "time" + + "agent-system-backend/internal/bus" + "agent-system-backend/internal/profiles" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + "agent-system-backend/internal/tools" + "agent-system-backend/internal/workflow" + core "agent-system-core" +) + +type launchCapture struct { + ch chan workflow.Launch +} + +func (c launchCapture) Name() string { return "workflow-capture" } + +func (c launchCapture) Execute(rc runtime.RunCtx) error { + src, ok := rc.(interface { + Workflow() (workflow.Launch, bool) + }) + if !ok { + return nil + } + launch, ok := src.Workflow() + if ok { + c.ch <- launch + } + return nil +} + +func TestStartWorkflowRunCopiesDefinition(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "wf.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + if err := st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": true}}); err != nil { + t.Fatal(err) + } + reg := tools.NewRegistry() + prof := profiles.New(st) + if err := prof.EnsureDefault(); err != nil { + t.Fatal(err) + } + direct := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "direct"}}} + cap := launchCapture{ch: make(chan workflow.Launch, 1)} + mgr := New(st, bus.New(), reg, prof, settings.NewService(st), direct, noopRunner{}, &runtime.MockRunner{}, &runtime.MockSubagent{}) + mgr.SetWorkflowRunner(cap) + + proj, err := st.CreateProject(store.OwnerDefault, "p") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation(store.OwnerDefault, proj.ID, "c", "") + if err != nil { + t.Fatal(err) + } + msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "写简报"}}} + def := core.WorkflowDef{ID: "demo.brief", Title: "简报", Steps: []core.WorkflowStep{{ID: "s", Kind: "hitl"}}} + if _, err := mgr.StartWorkflowRun(conv, msg, def, map[string]string{"topic": "核能"}, []workflow.SlotQuestion{{Key: "audience", Question: "给谁看?"}}); err != nil { + t.Fatal(err) + } + def.Title = "被改掉" + select { + case launch := <-cap.ch: + if launch.Def.ID != "demo.brief" || launch.Def.Title != "简报" || launch.Slots["topic"] != "核能" { + t.Fatalf("%+v", launch) + } + if len(launch.Missing) != 1 || launch.Missing[0].Question != "给谁看?" { + t.Fatalf("missing %+v", launch.Missing) + } + case <-time.After(time.Second): + t.Fatal("workflow runner was not selected") + } +} + +func TestStartWorkflowRunCopiesScriptBody(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "wf.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + if err := st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": true}}); err != nil { + t.Fatal(err) + } + reg := tools.NewRegistry() + prof := profiles.New(st) + if err := prof.EnsureDefault(); err != nil { + t.Fatal(err) + } + direct := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "direct"}}} + cap := launchCapture{ch: make(chan workflow.Launch, 1)} + mgr := New(st, bus.New(), reg, prof, settings.NewService(st), direct, noopRunner{}, &runtime.MockRunner{}, &runtime.MockSubagent{}) + mgr.SetWorkflowRunner(cap) + wf := workflow.NewRegistry() + const path = "workflows/scripts/demo.py" + wf.Set([]core.WorkflowDef{{ + ID: "demo.script", Title: "脚本", + Steps: []core.WorkflowStep{{ID: "run", Kind: "script", Script: path}}, + }}, map[string]string{workflow.ScriptKey("demo.script", path): "print(1)\n"}) + mgr.SetWorkflows(wf) + + proj, err := st.CreateProject(store.OwnerDefault, "p") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation(store.OwnerDefault, proj.ID, "c", "") + if err != nil { + t.Fatal(err) + } + msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "跑"}}} + def, _ := wf.Get("demo.script") + if _, err := mgr.StartWorkflowRun(conv, msg, def, nil, nil); err != nil { + t.Fatal(err) + } + wf.Set(nil, nil) + select { + case launch := <-cap.ch: + if launch.Scripts[path] != "print(1)\n" { + t.Fatalf("%+v", launch.Scripts) + } + case <-time.After(time.Second): + t.Fatal("workflow runner was not selected") + } +} diff --git a/internal/runtime/agentwf.py b/internal/runtime/agentwf.py new file mode 100644 index 0000000..d31d477 --- /dev/null +++ b/internal/runtime/agentwf.py @@ -0,0 +1,51 @@ +"""工作流脚本回调。用法:import agentwf as wf。 + +请求写到 fd 3,响应从 fd 4 按行读回。宿主一次只处理一条。 +""" + +import json +import os + +_seq = 0 +_out = os.fdopen(3, "w", encoding="utf-8", buffering=1) +_inp = os.fdopen(4, "r", encoding="utf-8", buffering=1) + + +def _rpc(op, **fields): + global _seq + _seq += 1 + msg = {"id": str(_seq), "op": op} + msg.update(fields) + _out.write(json.dumps(msg, ensure_ascii=False) + "\n") + _out.flush() + line = _inp.readline() + if not line: + raise RuntimeError("workflow host closed") + resp = json.loads(line) + if not resp.get("ok"): + raise RuntimeError(resp.get("error") or "workflow call failed") + return resp.get("output", "") + + +def tool(name, args=None): + return _rpc("tool.call", name=name, args=args or {}) + + +def ask(question, choices=None): + return _rpc("hitl.ask", question=question, choices=choices or []) + + +def agent(name, task): + return _rpc("agent.spawn", agent=name, task=task) + + +def artifact(name, content): + return _rpc("artifact.register", name=name, content=content) + + +def card(title, kind="result"): + return _rpc("card.create", title=title, kind=kind) + + +def step(step_id, title, status): + return _rpc("step", stepId=step_id, title=title, status=status) diff --git a/internal/runtime/llm.go b/internal/runtime/llm.go index 9560ae0..fffdb65 100644 --- a/internal/runtime/llm.go +++ b/internal/runtime/llm.go @@ -46,12 +46,13 @@ func durationMs(start, end time.Time) float64 { } const ( - toolDelegate = "delegate_task" - toolReview = "review_findings" - toolPlan = "update_plan" - toolTodos = "update_todos" - toolSquad = "squad_orchestrate" - toolCard = "submit_card" // 产物卡提交(ui.cardflow 开启时声明;D39 显式路由) + toolDelegate = "delegate_task" + toolReview = "review_findings" + toolPlan = "update_plan" + toolTodos = "update_todos" + toolSquad = "squad_orchestrate" + toolRunWorkflow = "run_workflow" + toolCard = "submit_card" // 产物卡提交(ui.cardflow 开启时声明;D39 显式路由) // 流式产卡三段式(同门控声明;D44/D45):begin 建卡、append 逐段追加、 // commit 收尾——长正文优先走此路径,submit_card 留给一次性短产物。 toolCardBegin = "card_begin" @@ -240,6 +241,13 @@ func (l *LLMRunner) Execute(rc RunCtx) error { agentllm.ToolDecl{Name: toolPlan, Description: l.toolDesc(prof.Owner, toolPlan, "创建或更新当前会话的计划(步骤列表)"), Parameters: json.RawMessage(`{"type":"object","properties":{"title":{"type":"string"},"steps":{"type":"array","items":{"type":"object","properties":{"id":{"type":"string"},"text":{"type":"string"},"status":{"type":"string","enum":["pending","in_progress","done","skipped"]}},"required":["id","text","status"]}}},"required":["title","steps"]}`)}, agentllm.ToolDecl{Name: toolTodos, Description: l.toolDesc(prof.Owner, toolTodos, "更新当前会话的待办清单"), Parameters: json.RawMessage(`{"type":"object","properties":{"items":{"type":"array","items":{"type":"object","properties":{"id":{"type":"string"},"text":{"type":"string"},"status":{"type":"string","enum":["pending","in_progress","done"]}},"required":["id","text","status"]}}},"required":["items"]}`)}, ) + if eff.GetString("runtime.entry", "direct") != "squad" { + decls = append(decls, agentllm.ToolDecl{ + Name: toolRunWorkflow, + Description: l.toolDesc(prof.Owner, toolRunWorkflow, "运行一个工作流。workflow 运行已注册的定义并带上 inputs;goal 让模型先生成步骤再请用户确认;script 在开关打开后运行一段脚本。三者只能填一个"), + Parameters: json.RawMessage(`{"type":"object","properties":{"workflow":{"type":"string","description":"已注册工作流 id"},"inputs":{"type":"object","additionalProperties":{"type":"string"},"description":"槽位初值"},"goal":{"type":"string","description":"用模型生成步骤"},"script":{"type":"string","description":"要运行的脚本"}},"additionalProperties":false}`), + }) + } cardflow := eff.GetBool("ui.cardflow", false) if cardflow { decls = append(decls, agentllm.ToolDecl{ diff --git a/internal/runtime/llm_direct.go b/internal/runtime/llm_direct.go index b492397..660aca5 100644 --- a/internal/runtime/llm_direct.go +++ b/internal/runtime/llm_direct.go @@ -407,6 +407,8 @@ func (st *directRun) runTool(name, args string) agenttools.Result { } else { out = "待办参数错误: " + uerr.Error() } + case toolRunWorkflow: + out = st.runRegisteredWorkflow(call.Arguments) case toolCard: out = st.submitCard(call.Arguments) case toolCardBegin: @@ -452,6 +454,88 @@ func (st *directRun) runTool(name, args string) agenttools.Result { return agenttools.Result{Content: out, IsError: err != nil} } +func (st *directRun) runRegisteredWorkflow(args string) string { + var a struct { + Workflow string `json:"workflow"` + Inputs map[string]string `json:"inputs"` + Goal string `json:"goal"` + Script string `json:"script"` + } + if err := json.Unmarshal([]byte(args), &a); err != nil { + return "run_workflow 参数不是合法 JSON" + } + n := 0 + for _, s := range []string{a.Workflow, a.Goal, a.Script} { + if strings.TrimSpace(s) != "" { + n++ + } + } + if n != 1 { + return "run_workflow 需要且只能指定 workflow、goal、script 之一" + } + if strings.TrimSpace(a.Goal) != "" { + return st.runPlannedWorkflow(strings.TrimSpace(a.Goal)) + } + if strings.TrimSpace(a.Script) != "" { + return st.runScriptWorkflow(a.Script) + } + lookup, ok := st.rc.(interface { + WorkflowByID(id string) (core.WorkflowDef, bool) + }) + if !ok { + return "当前运行不能查找工作流" + } + def, ok := lookup.WorkflowByID(a.Workflow) + if !ok { + return "未注册的工作流 " + a.Workflow + } + res, err := (WorkflowRunner{}).run(st.rc, registeredWorkflowLaunch(st.rc, def, a.Inputs)) + st.pinSubagentArtifacts(res.Summary) + if err != nil { + if res.Summary != "" { + return res.Summary + } + return err.Error() + } + return res.Summary +} + +func (st *directRun) runScriptWorkflow(body string) string { + text, arts, err := runModelScript(st.rc, WorkflowRunner{}, body) + var ids []string + for _, art := range arts { + if art != nil { + ids = append(ids, string(art.ArtifactID)) + } + } + if len(ids) > 0 { + if text != "" { + text += "\n" + } + text += "工件:" + strings.Join(ids, "、") + } + st.pinSubagentArtifacts(text) + if err != nil { + if text != "" { + return text + "\n" + err.Error() + } + return err.Error() + } + return text +} + +func (st *directRun) runPlannedWorkflow(goal string) string { + res, err := planWorkflow(st.rc, goal, WorkflowRunner{}) + st.pinSubagentArtifacts(res.Summary) + if err != nil { + if res.Summary != "" { + return res.Summary + } + return err.Error() + } + return res.Summary +} + func (st *directRun) submitCard(args string) string { if !st.cardflow { return "submit_card 未启用(设置 ui.cardflow 关闭);内容直接写在对话回复里" diff --git a/internal/runtime/mock.go b/internal/runtime/mock.go index 40c2613..9e4f066 100644 --- a/internal/runtime/mock.go +++ b/internal/runtime/mock.go @@ -7,6 +7,7 @@ import ( "agent-system-backend/internal/docgen" "agent-system-backend/internal/squadadapter" + "agent-system-backend/internal/workflow" core "agent-system-core" ) @@ -212,6 +213,17 @@ func (m *MockRunner) Execute(rc RunCtx) error { addText("Agent-Squad 编排完成:" + summary + "。步骤与子代理见右栏「计划」「代理」。") } + case strings.Contains(user, "#workflow"): + res, err := (WorkflowRunner{}).run(rc, workflow.Launch{Def: mockWorkflow()}) + if err != nil { + if res.Summary != "" { + addText(res.Summary) + } + addText("演示工作流没有跑完:" + err.Error()) + break + } + parts = append(parts, res.Parts...) + case strings.Contains(user, "#error"): return fmt.Errorf("mock: 演示失败路径(#error 指令)") @@ -229,7 +241,7 @@ func (m *MockRunner) Execute(rc RunCtx) error { core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: callID, Status: res.Status, Summary: truncateForJSON(res.Output, 120)}}, ) rc.Emit(core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseResponding}) - reply := fmt.Sprintf("(mock 回复)你说:%s\n\n这是 mock runner 的脚本化回复——未调用真实模型。可发送 #hitl / #agents / #doc / #error 触发对应演示路径。", user) + reply := fmt.Sprintf("(mock 回复)你说:%s\n\n这是 mock runner 的脚本化回复——未调用真实模型。可发送 #hitl / #agents / #doc / #workflow / #error 触发对应演示路径。", user) idx := 0 for _, r := range reply { rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: idx, Text: string(r)}) @@ -257,6 +269,23 @@ func (m *MockSubagent) Execute(rc RunCtx) error { return nil } +func mockWorkflow() core.WorkflowDef { + return core.WorkflowDef{ + ID: "mock.workflow", Title: "演示工作流", + Steps: []core.WorkflowStep{ + {ID: "length", Title: "篇幅", Kind: "hitl", Hitl: &core.HITLCardTemplate{ + Question: "要多长?", Kind: core.HITLKindChoice, AllowFreeInput: false, + Choices: []core.HITLChoice{ + {ID: "short", Label: "短", Description: "一段", Impact: "只写一段"}, + {ID: "long", Label: "长", Description: "几段", Impact: "会写得更长"}, + }, + }}, + {ID: "save", Title: "成稿", Kind: "artifact", Artifact: "mock-brief.md", Prompt: "篇幅:{{steps.length.output}}\n{{input}}"}, + {ID: "note", Title: "收件", Kind: "card", Card: &core.ActivityCardTemplate{Kind: core.CardKindResult, Title: "演示工作流完成"}}, + }, + } +} + func truncateForJSON(s string, n int) string { r := []rune(s) if len(r) <= n { diff --git a/internal/runtime/workflow.go b/internal/runtime/workflow.go new file mode 100644 index 0000000..1a3bde3 --- /dev/null +++ b/internal/runtime/workflow.go @@ -0,0 +1,423 @@ +package runtime + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + agentllm "agent-native/runtime/llm" + "agent-system-backend/internal/workflow" + core "agent-system-core" +) + +const workflowPlannerID = "agent.sub.workflow-plan" + +// WorkflowRunner 执行启动时拷贝下来的工作流定义。 +type WorkflowRunner struct { + // JudgeStream 注入验收模型。nil 时按轻量模型构造客户端。 + JudgeStream agentllm.StreamFunc + // ScriptTimeout 覆盖脚本墙钟超时。0 表示 600 秒,超过 1800 秒会被截断。 + ScriptTimeout time.Duration +} + +func (WorkflowRunner) Name() string { return "workflow" } + +type workflowSource interface { + Workflow() (workflow.Launch, bool) +} + +type workflowOutcome struct { + Summary string + Parts []core.Part + Artifacts []*core.ArtifactRef +} + +// planWorkflow 让规划子代理写出步骤,校验通过后必须经用户确认才执行。 +func planWorkflow(rc RunCtx, goal string, runner WorkflowRunner) (workflowOutcome, error) { + sp, ok := rc.(interface { + SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) + }) + if !ok { + return workflowOutcome{}, fmt.Errorf("workflow: 当前运行不能规划") + } + res, err := sp.SpawnAgent(workflowPlannerID, "规划", "把下面的目标规划成 WorkflowDef。只输出 JSON,不要完成目标本身。\n目标:"+goal, "") + if err != nil { + return workflowOutcome{}, err + } + text := res.Text + if text == "" { + text = res.Summary + } + def, err := parseWorkflowDef(text) + if err != nil { + return workflowOutcome{}, err + } + known := map[string]bool{} + for _, tool := range rc.Tools() { + known[tool.Name] = true + } + if err := workflow.Validate(def, workflow.DynamicLimits(), func(name string) bool { return known[name] }); err != nil { + return workflowOutcome{}, err + } + resp, err := rc.AskHITL(core.HITLCard{ + Question: planPreview(def), Kind: core.HITLKindConfirm, AllowFreeInput: false, + Choices: []core.HITLChoice{ + {ID: "yes", Label: "执行", Description: "按这些步骤做", Impact: "会调用工具并可能写入工件"}, + {ID: "no", Label: "取消", Description: "不执行", Impact: "不会改动任何东西"}, + }, + }) + if err != nil { + return workflowOutcome{}, err + } + if resp.ChoiceID != "yes" { + return workflowOutcome{Summary: "未确认,未执行"}, nil + } + return runner.run(rc, workflow.Launch{Def: def}) +} + +func parseWorkflowDef(raw string) (core.WorkflowDef, error) { + s := stripFence(raw) + start := strings.Index(s, "{") + end := strings.LastIndex(s, "}") + if start < 0 || end < start { + return core.WorkflowDef{}, fmt.Errorf("workflow: 规划结果无法解析") + } + var def core.WorkflowDef + if err := json.Unmarshal([]byte(s[start:end+1]), &def); err != nil { + return core.WorkflowDef{}, fmt.Errorf("workflow: 规划结果无法解析") + } + if strings.TrimSpace(def.ID) == "" { + def.ID = "dynamic" + } + if strings.TrimSpace(def.Title) == "" { + def.Title = "临时工作流" + } + return def, nil +} + +func planPreview(def core.WorkflowDef) string { + var b strings.Builder + fmt.Fprintf(&b, "按这个计划执行「%s」?\n", def.Title) + for i, step := range def.Steps { + fmt.Fprintf(&b, "%d. %s(%s)\n", i+1, step.Title, step.Kind) + } + return strings.TrimRight(b.String(), "\n") +} + +func (w WorkflowRunner) Execute(rc RunCtx) error { + src, ok := rc.(workflowSource) + if !ok { + return fmt.Errorf("workflow: 当前运行没有工作流定义") + } + launch, ok := src.Workflow() + if !ok { + return fmt.Errorf("workflow: 当前运行没有工作流定义") + } + res, err := w.run(rc, launch) + if len(res.Parts) > 0 { + _ = rc.SetAssistantParts(res.Parts) + } + return err +} + +func (w WorkflowRunner) run(rc RunCtx, launch workflow.Launch) (workflowOutcome, error) { + def := launch.Def + vars := workflow.Vars{ + Input: currentUserText(rc), + Slots: map[string]string{}, + Steps: map[string]string{}, + } + for k, v := range launch.Slots { + vars.Slots[k] = v + } + for _, missing := range launch.Missing { + if rc.Context().Err() != nil { + return workflowOutcome{}, rc.Context().Err() + } + resp, err := rc.AskHITL(core.HITLCard{ + Question: missing.Question, Kind: core.HITLKindInput, AllowFreeInput: true, + }) + if err != nil { + return workflowOutcome{}, err + } + answer := strings.TrimSpace(resp.FreeText) + if answer == "" { + answer = resp.ChoiceID + } + vars.Slots[missing.Key] = answer + } + + plan := core.PlanSnapshot{Title: def.Title, Steps: make([]core.PlanStep, len(def.Steps))} + for i, step := range def.Steps { + plan.Steps[i] = core.PlanStep{ID: step.ID, Text: step.Title, Status: "pending"} + } + if err := rc.UpdatePlan(plan); err != nil { + return workflowOutcome{}, err + } + + wholeLeft := def.MaxRetries + if wholeLeft <= 0 || wholeLeft > workflow.MaxWorkflowRetries { + wholeLeft = workflow.MaxWorkflowRetries + } + var drops []string + var arts []*core.ArtifactRef + var prev string + statuses := make([]string, len(def.Steps)) + for i := range def.Steps { + if rc.Context().Err() != nil { + return outcome(def, statuses, arts, drops, ""), rc.Context().Err() + } + step := def.Steps[i] + plan.Steps[i].Status = "in_progress" + if err := rc.UpdatePlan(plan); err != nil { + return workflowOutcome{}, err + } + rendered, err := workflow.RenderStep(step, vars) + if err != nil { + return outcome(def, statuses, arts, drops, err.Error()), err + } + output, stepArts, reason, err := w.runStepWithCritic(rc, def, rendered, vars, prev, launch.Scripts, &wholeLeft, &drops) + arts = append(arts, stepArts...) + if err != nil { + statuses[i] = "failed" + plan.Steps[i].Status = "skipped" + _ = rc.UpdatePlan(plan) + res := outcome(def, statuses, arts, drops, reason) + return res, err + } + vars.Steps[step.ID] = output + prev = output + statuses[i] = "done" + plan.Steps[i].Status = "done" + if err := rc.UpdatePlan(plan); err != nil { + return workflowOutcome{}, err + } + } + res := outcome(def, statuses, arts, drops, "") + return res, nil +} + +func (w WorkflowRunner) runStepWithCritic(rc RunCtx, def core.WorkflowDef, step core.WorkflowStep, vars workflow.Vars, prev string, scripts map[string]string, wholeLeft *int, drops *[]string) (string, []*core.ArtifactRef, string, error) { + stepRetries := 1 + if step.Critic != nil && step.Critic.MaxRetries > 0 { + stepRetries = step.Critic.MaxRetries + } + if stepRetries > workflow.MaxStepRetries { + stepRetries = workflow.MaxStepRetries + } + used := 0 + note := "" + var arts []*core.ArtifactRef + for { + current := step + if note != "" && current.Prompt != "" { + current.Prompt = current.Prompt + "\n\n未通过验收:" + note + } else if note != "" { + current.Prompt = "未通过验收:" + note + } + output, got, err := w.runWorkflowStep(rc, current, vars, prev, scripts) + arts = append(arts, got...) + if err != nil { + return output, arts, err.Error(), err + } + if step.Critic == nil || strings.TrimSpace(step.Critic.Criteria) == "" { + return output, arts, "", nil + } + verdict := w.judgeStep(rc, def.Title, step.Critic.Criteria, output) + if verdict.Passed { + return output, arts, "", nil + } + why := verdict.Reason + if verdict.NextAction != "" { + why += "。下一步:" + verdict.NextAction + } + *drops = append(*drops, step.ID+":"+why) + if used >= stepRetries || *wholeLeft <= 0 { + return output, arts, "步骤 " + step.Title + " 未通过验收:" + why, fmt.Errorf("workflow: 步骤 %s 未通过验收", step.ID) + } + used++ + *wholeLeft-- + note = why + } +} + +func (w WorkflowRunner) runWorkflowStep(rc RunCtx, step core.WorkflowStep, vars workflow.Vars, prev string, scripts map[string]string) (string, []*core.ArtifactRef, error) { + switch step.Kind { + case "prompt": + return runPromptStep(rc, step) + case "tool": + res := rc.ExecuteTool(core.ToolCallRequest{Name: step.Tool, Arguments: step.Args}) + if res.Status == "error" { + return res.Output, nil, fmt.Errorf("workflow: 工具 %s 失败:%s", step.Tool, res.Output) + } + var arts []*core.ArtifactRef + if res.Artifact != nil { + arts = append(arts, &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind}) + } + return res.Output, arts, nil + case "hitl": + card := core.HITLCard{Question: step.Title, Kind: core.HITLKindInput, AllowFreeInput: true} + if step.Hitl != nil { + card.Question = step.Hitl.Question + card.Kind = step.Hitl.Kind + card.Choices = step.Hitl.Choices + card.AllowFreeInput = step.Hitl.AllowFreeInput + card.FreeInputHint = step.Hitl.FreeInputHint + } + resp, err := rc.AskHITL(card) + if err != nil { + return "", nil, err + } + if strings.TrimSpace(resp.FreeText) != "" { + return strings.TrimSpace(resp.FreeText), nil, nil + } + return resp.ChoiceID, nil, nil + case "artifact": + content := step.Prompt + if strings.TrimSpace(content) == "" { + content = prev + } + name := step.Artifact + if name == "" { + name = step.ID + ".md" + } + art, err := rc.WriteArtifact(name, "document", "text/markdown", []byte(content)) + if err != nil { + return "", nil, err + } + return string(art.ID), []*core.ArtifactRef{{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}}, nil + case "card": + card := core.ActivityCard{Title: step.Title, Kind: core.CardKindResult} + if step.Card != nil { + card.Kind = step.Card.Kind + card.Title = step.Card.Title + card.Tone = step.Card.Tone + card.Body = step.Card.Body + card.Actions = step.Card.Actions + } + created, err := rc.CreateCard(card) + if err != nil { + return "", nil, err + } + return string(created.ID), nil, nil + case "script": + return executeScriptStep(rc, w, step, scripts) + default: + return "", nil, fmt.Errorf("workflow: 未知步骤类型 %s", step.Kind) + } +} + +func runPromptStep(rc RunCtx, step core.WorkflowStep) (string, []*core.ArtifactRef, error) { + agent := step.Agent + if strings.TrimSpace(agent) == "" { + agent = "general" + } + var res core.AgentResultPayload + var err error + if sp, ok := rc.(interface { + SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) + }); ok { + res, err = sp.SpawnAgent(subagentProfileID(agent), step.Title, step.Prompt, "") + } else { + res, err = rc.SpawnSubagent(step.Title, step.Prompt, "") + } + if err != nil { + return "", nil, err + } + text := res.Text + if text == "" { + text = res.Summary + } + return text, artifactsIn(rc, text), nil +} + +func artifactsIn(rc RunCtx, text string) []*core.ArtifactRef { + var out []*core.ArtifactRef + seen := map[string]bool{} + for _, id := range artIDRe.FindAllString(text, -1) { + if seen[id] { + continue + } + art, _, ok := rc.Artifact(id) + if !ok { + continue + } + seen[id] = true + out = append(out, &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}) + } + return out +} + +func (w WorkflowRunner) judgeStep(rc RunCtx, goal, criteria, output string) verifyVerdict { + stream := w.JudgeStream + model := "verify" + if stream == nil { + route := ResolveLightweightModel(rc.Profile(), rc.Settings()) + if route.Model == "" || route.Base == "" { + return verifyVerdict{Passed: false, Reason: "验证器不可用", NextAction: "检查轻量模型配置"} + } + model = route.Model + client := &agentllm.Client{BaseURL: route.Base, APIKey: route.Key} + stream = client.Stream + } + prompt := fmt.Sprintf("你是验收员。目标:%s\n验收标准:%s\n这一步的产出:\n%s\n标准里要求出现的句子,产出里必须逐字包含,否则 passed 必须为 false,nextAction 写出要补上的原句。只输出一个 JSON 对象,字段为 passed、reason、nextAction。", goal, criteria, output) + var b strings.Builder + for ev := range stream(rc.Context(), model, []agentllm.Message{{Role: agentllm.RoleUser, Content: prompt}}, nil, agentllm.StreamOptions{}) { + switch ev.Kind { + case agentllm.KindTextDelta: + b.WriteString(ev.Text) + case agentllm.KindError: + return verifyVerdict{Passed: false, Reason: "验证器不可用", NextAction: "稍后再试"} + case agentllm.KindMessageEnd: + if b.Len() == 0 { + b.WriteString(ev.Message.Content) + } + } + } + verdict, err := parseVerdict(b.String()) + if err != nil { + return verifyVerdict{Passed: false, Reason: "无法解析", NextAction: "按验收标准重做"} + } + return verdict +} + +func outcome(def core.WorkflowDef, statuses []string, arts []*core.ArtifactRef, drops []string, stopped string) workflowOutcome { + var b strings.Builder + title := def.Title + if title == "" { + title = def.ID + } + if stopped != "" { + fmt.Fprintf(&b, "工作流「%s」停止:%s\n", title, stopped) + } else { + fmt.Fprintf(&b, "工作流「%s」完成。\n", title) + } + for i, step := range def.Steps { + state := "pending" + if i < len(statuses) && statuses[i] != "" { + state = statuses[i] + } + fmt.Fprintf(&b, "- %s:%s\n", step.Title, state) + } + if len(arts) > 0 { + b.WriteString("工件:") + for i, art := range arts { + if i > 0 { + b.WriteString("、") + } + b.WriteString(string(art.ArtifactID)) + } + b.WriteString("\n") + } + if len(drops) > 0 { + b.WriteString("已丢弃:") + b.WriteString(strings.Join(drops, ";")) + b.WriteString("\n") + } + parts := []core.Part{{Type: core.PartText, Text: strings.TrimRight(b.String(), "\n")}} + for _, art := range arts { + parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: art}) + } + return workflowOutcome{Summary: parts[0].Text, Parts: parts, Artifacts: arts} +} diff --git a/internal/runtime/workflow_exec_test.go b/internal/runtime/workflow_exec_test.go new file mode 100644 index 0000000..8e8d097 --- /dev/null +++ b/internal/runtime/workflow_exec_test.go @@ -0,0 +1,347 @@ +package runtime + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "agent-system-backend/internal/workflow" + core "agent-system-core" +) + +type wfFake struct { + ctx context.Context + launch workflow.Launch + hitl []core.HITLResponse + hitlQ []string + plans []core.PlanSnapshot + tools []core.ToolCallRequest + prompts []string + writes []string + cards []core.ActivityCard + assistant []core.Part + arts map[string]core.Artifact + byID map[string]core.WorkflowDef + spawnText string + spawnIDs []string + toolNames []string + settings core.EffectiveSettings + sandbox string + scripts map[string]string +} + +func (f *wfFake) Context() context.Context { + if f.ctx == nil { + return context.Background() + } + return f.ctx +} +func (f *wfFake) Run() core.Run { return core.Run{ID: "run_wf"} } +func (f *wfFake) Profile() core.AgentProfile { return core.AgentProfile{} } +func (f *wfFake) Settings() core.EffectiveSettings { + if f.settings == nil { + return core.EffectiveSettings{} + } + return f.settings +} +func (f *wfFake) ConversationSandbox() string { return f.sandbox } +func (f *wfFake) Plan() (core.PlanSnapshot, bool) { + if len(f.plans) == 0 { + return core.PlanSnapshot{}, false + } + p := f.plans[len(f.plans)-1] + p.Steps = append([]core.PlanStep(nil), p.Steps...) + return p, true +} +func (f *wfFake) WorkflowScript(id, path string) (string, bool) { + body, ok := f.scripts[workflow.ScriptKey(id, path)] + return body, ok +} +func (f *wfFake) Emit(core.StreamEventType, any) error { return nil } +func (f *wfFake) SetAssistantParts(parts []core.Part) error { + f.assistant = parts + return nil +} +func (f *wfFake) AskHITL(card core.HITLCard) (core.HITLResponse, error) { + f.hitlQ = append(f.hitlQ, card.Question) + if len(f.hitl) == 0 { + return core.HITLResponse{}, context.Canceled + } + resp := f.hitl[0] + f.hitl = f.hitl[1:] + return resp, nil +} +func (f *wfFake) SpawnSubagent(name, task, hint string) (core.AgentResultPayload, error) { + return f.SpawnAgent("agent.sub.general", name, task, hint) +} +func (f *wfFake) SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) { + f.prompts = append(f.prompts, agentID+"|"+task) + f.spawnIDs = append(f.spawnIDs, agentID) + text := f.spawnText + if text == "" { + text = "正文 art_0123456789ab" + } + return core.AgentResultPayload{Text: text, Status: core.RunCompleted}, nil +} +func (f *wfFake) UpdatePlan(p core.PlanSnapshot) error { + p.Steps = append([]core.PlanStep(nil), p.Steps...) + f.plans = append(f.plans, p) + return nil +} +func (f *wfFake) UpdateTodos(core.TodoList) error { return nil } +func (f *wfFake) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) { + f.writes = append(f.writes, name+"="+string(content)) + art := core.Artifact{ID: "art_aaaaaaaaaaaa", Name: name, Kind: kind} + if f.arts == nil { + f.arts = map[string]core.Artifact{} + } + f.arts[string(art.ID)] = art + return art, nil +} +func (f *wfFake) Retrieve(core.RetrievalQuery) ([]core.RetrievalHit, error) { return nil, nil } +func (f *wfFake) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult { + f.tools = append(f.tools, call) + if call.Name == "make_file" { + art := core.Artifact{ID: "art_bbbbbbbbbbbb", Name: "made.md"} + if f.arts == nil { + f.arts = map[string]core.Artifact{} + } + f.arts[string(art.ID)] = art + return core.ToolCallResult{Status: "ok", Output: "wrote", Artifact: &art} + } + return core.ToolCallResult{Status: "ok", Output: "12:00"} +} +func (f *wfFake) History() []core.Message { return nil } +func (f *wfFake) Attachments() []core.ParsedDocument { return nil } +func (f *wfFake) Tools() []core.ToolDesc { + var out []core.ToolDesc + for _, name := range f.toolNames { + out = append(out, core.ToolDesc{Name: name}) + } + return out +} +func (f *wfFake) Subtask() string { return "" } +func (f *wfFake) CreateCard(card core.ActivityCard) (core.ActivityCard, error) { + card.ID = "card_1" + f.cards = append(f.cards, card) + return card, nil +} +func (f *wfFake) SetCardStatus(string, core.CardStatus, string, string) error { return nil } +func (f *wfFake) SubmitCard(card core.ActivityCard) (core.ActivityCard, error) { + return card, nil +} +func (f *wfFake) BeginStreamCard(card core.ActivityCard) (core.ActivityCard, error) { + return card, nil +} +func (f *wfFake) AppendStreamCard(string) error { return nil } +func (f *wfFake) CommitStreamCard() (core.ActivityCard, error) { return core.ActivityCard{}, nil } +func (f *wfFake) Card(string) *core.ActivityCard { return nil } +func (f *wfFake) Artifact(id string) (core.Artifact, []byte, bool) { + if f.arts == nil { + f.arts = map[string]core.Artifact{} + } + if id == "art_0123456789ab" { + return core.Artifact{ID: core.ID(id), Name: "note.md"}, []byte("x"), true + } + art, ok := f.arts[id] + return art, nil, ok +} +func (f *wfFake) SetContextObservation(core.ContextObservation) {} +func (f *wfFake) Workflow() (workflow.Launch, bool) { return f.launch, true } +func (f *wfFake) WorkflowByID(id string) (core.WorkflowDef, bool) { + def, ok := f.byID[id] + return def, ok +} + +func TestWorkflowRunsFiveKindsAndAsksMissingSlot(t *testing.T) { + f := &wfFake{hitl: []core.HITLResponse{{FreeText: "核能"}, {ChoiceID: "short"}}} + f.launch = workflow.Launch{ + Missing: []workflow.SlotQuestion{{Key: "topic", Question: "主题是什么?"}}, + Def: core.WorkflowDef{ID: "demo.brief", Title: "简报", Steps: []core.WorkflowStep{ + {ID: "len", Title: "篇幅", Kind: "hitl", Hitl: &core.HITLCardTemplate{Question: "篇幅?", Kind: core.HITLKindChoice, Choices: []core.HITLChoice{{ID: "short", Label: "短"}}}}, + {ID: "draft", Title: "起草", Kind: "prompt", Agent: "research", Prompt: "写 {{slots.topic}}"}, + {ID: "now", Title: "时间", Kind: "tool", Tool: "time_now", Args: []byte(`{"q":"{{slots.topic}}"}`)}, + {ID: "save", Title: "存稿", Kind: "artifact", Artifact: "brief.md", Prompt: "{{steps.draft.output}}"}, + {ID: "note", Title: "通知", Kind: "card", Card: &core.ActivityCardTemplate{Kind: core.CardKindResult, Title: "好了"}}, + }}, + } + if err := (WorkflowRunner{}).Execute(f); err != nil { + t.Fatal(err) + } + if f.hitlQ[0] != "主题是什么?" || f.hitlQ[1] != "篇幅?" { + t.Fatalf("questions %v", f.hitlQ) + } + if len(f.prompts) != 1 || !strings.Contains(f.prompts[0], "agent.sub.research|写 核能") { + t.Fatalf("prompt %v", f.prompts) + } + if len(f.tools) != 1 || string(f.tools[0].Arguments) == "" || !strings.Contains(string(f.tools[0].Arguments), "核能") { + t.Fatalf("tool %#v", f.tools) + } + if len(f.writes) != 1 || f.writes[0] != "brief.md=正文 art_0123456789ab" { + t.Fatalf("writes %v", f.writes) + } + if len(f.cards) != 1 || f.cards[0].Title != "好了" { + t.Fatalf("cards %+v", f.cards) + } + var sawProgress, sawDone bool + for _, p := range f.plans { + for _, s := range p.Steps { + if s.ID == "draft" && s.Status == "in_progress" { + sawProgress = true + } + if s.ID == "note" && s.Status == "done" { + sawDone = true + } + } + } + if !sawProgress || !sawDone { + t.Fatal("plan did not move from in_progress to done") + } + text := f.assistant[0].Text + if !strings.Contains(text, "art_0123456789ab") || !strings.Contains(text, "art_aaaaaaaaaaaa") { + t.Fatalf("summary %s", text) + } +} + +func TestCriticRetriesThenStops(t *testing.T) { + f := &wfFake{} + f.launch = workflow.Launch{Def: core.WorkflowDef{ + ID: "a", Title: "验收", MaxRetries: 6, + Steps: []core.WorkflowStep{{ + ID: "draft", Title: "起草", Kind: "prompt", Prompt: "写", + Critic: &core.StepCritic{Criteria: "必须点明主题", MaxRetries: 1}, + }}, + }} + runner := WorkflowRunner{JudgeStream: passedStream( + `{"passed":false,"reason":"没点明主题","nextAction":"补上主题"}`, + `{"passed":true,"reason":"可以","nextAction":""}`, + )} + if err := runner.Execute(f); err != nil { + t.Fatal(err) + } + if len(f.prompts) != 2 || !strings.Contains(f.prompts[1], "没点明主题") || !strings.Contains(f.prompts[1], "补上主题") { + t.Fatalf("prompts %v", f.prompts) + } + if !strings.Contains(f.assistant[0].Text, "已丢弃") || !strings.Contains(f.assistant[0].Text, "没点明主题") { + t.Fatalf("summary %s", f.assistant[0].Text) + } + + f2 := &wfFake{} + f2.launch = f.launch + bad := WorkflowRunner{JudgeStream: passedStream("不是 json", "不是 json")} + err := bad.Execute(f2) + if err == nil || len(f2.prompts) != 2 { + t.Fatalf("err %v prompts %d", err, len(f2.prompts)) + } + if !strings.Contains(f2.assistant[0].Text, "无法解析") { + t.Fatalf("summary %s", f2.assistant[0].Text) + } + + f3 := &wfFake{} + f3.launch = workflow.Launch{Def: core.WorkflowDef{ + ID: "a", Title: "验收", MaxRetries: 1, + Steps: []core.WorkflowStep{{ + ID: "draft", Title: "起草", Kind: "prompt", Prompt: "写", + Critic: &core.StepCritic{Criteria: "要过", MaxRetries: 3}, + }}, + }} + fail := `{"passed":false,"reason":"不行","nextAction":"重写"}` + stop := WorkflowRunner{JudgeStream: passedStream(fail, fail, fail)} + if err := stop.Execute(f3); err == nil || len(f3.prompts) != 2 { + t.Fatalf("whole cap err %v attempts %d", err, len(f3.prompts)) + } +} + +func TestRunWorkflowToolUsesRegisteredDefinition(t *testing.T) { + f := &wfFake{hitl: []core.HITLResponse{{ChoiceID: "yes"}}} + f.byID = map[string]core.WorkflowDef{ + "demo.brief": {ID: "demo.brief", Title: "简报", Steps: []core.WorkflowStep{ + {ID: "ask", Title: "确认", Kind: "hitl", Hitl: &core.HITLCardTemplate{Question: "可以吗?", Kind: core.HITLKindConfirm}}, + }}, + } + st := &directRun{rc: f} + out := st.runRegisteredWorkflow(`{"workflow":"demo.brief","inputs":{"topic":"核能"}}`) + if !strings.Contains(out, "完成") || f.hitlQ[0] != "可以吗?" { + t.Fatalf("out %s questions %v", out, f.hitlQ) + } + if got := st.runRegisteredWorkflow(`{"workflow":"missing"}`); !strings.Contains(got, "未注册") { + t.Fatalf("%s", got) + } + if got := st.runRegisteredWorkflow(`{"goal":"做一件事","workflow":"demo.brief"}`); !strings.Contains(got, "只能") { + t.Fatalf("%s", got) + } +} + +func TestJudgeUsesConfiguredModel(t *testing.T) { + var got string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + got, _ = body["model"].(string) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"{\\\"passed\\\":true,\\\"reason\\\":\\\"ok\\\",\\\"nextAction\\\":\\\"\\\"}\"}}]}\n\n") + _, _ = io.WriteString(w, "data: [DONE]\n\n") + })) + t.Cleanup(srv.Close) + f := &wfFake{ + settings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{ + "baseUrl": srv.URL, "apiKey": "test-key", "model": "MiniMax-test", + }}}, + launch: workflow.Launch{Def: core.WorkflowDef{ + ID: "c", Title: "验收", + Steps: []core.WorkflowStep{{ + ID: "draft", Title: "起草", Kind: "prompt", Prompt: "写", + Critic: &core.StepCritic{Criteria: "要过", MaxRetries: 1}, + }}, + }}, + } + if err := (WorkflowRunner{}).Execute(f); err != nil { + t.Fatal(err) + } + if got != "MiniMax-test" { + t.Fatalf("model %q", got) + } +} + +func TestPlannedWorkflowNeedsConfirm(t *testing.T) { + plan := `{"id":"plan.brief","title":"简报","steps":[{"id":"ask","title":"确认","kind":"hitl","hitl":{"question":"可以吗?","kind":"confirm","allowFreeInput":false}}]}` + f := &wfFake{ + spawnText: plan, + hitl: []core.HITLResponse{{ChoiceID: "yes"}, {ChoiceID: "ok"}}, + } + st := &directRun{rc: f} + out := st.runRegisteredWorkflow(`{"goal":"写一份简报"}`) + if f.spawnIDs[0] != "agent.sub.workflow-plan" || !strings.Contains(f.hitlQ[0], "按这个计划执行") { + t.Fatalf("spawn %v questions %v", f.spawnIDs, f.hitlQ) + } + if !strings.Contains(out, "完成") || f.hitlQ[1] != "可以吗?" { + t.Fatalf("out %s questions %v", out, f.hitlQ) + } + + f2 := &wfFake{spawnText: plan, hitl: []core.HITLResponse{{ChoiceID: "no"}}} + out = (&directRun{rc: f2}).runRegisteredWorkflow(`{"goal":"写一份简报"}`) + if out != "未确认,未执行" || len(f2.hitlQ) != 1 { + t.Fatalf("out %s questions %v", out, f2.hitlQ) + } + + var steps []string + for i := 0; i < 9; i++ { + steps = append(steps, fmt.Sprintf(`{"id":"s%d","title":"步","kind":"hitl"}`, i)) + } + f3 := &wfFake{spawnText: `{"id":"too.big","title":"太多","steps":[` + strings.Join(steps, ",") + `]}`} + out = (&directRun{rc: f3}).runRegisteredWorkflow(`{"goal":"做很多事"}`) + if !strings.Contains(out, "超出上限") || len(f3.hitlQ) != 0 { + t.Fatalf("out %s questions %v", out, f3.hitlQ) + } + + f4 := &wfFake{spawnText: `{"id":"bad.script","title":"脚本","steps":[{"id":"run","title":"跑","kind":"script","script":"a.py"}]}`} + out = (&directRun{rc: f4}).runRegisteredWorkflow(`{"goal":"跑脚本"}`) + if !strings.Contains(out, "script") || len(f4.hitlQ) != 0 { + t.Fatalf("out %s questions %v", out, f4.hitlQ) + } +} diff --git a/internal/runtime/workflow_script.go b/internal/runtime/workflow_script.go new file mode 100644 index 0000000..0e126fa --- /dev/null +++ b/internal/runtime/workflow_script.go @@ -0,0 +1,468 @@ +package runtime + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "agent-system-backend/internal/tools" + "agent-system-backend/internal/workflow" + core "agent-system-core" + _ "embed" +) + +//go:embed agentwf.py +var agentwfSource string + +const ( + scriptDefaultTimeout = 600 * time.Second + scriptMaxTimeout = 1800 * time.Second + scriptMaxCalls = 50 + scriptMaxLine = 1 << 20 +) + +var ( + agentwfOnce sync.Once + agentwfDir string + agentwfErr error +) + +func agentwfHelperDir() (string, error) { + agentwfOnce.Do(func() { + dir, err := os.MkdirTemp("", "agentwf-*") + if err != nil { + agentwfErr = err + return + } + if err := os.WriteFile(filepath.Join(dir, "agentwf.py"), []byte(agentwfSource), 0o644); err != nil { + agentwfErr = err + return + } + agentwfDir = dir + }) + return agentwfDir, agentwfErr +} + +func scriptsEnabled(rc RunCtx) bool { + return rc.Settings().GetBool("runtime.workflowScripts", false) +} + +func scriptSandbox(rc RunCtx) (string, error) { + sandbox := strings.TrimSpace(conversationSandbox(rc)) + if sandbox == "" { + return "", fmt.Errorf("workflow: 没有会话沙箱") + } + return sandbox, nil +} + +func executeScriptStep(rc RunCtx, runner WorkflowRunner, step core.WorkflowStep, scripts map[string]string) (string, []*core.ArtifactRef, error) { + body := scripts[step.Script] + if strings.TrimSpace(body) == "" { + var err error + body, err = scriptBody(rc, step.Script) + if err != nil { + return "", nil, err + } + } + sandbox, err := scriptSandbox(rc) + if err != nil { + return "", nil, err + } + path, err := writeScriptFile(sandbox, "step-"+step.ID, body) + if err != nil { + return "", nil, err + } + return runScriptFile(rc, runner, path) +} + +func scriptBody(rc RunCtx, path string) (string, error) { + src, ok := rc.(interface { + Workflow() (workflow.Launch, bool) + }) + if !ok { + return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) + } + launch, ok := src.Workflow() + if !ok { + return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) + } + body, ok := launch.Scripts[path] + if !ok || strings.TrimSpace(body) == "" { + return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) + } + return body, nil +} + +func registeredWorkflowLaunch(rc RunCtx, def core.WorkflowDef, slots map[string]string) workflow.Launch { + launch := workflow.Launch{Def: def, Slots: slots, Scripts: map[string]string{}} + lookup, ok := rc.(interface { + WorkflowScript(id, path string) (string, bool) + }) + if !ok { + return launch + } + for _, step := range def.Steps { + if step.Kind != "script" || strings.TrimSpace(step.Script) == "" { + continue + } + if body, found := lookup.WorkflowScript(def.ID, step.Script); found { + launch.Scripts[step.Script] = body + } + } + return launch +} + +func runModelScript(rc RunCtx, runner WorkflowRunner, body string) (string, []*core.ArtifactRef, error) { + if !scriptsEnabled(rc) { + return "", nil, fmt.Errorf("workflow: 工作流脚本未开启") + } + preview := body + if len(preview) > 4000 { + preview = preview[:4000] + "\n…(脚本超长,已截断展示)" + } + resp, err := rc.AskHITL(core.HITLCard{ + Question: "运行这段脚本?\n" + preview, Kind: core.HITLKindConfirm, AllowFreeInput: false, + Choices: []core.HITLChoice{ + {ID: "yes", Label: "运行", Description: "在会话沙箱里执行", Impact: "会调用工具并可能写入工件"}, + {ID: "no", Label: "取消", Description: "不执行", Impact: "不会改动任何东西"}, + }, + }) + if err != nil { + return "", nil, err + } + if resp.ChoiceID != "yes" { + return "未确认,未执行", nil, nil + } + sandbox, err := scriptSandbox(rc) + if err != nil { + return "", nil, err + } + path, err := writeScriptFile(sandbox, "model", body) + if err != nil { + return "", nil, err + } + return runScriptFile(rc, runner, path) +} + +func writeScriptFile(sandbox, prefix, body string) (string, error) { + dir := filepath.Join(sandbox, ".agent") + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + path := filepath.Join(dir, fmt.Sprintf("wf-%s-%d.py", prefix, time.Now().UnixNano())) + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + return "", err + } + return path, nil +} + +func runScriptFile(rc RunCtx, runner WorkflowRunner, path string) (string, []*core.ArtifactRef, error) { + if !scriptsEnabled(rc) { + return "", nil, fmt.Errorf("workflow: 工作流脚本未开启") + } + sandbox, err := scriptSandbox(rc) + if err != nil { + return "", nil, err + } + timeout := runner.ScriptTimeout + if timeout <= 0 { + timeout = scriptDefaultTimeout + } + if timeout > scriptMaxTimeout { + timeout = scriptMaxTimeout + } + helper, err := agentwfHelperDir() + if err != nil { + return "", nil, err + } + parent := rc.Context() + if parent == nil { + parent = context.Background() + } + ctx, cancel := context.WithTimeout(parent, timeout) + defer cancel() + cmd, err := tools.ScriptCommand(ctx, sandbox, path) + if err != nil { + return "", nil, err + } + cmd.Env = prependPythonPath(cmd.Env, helper) + reqR, reqW, err := os.Pipe() + if err != nil { + return "", nil, err + } + respR, respW, err := os.Pipe() + if err != nil { + reqR.Close() + reqW.Close() + return "", nil, err + } + cmd.ExtraFiles = []*os.File{reqW, respR} + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + if err := cmd.Start(); err != nil { + reqR.Close() + reqW.Close() + respR.Close() + respW.Close() + return "", nil, err + } + reqW.Close() + respR.Close() + defer reqR.Close() + defer respW.Close() + arts, callErr := serveScript(ctx, rc, reqR, respW) + if callErr != nil { + cancel() + } + waitErr := cmd.Wait() + text := clipScriptOutput(&stdout, &stderr) + if callErr != nil { + return text, arts, callErr + } + if ctx.Err() == context.DeadlineExceeded { + return text, arts, fmt.Errorf("workflow: 脚本超时") + } + if ctx.Err() == context.Canceled { + return text, arts, fmt.Errorf("workflow: 脚本已取消") + } + if waitErr != nil { + return text, arts, fmt.Errorf("workflow: 脚本退出:%v\n%s", waitErr, strings.TrimSpace(stderr.String())) + } + return text, arts, nil +} + +func clipScriptOutput(stdout, stderr *bytes.Buffer) string { + text := strings.TrimSpace(stdout.String()) + errText := strings.TrimSpace(stderr.String()) + if errText != "" { + if text != "" { + text += "\n" + } + text += errText + } + if len(text) > 16000 { + text = text[:16000] + "…(超长截断)" + } + return text +} + +func prependPythonPath(env []string, dir string) []string { + for i, kv := range env { + if strings.HasPrefix(kv, "PYTHONPATH=") { + env[i] = "PYTHONPATH=" + dir + string(os.PathListSeparator) + strings.TrimPrefix(kv, "PYTHONPATH=") + return env + } + } + return append(env, "PYTHONPATH="+dir) +} + +type scriptRequest struct { + ID string `json:"id"` + Op string `json:"op"` + Name string `json:"name"` + Args json.RawMessage `json:"args"` + Question string `json:"question"` + Choices []core.HITLChoice `json:"choices"` + Agent string `json:"agent"` + Task string `json:"task"` + Content string `json:"content"` + Title string `json:"title"` + Kind string `json:"kind"` + Status string `json:"status"` + StepID string `json:"stepId"` +} + +type scriptResponse struct { + ID string `json:"id"` + OK bool `json:"ok"` + Output string `json:"output,omitempty"` + Error string `json:"error,omitempty"` +} + +func serveScript(ctx context.Context, rc RunCtx, reqR, respW *os.File) ([]*core.ArtifactRef, error) { + sc := bufio.NewScanner(reqR) + sc.Buffer(make([]byte, 64*1024), scriptMaxLine) + var arts []*core.ArtifactRef + calls := 0 + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + calls++ + var req scriptRequest + if err := json.Unmarshal([]byte(line), &req); err != nil { + _ = writeScriptResponse(respW, scriptResponse{OK: false, Error: "请求不是合法 JSON"}) + continue + } + if calls > scriptMaxCalls { + _ = writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: false, Error: "脚本回调超过 50 次"}) + return arts, fmt.Errorf("workflow: 脚本回调超过 50 次") + } + if ctx.Err() != nil { + return arts, ctx.Err() + } + output, got, err := dispatchScript(rc, req) + arts = append(arts, got...) + if err != nil { + _ = writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: false, Error: err.Error()}) + if ctx.Err() != nil { + return arts, ctx.Err() + } + continue + } + if err := writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: true, Output: output}); err != nil { + return arts, err + } + } + if err := sc.Err(); err != nil { + if err == bufio.ErrTooLong { + return arts, fmt.Errorf("workflow: 单行请求超过 1MB") + } + if ctx.Err() != nil { + return arts, ctx.Err() + } + return arts, err + } + return arts, nil +} + +func writeScriptResponse(w io.Writer, resp scriptResponse) error { + raw, err := json.Marshal(resp) + if err != nil { + return err + } + _, err = w.Write(append(raw, '\n')) + return err +} + +func dispatchScript(rc RunCtx, req scriptRequest) (string, []*core.ArtifactRef, error) { + switch req.Op { + case "tool.call": + if _, denied := workflow.DeniedTools[req.Name]; denied { + return "", nil, fmt.Errorf("不能调用 %s", req.Name) + } + res := rc.ExecuteTool(core.ToolCallRequest{Name: req.Name, Arguments: req.Args}) + if res.Status == "error" { + return "", nil, fmt.Errorf("%s", res.Output) + } + var arts []*core.ArtifactRef + if res.Artifact != nil { + arts = append(arts, &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind}) + } + arts = append(arts, artifactsIn(rc, res.Output)...) + return res.Output, arts, nil + case "hitl.ask": + card := core.HITLCard{Question: req.Question, Kind: core.HITLKindInput, AllowFreeInput: true} + if len(req.Choices) > 0 { + card.Kind = core.HITLKindChoice + card.AllowFreeInput = false + card.Choices = req.Choices + for i := range card.Choices { + if card.Choices[i].Impact == "" { + card.Choices[i].Impact = "会按这个选择继续" + } + } + } + resp, err := rc.AskHITL(card) + if err != nil { + return "", nil, err + } + if strings.TrimSpace(resp.FreeText) != "" { + return strings.TrimSpace(resp.FreeText), nil, nil + } + return resp.ChoiceID, nil, nil + case "agent.spawn": + agent := req.Agent + if strings.TrimSpace(agent) == "" { + agent = "general" + } + var res core.AgentResultPayload + var err error + if sp, ok := rc.(interface { + SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) + }); ok { + res, err = sp.SpawnAgent(subagentProfileID(agent), req.Title, req.Task, "") + } else { + res, err = rc.SpawnSubagent(req.Title, req.Task, "") + } + if err != nil { + return "", nil, err + } + text := res.Text + if text == "" { + text = res.Summary + } + return text, artifactsIn(rc, text), nil + case "artifact.register": + name := req.Name + if name == "" { + name = "script.md" + } + art, err := rc.WriteArtifact(name, "document", "text/markdown", []byte(req.Content)) + if err != nil { + return "", nil, err + } + return string(art.ID), []*core.ArtifactRef{{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}}, nil + case "card.create": + kind := core.CardKind(req.Kind) + if kind == "" { + kind = core.CardKindResult + } + title := req.Title + if title == "" { + title = req.Name + } + created, err := rc.CreateCard(core.ActivityCard{Title: title, Kind: kind}) + if err != nil { + return "", nil, err + } + return string(created.ID), nil, nil + case "step": + applyScriptPlan(rc, req.StepID, req.Title, req.Status) + return req.Status, nil, nil + default: + return "", nil, fmt.Errorf("未知回调 %s", req.Op) + } +} + +func applyScriptPlan(rc RunCtx, id, title, status string) { + switch status { + case "pending", "in_progress", "done", "skipped": + default: + status = "done" + } + snap := core.PlanSnapshot{} + if src, ok := rc.(interface { + Plan() (core.PlanSnapshot, bool) + }); ok { + if cur, found := src.Plan(); found { + snap = cur + snap.Steps = append([]core.PlanStep(nil), cur.Steps...) + } + } + found := false + for i := range snap.Steps { + if snap.Steps[i].ID == id { + snap.Steps[i].Status = status + if title != "" { + snap.Steps[i].Text = title + } + found = true + break + } + } + if !found { + snap.Steps = append(snap.Steps, core.PlanStep{ID: id, Text: title, Status: status}) + } + _ = rc.UpdatePlan(snap) +} diff --git a/internal/runtime/workflow_script_test.go b/internal/runtime/workflow_script_test.go new file mode 100644 index 0000000..c26ac10 --- /dev/null +++ b/internal/runtime/workflow_script_test.go @@ -0,0 +1,243 @@ +package runtime + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "agent-system-backend/internal/workflow" + core "agent-system-core" +) + +func scriptSettings() core.EffectiveSettings { + return core.EffectiveSettings{"runtime": map[string]any{"workflowScripts": true}} +} + +func TestScriptStepRunsEachCallback(t *testing.T) { + body := `import agentwf as wf +print("tool", wf.tool("time_now", {})) +print("ask", wf.ask("继续?", [{"id":"yes","label":"好","impact":"继续"}])) +print("agent", wf.agent("research", "写一句")) +print("art", wf.artifact("note.md", "你好")) +print("card", wf.card("结果", "result")) +print("step", wf.step("s1", "一步", "done")) +print("file", wf.tool("make_file", {})) +` + f := &wfFake{ + sandbox: t.TempDir(), + settings: scriptSettings(), + hitl: []core.HITLResponse{{ChoiceID: "yes"}}, + launch: workflow.Launch{ + Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{ + {ID: "run", Title: "跑", Kind: "script", Script: "workflows/scripts/demo.py"}, + {ID: "save", Title: "留下", Kind: "artifact", Prompt: "{{steps.run.output}}"}, + }}, + Scripts: map[string]string{"workflows/scripts/demo.py": body}, + }, + } + if err := (WorkflowRunner{}).Execute(f); err != nil { + t.Fatal(err) + } + saved := f.writes[len(f.writes)-1] + for _, want := range []string{"tool 12:00", "ask yes", "agent 正文", "art art_aaaaaaaaaaaa", "card card_1", "file wrote"} { + if !strings.Contains(saved, want) { + t.Fatalf("missing %q in %s", want, saved) + } + } + if len(f.tools) != 2 || f.tools[0].Name != "time_now" || f.tools[1].Name != "make_file" { + t.Fatalf("tools %+v", f.tools) + } + if len(f.cards) != 1 || f.cards[0].Title != "结果" { + t.Fatalf("cards %+v", f.cards) + } + if len(f.spawnIDs) != 1 || f.spawnIDs[0] != "agent.sub.research" { + t.Fatalf("spawn %v", f.spawnIDs) + } + if f.writes[0] != "note.md=你好" { + t.Fatalf("writes %v", f.writes) + } + sawStep := false + var ids []string + for _, plan := range f.plans { + for _, step := range plan.Steps { + if step.ID == "s1" && step.Status == "done" { + sawStep = true + } + } + } + for _, part := range f.assistant { + if part.Artifact != nil { + ids = append(ids, string(part.Artifact.ArtifactID)) + } + } + if !sawStep { + t.Fatalf("plans %+v", f.plans) + } + joined := strings.Join(ids, " ") + if !strings.Contains(joined, "art_aaaaaaaaaaaa") || !strings.Contains(joined, "art_bbbbbbbbbbbb") { + t.Fatalf("deliverables %v", ids) + } +} + +func TestScriptRejectsDeniedTool(t *testing.T) { + dir := t.TempDir() + body := `import agentwf as wf +try: + wf.tool("run_python", {}) +except Exception as exc: + open("ERR", "w").write(str(exc)) +` + f := &wfFake{ + sandbox: dir, + settings: scriptSettings(), + launch: workflow.Launch{ + Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ + ID: "run", Title: "跑", Kind: "script", Script: "a.py", + }}}, + Scripts: map[string]string{"a.py": body}, + }, + } + if err := (WorkflowRunner{}).Execute(f); err != nil { + t.Fatal(err) + } + raw, err := os.ReadFile(filepath.Join(dir, "ERR")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(raw), "不能调用 run_python") || len(f.tools) != 0 { + t.Fatalf("err %s tools %+v", raw, f.tools) + } +} + +func TestScriptStopsAfterCallCap(t *testing.T) { + body := `import agentwf as wf +for i in range(60): + wf.tool("time_now", {}) +` + f := &wfFake{ + sandbox: t.TempDir(), + settings: scriptSettings(), + launch: workflow.Launch{ + Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ + ID: "run", Title: "跑", Kind: "script", Script: "a.py", + }}}, + Scripts: map[string]string{"a.py": body}, + }, + } + err := (WorkflowRunner{}).Execute(f) + if err == nil || !strings.Contains(err.Error(), "超过 50") || len(f.tools) != 50 { + t.Fatalf("err %v calls %d", err, len(f.tools)) + } +} + +func TestScriptCancelKillsProcess(t *testing.T) { + dir := t.TempDir() + body := `import os, time +open("PID", "w").write(str(os.getpid())) +time.sleep(30) +open("AFTER", "w").write("ran") +` + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &wfFake{ + ctx: ctx, + sandbox: dir, + settings: scriptSettings(), + launch: workflow.Launch{ + Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ + ID: "run", Title: "跑", Kind: "script", Script: "a.py", + }}}, + Scripts: map[string]string{"a.py": body}, + }, + } + done := make(chan error, 1) + go func() { done <- (WorkflowRunner{}).Execute(f) }() + deadline := time.Now().Add(5 * time.Second) + for { + if _, err := os.Stat(filepath.Join(dir, "PID")); err == nil { + break + } + if time.Now().After(deadline) { + cancel() + t.Fatal("script did not start") + } + time.Sleep(20 * time.Millisecond) + } + raw, err := os.ReadFile(filepath.Join(dir, "PID")) + if err != nil { + t.Fatal(err) + } + cancel() + select { + case err = <-done: + if err == nil || !strings.Contains(err.Error(), "取消") { + t.Fatalf("%v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("cancel did not stop the script") + } + if _, err := os.Stat("/proc/" + strings.TrimSpace(string(raw))); err == nil { + t.Fatal("script process still alive") + } + if _, err := os.Stat(filepath.Join(dir, "AFTER")); err == nil { + t.Fatal("script kept running after cancel") + } +} + +func TestScriptLineLimit(t *testing.T) { + body := `import os +f = os.fdopen(3, "w", encoding="utf-8", buffering=1) +f.write("x" * (1024 * 1024 + 32) + "\n") +f.flush() +` + f := &wfFake{ + sandbox: t.TempDir(), + settings: scriptSettings(), + launch: workflow.Launch{ + Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ + ID: "run", Title: "跑", Kind: "script", Script: "a.py", + }}}, + Scripts: map[string]string{"a.py": body}, + }, + } + err := (WorkflowRunner{}).Execute(f) + if err == nil || !strings.Contains(err.Error(), "1MB") { + t.Fatal(err) + } +} + +func TestRegisteredWorkflowUsesItsOwnScriptCopy(t *testing.T) { + f := &wfFake{sandbox: t.TempDir(), settings: scriptSettings()} + res, err := (WorkflowRunner{}).run(f, workflow.Launch{ + Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ + ID: "run", Title: "跑", Kind: "script", Script: "workflows/scripts/demo.py", + }}}, + Scripts: map[string]string{"workflows/scripts/demo.py": "print('from-launch')\n"}, + }) + if err != nil || !strings.Contains(res.Summary, "完成") { + t.Fatalf("err %v summary %s", err, res.Summary) + } +} + +func TestModelScriptNeedsSwitchAndConfirm(t *testing.T) { + f := &wfFake{sandbox: t.TempDir()} + out := (&directRun{rc: f}).runRegisteredWorkflow(`{"script":"print('pong')"}`) + if !strings.Contains(out, "未开启") || len(f.hitlQ) != 0 { + t.Fatalf("%s questions %v", out, f.hitlQ) + } + + f2 := &wfFake{sandbox: t.TempDir(), settings: scriptSettings(), hitl: []core.HITLResponse{{ChoiceID: "no"}}} + out = (&directRun{rc: f2}).runRegisteredWorkflow(`{"script":"print('pong')"}`) + if out != "未确认,未执行" || !strings.Contains(f2.hitlQ[0], "print('pong')") { + t.Fatalf("%s questions %v", out, f2.hitlQ) + } + + f3 := &wfFake{sandbox: t.TempDir(), settings: scriptSettings(), hitl: []core.HITLResponse{{ChoiceID: "yes"}}} + out = (&directRun{rc: f3}).runRegisteredWorkflow(`{"script":"print('pong')"}`) + if !strings.Contains(out, "pong") { + t.Fatalf("%s", out) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index d3f7c9a..927f17a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -22,6 +22,7 @@ import ( "agent-system-backend/internal/cardsvc" "agent-system-backend/internal/cmdbus" "agent-system-backend/internal/ingest" + "agent-system-backend/internal/intentgate" "agent-system-backend/internal/plugins" "agent-system-backend/internal/profiles" "agent-system-backend/internal/prompts" @@ -30,6 +31,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 +44,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 @@ -68,6 +71,18 @@ type NLIntentGate interface { Match(ctx context.Context, input string) (string, string) } +type intentRecognizer interface { + Recognize(ctx context.Context, input string) intentgate.MatchResult +} + +func (s *Server) recognize(ctx context.Context, text string) intentgate.MatchResult { + if rg, ok := s.Gate.(intentRecognizer); ok { + return rg.Recognize(ctx, text) + } + branch, id := s.Gate.Match(ctx, text) + return intentgate.MatchResult{Branch: branch, ID: id} +} + // WireBuiltinCommands 注册内置命令(`/` 菜单全清单的数据源;处理器为宿主用例适配)。 // 响应保持既有 {command, ...} 契约(frontend store.sendMessage 按此消费)。 func (s *Server) WireBuiltinCommands() { @@ -560,14 +575,15 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) { writeStoreErr(w, err) return } - // chat 入口同源(D27):意图闸门命中且命令总线承接 → 统一分发(同审计),不起 run。 - if s.Bus != nil && s.Gate != nil && !strings.HasPrefix(text, "/") { - if branch, topID := s.Gate.Match(r.Context(), text); branch == "matched" && s.Bus.HasCommand(topID) { + // 先命令总线,再绑定工作流,最后才是普通回合。 + if s.Gate != nil && !strings.HasPrefix(text, "/") { + recognized := s.recognize(r.Context(), text) + if recognized.Branch == "matched" && s.Bus != nil && s.Bus.HasCommand(recognized.ID) { resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{ - ID: topID, Params: map[string]any{"text": text}, Entrance: "chat", Actor: owner, + ID: recognized.ID, Params: map[string]any{"text": text}, Entrance: "chat", Actor: owner, Owner: owner, ConversationID: convID, AnchorMessageID: msg.ID, }) - out := map[string]any{"message": msg, "receipt": resp.Receipt, "command": topID} + out := map[string]any{"message": msg, "receipt": resp.Receipt, "command": recognized.ID} if resp.Card != nil { out["card"] = resp.Card } @@ -579,6 +595,17 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, out) return } + if recognized.Branch == "matched" && s.Workflows != nil { + if def, ok := s.Workflows.ByIntent(recognized.ID); ok { + missing := make([]workflow.SlotQuestion, len(recognized.Missing)) + for i, q := range recognized.Missing { + missing[i] = workflow.SlotQuestion{Key: q.Key, Question: q.Question} + } + run, err := s.mgr.StartWorkflowRun(conv, msg, def, recognized.Slots, missing) + writeJSONOr(w, map[string]any{"message": msg, "run": run}, err) + return + } + } } run, err := s.mgr.StartConversationRun(conv, msg, parsedDocs, inputs) writeJSONOr(w, map[string]any{"message": msg, "run": run}, err) diff --git a/internal/server/workflow_route_test.go b/internal/server/workflow_route_test.go new file mode 100644 index 0000000..2a7be82 --- /dev/null +++ b/internal/server/workflow_route_test.go @@ -0,0 +1,111 @@ +package server + +import ( + "context" + "encoding/json" + "testing" + "time" + + agentcommand "agent-command" + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/intentgate" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + "agent-system-backend/internal/workflow" + core "agent-system-core" +) + +type routeCapture struct { + ch chan workflow.Launch +} + +func (c routeCapture) Name() string { return "route-capture" } + +func (c routeCapture) Execute(rc runtime.RunCtx) error { + src, _ := rc.(interface { + Workflow() (workflow.Launch, bool) + }) + if src == nil { + return nil + } + if launch, ok := src.Workflow(); ok { + c.ch <- launch + } + return nil +} + +func TestCommandBusBeatsWorkflow(t *testing.T) { + srv, st := newTestServer(t) + if err := st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": false}}); err != nil { + t.Fatal(err) + } + ch := make(chan workflow.Launch, 1) + srv.mgr.SetWorkflowRunner(routeCapture{ch}) + gate := intentgate.New() + if err := gate.Rebuild(context.Background(), []json.RawMessage{briefIntent()}); err != nil { + t.Fatal(err) + } + srv.Gate = gate + reg := workflow.NewRegistry() + reg.Set([]core.WorkflowDef{{ + ID: "demo.brief", Title: "简报", IntentID: "demo.brief", + Steps: []core.WorkflowStep{{ID: "s", Kind: "hitl"}}, + }}, nil) + srv.Workflows = reg + if err := srv.Bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.brief", Title: "简报命令"}, "", func(context.Context, cmdbus.Request) (any, error) { + return map[string]any{"ok": true}, nil + }); err != nil { + t.Fatal(err) + } + _, convID := setupConversation(t, srv.Handler()) + code, out := doJSON(t, srv.Handler(), "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "写一份简报"}) + if code != 200 || out["command"] != "demo.brief" { + t.Fatalf("status %d body %v", code, out) + } + if _, ok := out["run"]; ok { + t.Fatal("a command hit must not start a workflow run") + } + select { + case <-ch: + t.Fatal("workflow runner started for a command") + default: + } +} + +func TestIntentStartsWorkflowRun(t *testing.T) { + srv, st := newTestServer(t) + if err := st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": false}}); err != nil { + t.Fatal(err) + } + ch := make(chan workflow.Launch, 1) + srv.mgr.SetWorkflowRunner(routeCapture{ch}) + gate := intentgate.New() + if err := gate.Rebuild(context.Background(), []json.RawMessage{briefIntent()}); err != nil { + t.Fatal(err) + } + srv.Gate = gate + reg := workflow.NewRegistry() + reg.Set([]core.WorkflowDef{{ + ID: "demo.brief", Title: "简报", IntentID: "demo.brief", + Steps: []core.WorkflowStep{{ID: "s", Kind: "hitl"}}, + }}, nil) + srv.Workflows = reg + _, convID := setupConversation(t, srv.Handler()) + code, out := doJSON(t, srv.Handler(), "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "写一份简报"}) + if code != 200 || out["run"] == nil { + t.Fatalf("status %d body %v", code, out) + } + select { + case launch := <-ch: + if launch.Def.ID != "demo.brief" || len(launch.Missing) != 1 || launch.Missing[0].Question != "主题是什么?" { + t.Fatalf("%+v", launch) + } + case <-time.After(time.Second): + t.Fatal("workflow runner was not selected") + } +} + +func briefIntent() json.RawMessage { + return json.RawMessage(`{"id":"demo.brief","domain":"demo","version":1,"matcher":{"includes_any":["简报"]},"slots":[{"key":"topic","required":true,"pattern":"主题[::]\\s*(\\S+)","question":"主题是什么?"}]}`) +} diff --git a/internal/tools/exec_interpreter.go b/internal/tools/exec_interpreter.go index c79e3df..4de4138 100644 --- a/internal/tools/exec_interpreter.go +++ b/internal/tools/exec_interpreter.go @@ -211,6 +211,34 @@ func execShell(rc ToolExecCtx, args json.RawMessage) (string, error) { return runInterpreter(execParentCtx(rc), "bash", p, sandbox, firstPos(a.TimeoutS, a.TimeoutSec)) } +// ScriptCommand 准备一条以当前用户身份、工作目录为会话沙箱的 python3 命令。 +// 调用方在 Start 之前设置 ExtraFiles、Stdout 和 Stderr。取消 ctx 会杀掉整个进程组。 +func ScriptCommand(ctx context.Context, sandbox, path string) (*exec.Cmd, error) { + if ctx == nil { + ctx = context.Background() + } + bin, err := resolveInterpreter("python3") + if err != nil { + return nil, err + } + cmd := exec.CommandContext(ctx, bin, path) + cmd.Dir = sandbox + cmd.Env = execEnv() + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.WaitDelay = 2 * time.Second + cmd.Cancel = func() error { + if cmd.Process == nil { + return os.ErrProcessDone + } + err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + if errors.Is(err, syscall.ESRCH) { + return nil + } + return err + } + return cmd, nil +} + func execPython(rc ToolExecCtx, args json.RawMessage) (string, error) { if err := execGate(rc); err != nil { return "", err diff --git a/internal/workflow/registry.go b/internal/workflow/registry.go new file mode 100644 index 0000000..da77d32 --- /dev/null +++ b/internal/workflow/registry.go @@ -0,0 +1,107 @@ +// 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 +} + +// SlotQuestion 是启动工作流时还没填上的必填槽位。 +type SlotQuestion struct { + Key string + Question string +} + +// Launch 是一次工作流 run 启动时的副本。之后插件卸载不能改到它。 +type Launch struct { + Def core.WorkflowDef + Slots map[string]string + Missing []SlotQuestion + Scripts map[string]string // 键是步骤里的相对路径,值是当时读到的脚本正文 +} + +// Clone 拷贝定义,调用方之后的修改不影响副本。 +func Clone(def core.WorkflowDef) core.WorkflowDef { return cloneDef(def) } + +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 +} diff --git a/internal/workflow/render.go b/internal/workflow/render.go new file mode 100644 index 0000000..f82a482 --- /dev/null +++ b/internal/workflow/render.go @@ -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.}}、{{steps..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 +} diff --git a/internal/workflow/validate.go b/internal/workflow/validate.go new file mode 100644 index 0000000..12453fa --- /dev/null +++ b/internal/workflow/validate.go @@ -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 +} diff --git a/internal/workflow/workflow_test.go b/internal/workflow/workflow_test.go new file mode 100644 index 0000000..41084bd --- /dev/null +++ b/internal/workflow/workflow_test.go @@ -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") + } +}