Execute intent, planned, and script workflows #47

Merged
weichen merged 9 commits from feature/agent-harness-p4 into main 2026-10-02 01:03:55 +00:00
27 changed files with 2776 additions and 57 deletions
+8
View File
@@ -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
+58 -17
View File
@@ -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 报告是否装载了意图(诊断/设置展示用)。
+26
View File
@@ -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)
}
}
+80 -6
View File
@@ -16,6 +16,7 @@ import (
"agent-system-backend/internal/settings"
"agent-system-backend/internal/store"
"agent-system-backend/internal/tools"
"agent-system-backend/internal/workflow"
core "agent-system-core"
)
@@ -26,6 +27,7 @@ type Loaded struct {
Panels []core.PluginPanel
Cards map[string][]core.UINode // kind → 模板节点
Workflows []core.WorkflowDef
Scripts map[string]string // workflow.ScriptKey → 插件目录内的脚本正文
Intents []json.RawMessage // agent-intent overlay 形态(目录侧消费)
Commands []json.RawMessage // Agent-Command spec overlay(cmdbus 注册)
Workspaces []core.PluginWorkspace // 工作区弹层声明(OverlayHost 渲染)
@@ -57,6 +59,8 @@ type Host struct {
OnIntents func(overlay []json.RawMessage) error
// OnCommands 在装载集合变化后回调全部命令 spec overlay(宿主接命令总线;nil = 不接)。
OnCommands func(overlay []json.RawMessage) error
// OnWorkflows 在装载集合变化后回调全部工作流与脚本正文(nil = 不接)。
OnWorkflows func(defs []core.WorkflowDef, scripts map[string]string) error
}
func New(st *store.Store, reg *tools.Registry, s *settings.Service, dir string) *Host {
@@ -96,7 +100,8 @@ func (h *Host) LoadDir() map[string]error {
}
overlay := h.collectIntentsLocked()
cmdOverlay := h.collectCommandsLocked()
onIntents, onCommands := h.OnIntents, h.OnCommands
defs, scripts := h.collectWorkflowsLocked()
onIntents, onCommands, onWorkflows := h.OnIntents, h.OnCommands, h.OnWorkflows
h.mu.Unlock()
if onCommands != nil {
if err := onCommands(cmdOverlay); err != nil {
@@ -108,6 +113,11 @@ func (h *Host) LoadDir() map[string]error {
results["<intents>"] = err
}
}
if onWorkflows != nil {
if err := onWorkflows(defs, scripts); err != nil {
results["<workflows>"] = err
}
}
return results
}
@@ -132,7 +142,7 @@ func (h *Host) loadOne(dir string) error {
return fmt.Errorf("plugins: %s: apiVersion %s 不在兼容窗口(%s, %s)", m.ID, m.APIVersion, core.PluginAPIVersion, core.PluginAPIVersionExperimental)
}
ld := &Loaded{Manifest: m, Dir: dir, Cards: map[string][]core.UINode{}, Prompts: map[string]string{}, Defaults: map[string]any{}}
ld := &Loaded{Manifest: m, Dir: dir, Cards: map[string][]core.UINode{}, Prompts: map[string]string{}, Defaults: map[string]any{}, Scripts: map[string]string{}}
// 同 ID 重载:先注销旧注册(含旧版本工具),再装载新资产。
h.registry.UnregisterBySource(core.SourcePlugin, m.ID)
@@ -178,12 +188,28 @@ func (h *Host) loadOne(dir string) error {
ld.Cards[card.Kind] = card.Data
}
// workflows/*.json
// workflows/*.json。script 步骤的文件必须留在插件目录内,正文随定义一起交给注册表。
for _, name := range m.Provides.Workflows {
var wf core.WorkflowDef
if err := readAsset(dir, "workflows", name, &wf); err != nil {
return fmt.Errorf("plugins: %s: workflow %s: %w", m.ID, name, err)
}
if err := workflow.Validate(wf, workflow.DeclarativeLimits(), func(tool string) bool {
_, _, ok := h.registry.Get(tool)
return ok
}); err != nil {
return fmt.Errorf("plugins: %s: workflow %s: %w", m.ID, name, err)
}
for _, step := range wf.Steps {
if step.Kind != "script" {
continue
}
body, err := readPluginFile(dir, step.Script)
if err != nil {
return fmt.Errorf("plugins: %s: workflow %s script: %w", m.ID, name, err)
}
ld.Scripts[workflow.ScriptKey(wf.ID, step.Script)] = string(body)
}
ld.Workflows = append(ld.Workflows, wf)
}
@@ -245,7 +271,8 @@ func (h *Host) loadOne(dir string) error {
h.loaded[m.ID] = ld
overlay := h.collectIntentsLocked()
cmdOverlay := h.collectCommandsLocked()
onIntents, onCommands := h.OnIntents, h.OnCommands
defs, scripts := h.collectWorkflowsLocked()
onIntents, onCommands, onWorkflows := h.OnIntents, h.OnCommands, h.OnWorkflows
h.mu.Unlock()
_ = h.st.UpsertPlugin(m.ID, dir, m)
if onCommands != nil {
@@ -258,6 +285,11 @@ func (h *Host) loadOne(dir string) error {
return fmt.Errorf("plugins: %s: intents rejected by host gate (kept previous): %w", m.ID, err)
}
}
if onWorkflows != nil {
if err := onWorkflows(defs, scripts); err != nil {
return fmt.Errorf("plugins: %s: workflows rejected: %w", m.ID, err)
}
}
return nil
}
@@ -290,6 +322,37 @@ func (h *Host) collectCommandsLocked() []json.RawMessage {
return out
}
func (h *Host) collectWorkflowsLocked() ([]core.WorkflowDef, map[string]string) {
ids := make([]string, 0, len(h.loaded))
for id := range h.loaded {
ids = append(ids, id)
}
sort.Strings(ids)
var defs []core.WorkflowDef
scripts := map[string]string{}
for _, id := range ids {
ld := h.loaded[id]
defs = append(defs, ld.Workflows...)
for k, v := range ld.Scripts {
scripts[k] = v
}
}
return defs, scripts
}
func readPluginFile(root, rel string) ([]byte, error) {
cleaned := filepath.Clean(rel)
if cleaned == "." || filepath.IsAbs(cleaned) {
return nil, fmt.Errorf("path %s escapes plugin dir", rel)
}
full := filepath.Join(root, cleaned)
rooted, err := filepath.Rel(root, full)
if err != nil || rooted == ".." || strings.HasPrefix(rooted, ".."+string(filepath.Separator)) {
return nil, fmt.Errorf("path %s escapes plugin dir", rel)
}
return os.ReadFile(full)
}
func readAsset(dir, sub, name string, v any) error {
p := filepath.Join(dir, sub, name+".json")
if sub == "" {
@@ -345,12 +408,23 @@ func (h *Host) Unregister(id string) {
delete(h.loaded, id)
cmd := h.sidecars[id]
delete(h.sidecars, id)
overlay := h.collectIntentsLocked()
cmdOverlay := h.collectCommandsLocked()
defs, scripts := h.collectWorkflowsLocked()
onIntents, onCommands, onWorkflows := h.OnIntents, h.OnCommands, h.OnWorkflows
h.mu.Unlock()
if cmd != nil && cmd.Process != nil {
_ = cmd.Process.Kill()
}
h.registry.UnregisterBySource(core.SourcePlugin, id)
_ = h.st.RemovePlugin(id)
if onCommands != nil {
_ = onCommands(cmdOverlay)
}
if onIntents != nil {
_ = onIntents(overlay)
}
if onWorkflows != nil {
_ = onWorkflows(defs, scripts)
}
}
var _ = strings.TrimSpace
+97
View File
@@ -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)
}
}
@@ -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 问用户。
+1
View File
@@ -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 {
+1 -1
View File
@@ -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,
+8 -4
View File
@@ -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())
}
}
+88 -12
View File
@@ -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
}
+132
View File
@@ -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")
}
}
+51
View File
@@ -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)
+14 -6
View File
@@ -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{
+84
View File
@@ -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 关闭);内容直接写在对话回复里"
+30 -1
View File
@@ -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 {
+423
View File
@@ -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}
}
+347
View File
@@ -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)
}
}
+468
View File
@@ -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)
}
+243
View File
@@ -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)
}
}
+36 -9
View File
@@ -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)
+111
View File
@@ -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":"主题是什么?"}]}`)
}
+28
View File
@@ -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
+107
View File
@@ -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
}
+89
View File
@@ -0,0 +1,89 @@
package workflow
import (
"encoding/json"
"fmt"
"regexp"
core "agent-system-core"
)
// Vars 是一次执行里可以填进模板的值。
type Vars struct {
Input string
Slots map[string]string
Steps map[string]string
}
var placeholder = regexp.MustCompile(`\{\{(input|slots\.[A-Za-z_][A-Za-z0-9_]*|steps\.[A-Za-z_][A-Za-z0-9_]*\.output)\}\}`)
// Render 替换 {{input}}、{{slots.<key>}}、{{steps.<id>.output}}。缺值替换成空串。
func Render(s string, v Vars) string {
return placeholder.ReplaceAllStringFunc(s, func(token string) string {
body := token[2 : len(token)-2]
switch {
case body == "input":
return v.Input
case len(body) > 6 && body[:6] == "slots.":
return v.Slots[body[6:]]
case len(body) > 6 && body[:6] == "steps.":
id := body[6 : len(body)-len(".output")]
return v.Steps[id]
default:
return token
}
})
}
// RenderRaw 在 JSON 字符串内部替换占位符,结果仍是合法 JSON。
func RenderRaw(raw json.RawMessage, v Vars) (json.RawMessage, error) {
if len(raw) == 0 {
return nil, nil
}
var node any
if err := json.Unmarshal(raw, &node); err != nil {
return nil, fmt.Errorf("workflow: 参数不是合法 JSON")
}
out, err := json.Marshal(walk(node, v))
if err != nil {
return nil, err
}
return out, nil
}
func walk(node any, v Vars) any {
switch x := node.(type) {
case string:
return Render(x, v)
case []any:
for i := range x {
x[i] = walk(x[i], v)
}
return x
case map[string]any:
for k, val := range x {
x[k] = walk(val, v)
}
return x
default:
return node
}
}
// RenderStep 渲染这一步会交给执行器的文本和参数。
func RenderStep(step core.WorkflowStep, v Vars) (core.WorkflowStep, error) {
step.Prompt = Render(step.Prompt, v)
step.Artifact = Render(step.Artifact, v)
if step.Hitl != nil {
step.Hitl.Question = Render(step.Hitl.Question, v)
}
if step.Card != nil {
step.Card.Title = Render(step.Card.Title, v)
}
args, err := RenderRaw(step.Args, v)
if err != nil {
return step, err
}
step.Args = args
return step, nil
}
+114
View File
@@ -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
}
+102
View File
@@ -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")
}
}