From fb5f07176194e975a1c3551c002e7f6531dcaeb7 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 1 Oct 2026 16:18:59 +0000 Subject: [PATCH] Run workflow scripts through a capped callback host Why: Plugin scripts and model-written scripts need the same step runner, without giving the script a way to call interpreters or start another workflow. Evidence: go test ./... passed. Callbacks, denied tools, the 50-call cap, a 1MB line, and cancel-kills-the-process are covered. A model script does not run until the switch is on and the user confirms. Impact: runtime.workflowScripts stays off by default. Plugin script bodies are copied into the launch. Squad entry still has no run_workflow. Refs: feature/agent-harness-p4 Co-authored-by: Cursor --- internal/runmgr/runmgr.go | 25 ++ internal/runmgr/workflow_route_test.go | 50 +++ internal/runtime/agentwf.py | 51 +++ internal/runtime/llm_direct.go | 29 +- internal/runtime/workflow.go | 13 +- internal/runtime/workflow_exec_test.go | 42 +- internal/runtime/workflow_script.go | 464 +++++++++++++++++++++++ internal/runtime/workflow_script_test.go | 230 +++++++++++ internal/tools/exec_interpreter.go | 28 ++ internal/workflow/registry.go | 1 + 10 files changed, 919 insertions(+), 14 deletions(-) create mode 100644 internal/runtime/agentwf.py create mode 100644 internal/runtime/workflow_script.go create mode 100644 internal/runtime/workflow_script_test.go diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 95a06d8..f4cdaeb 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -169,6 +169,14 @@ func (c *runCtx) WorkflowByID(id string) (core.WorkflowDef, bool) { 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 } func (c *runCtx) Run() core.Run { return c.run } func (c *runCtx) Profile() core.AgentProfile { return c.prof } @@ -761,10 +769,27 @@ func (m *Manager) StartWorkflowRun(conv core.Conversation, userMessage core.Mess 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 { m.convMu.Lock() defer m.convMu.Unlock() diff --git a/internal/runmgr/workflow_route_test.go b/internal/runmgr/workflow_route_test.go index d0e8340..bdc3634 100644 --- a/internal/runmgr/workflow_route_test.go +++ b/internal/runmgr/workflow_route_test.go @@ -80,3 +80,53 @@ func TestStartWorkflowRunCopiesDefinition(t *testing.T) { t.Fatal("workflow runner was not selected") } } + +func TestStartWorkflowRunCopiesScriptBody(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "wf.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + if err := st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": true}}); err != nil { + t.Fatal(err) + } + reg := tools.NewRegistry() + prof := profiles.New(st) + if err := prof.EnsureDefault(); err != nil { + t.Fatal(err) + } + direct := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "direct"}}} + cap := launchCapture{ch: make(chan workflow.Launch, 1)} + mgr := New(st, bus.New(), reg, prof, settings.NewService(st), direct, noopRunner{}, &runtime.MockRunner{}, &runtime.MockSubagent{}) + mgr.SetWorkflowRunner(cap) + wf := workflow.NewRegistry() + const path = "workflows/scripts/demo.py" + wf.Set([]core.WorkflowDef{{ + ID: "demo.script", Title: "脚本", + Steps: []core.WorkflowStep{{ID: "run", Kind: "script", Script: path}}, + }}, map[string]string{workflow.ScriptKey("demo.script", path): "print(1)\n"}) + mgr.SetWorkflows(wf) + + proj, err := st.CreateProject(store.OwnerDefault, "p") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation(store.OwnerDefault, proj.ID, "c", "") + if err != nil { + t.Fatal(err) + } + msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "跑"}}} + def, _ := wf.Get("demo.script") + if _, err := mgr.StartWorkflowRun(conv, msg, def, nil, nil); err != nil { + t.Fatal(err) + } + wf.Set(nil, nil) + select { + case launch := <-cap.ch: + if launch.Scripts[path] != "print(1)\n" { + t.Fatalf("%+v", launch.Scripts) + } + case <-time.After(time.Second): + t.Fatal("workflow runner was not selected") + } +} diff --git a/internal/runtime/agentwf.py b/internal/runtime/agentwf.py new file mode 100644 index 0000000..d31d477 --- /dev/null +++ b/internal/runtime/agentwf.py @@ -0,0 +1,51 @@ +"""工作流脚本回调。用法：import agentwf as wf。 + +请求写到 fd 3，响应从 fd 4 按行读回。宿主一次只处理一条。 +""" + +import json +import os + +_seq = 0 +_out = os.fdopen(3, "w", encoding="utf-8", buffering=1) +_inp = os.fdopen(4, "r", encoding="utf-8", buffering=1) + + +def _rpc(op, **fields): + global _seq + _seq += 1 + msg = {"id": str(_seq), "op": op} + msg.update(fields) + _out.write(json.dumps(msg, ensure_ascii=False) + "\n") + _out.flush() + line = _inp.readline() + if not line: + raise RuntimeError("workflow host closed") + resp = json.loads(line) + if not resp.get("ok"): + raise RuntimeError(resp.get("error") or "workflow call failed") + return resp.get("output", "") + + +def tool(name, args=None): + return _rpc("tool.call", name=name, args=args or {}) + + +def ask(question, choices=None): + return _rpc("hitl.ask", question=question, choices=choices or []) + + +def agent(name, task): + return _rpc("agent.spawn", agent=name, task=task) + + +def artifact(name, content): + return _rpc("artifact.register", name=name, content=content) + + +def card(title, kind="result"): + return _rpc("card.create", title=title, kind=kind) + + +def step(step_id, title, status): + return _rpc("step", stepId=step_id, title=title, status=status) diff --git a/internal/runtime/llm_direct.go b/internal/runtime/llm_direct.go index e3875f3..660aca5 100644 --- a/internal/runtime/llm_direct.go +++ b/internal/runtime/llm_direct.go @@ -11,7 +11,6 @@ import ( agentllm "agent-native/runtime/llm" agenttools "agent-native/runtime/tools" "agent-system-backend/internal/squadadapter" - "agent-system-backend/internal/workflow" core "agent-system-core" ) @@ -478,7 +477,7 @@ func (st *directRun) runRegisteredWorkflow(args string) string { return st.runPlannedWorkflow(strings.TrimSpace(a.Goal)) } if strings.TrimSpace(a.Script) != "" { - return "这种启动方式还没接上" + return st.runScriptWorkflow(a.Script) } lookup, ok := st.rc.(interface { WorkflowByID(id string) (core.WorkflowDef, bool) @@ -490,7 +489,7 @@ func (st *directRun) runRegisteredWorkflow(args string) string { if !ok { return "未注册的工作流 " + a.Workflow } - res, err := (WorkflowRunner{}).run(st.rc, workflow.Launch{Def: def, Slots: a.Inputs}) + res, err := (WorkflowRunner{}).run(st.rc, registeredWorkflowLaunch(st.rc, def, a.Inputs)) st.pinSubagentArtifacts(res.Summary) if err != nil { if res.Summary != "" { @@ -501,6 +500,30 @@ func (st *directRun) runRegisteredWorkflow(args string) string { 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) diff --git a/internal/runtime/workflow.go b/internal/runtime/workflow.go index 69f006b..f050bb6 100644 --- a/internal/runtime/workflow.go +++ b/internal/runtime/workflow.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "strings" + "time" agentllm "agent-native/runtime/llm" "agent-system-backend/internal/workflow" @@ -16,6 +17,8 @@ const workflowPlannerID = "agent.sub.workflow-plan" type WorkflowRunner struct { // JudgeStream 注入验收模型。nil 时按轻量模型构造客户端。 JudgeStream agentllm.StreamFunc + // ScriptTimeout 覆盖脚本墙钟超时。0 表示 600 秒，超过 1800 秒会被截断。 + ScriptTimeout time.Duration } func (WorkflowRunner) Name() string { return "workflow" } @@ -213,7 +216,7 @@ func (w WorkflowRunner) runStepWithCritic(rc RunCtx, def core.WorkflowDef, step } else if note != "" { current.Prompt = "未通过验收：" + note } - output, got, err := runWorkflowStep(rc, current, vars, prev) + output, got, err := w.runWorkflowStep(rc, current, vars, prev) arts = append(arts, got...) if err != nil { return output, arts, err.Error(), err @@ -239,7 +242,7 @@ func (w WorkflowRunner) runStepWithCritic(rc RunCtx, def core.WorkflowDef, step } } -func runWorkflowStep(rc RunCtx, step core.WorkflowStep, vars workflow.Vars, prev string) (string, []*core.ArtifactRef, error) { +func (w WorkflowRunner) runWorkflowStep(rc RunCtx, step core.WorkflowStep, vars workflow.Vars, prev string) (string, []*core.ArtifactRef, error) { switch step.Kind { case "prompt": return runPromptStep(rc, step) @@ -299,7 +302,7 @@ func runWorkflowStep(rc RunCtx, step core.WorkflowStep, vars workflow.Vars, prev } return string(created.ID), nil, nil case "script": - return executeScriptStep(rc, step, vars) + return executeScriptStep(rc, w, step) default: return "", nil, fmt.Errorf("workflow: 未知步骤类型 %s", step.Kind) } @@ -346,10 +349,6 @@ func artifactsIn(rc RunCtx, text string) []*core.ArtifactRef { return out } -func executeScriptStep(rc RunCtx, step core.WorkflowStep, vars workflow.Vars) (string, []*core.ArtifactRef, error) { - return "", nil, fmt.Errorf("workflow: 脚本步骤还不能执行") -} - func (w WorkflowRunner) judgeStep(rc RunCtx, goal, criteria, output string) verifyVerdict { stream := w.JudgeStream if stream == nil { diff --git a/internal/runtime/workflow_exec_test.go b/internal/runtime/workflow_exec_test.go index be6b325..d4f7c99 100644 --- a/internal/runtime/workflow_exec_test.go +++ b/internal/runtime/workflow_exec_test.go @@ -26,6 +26,9 @@ type wfFake struct { spawnText string spawnIDs []string toolNames []string + settings core.EffectiveSettings + sandbox string + scripts map[string]string } func (f *wfFake) Context() context.Context { @@ -34,9 +37,27 @@ func (f *wfFake) Context() context.Context { } 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 { return core.EffectiveSettings{} } +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 @@ -71,11 +92,24 @@ func (f *wfFake) UpdatePlan(p core.PlanSnapshot) error { 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)) - return core.Artifact{ID: "art_aaaaaaaaaaaa", Name: name, Kind: kind}, nil + 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 } diff --git a/internal/runtime/workflow_script.go b/internal/runtime/workflow_script.go new file mode 100644 index 0000000..78cc7b3 --- /dev/null +++ b/internal/runtime/workflow_script.go @@ -0,0 +1,464 @@ +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) (string, []*core.ArtifactRef, error) { + body, err := scriptBody(rc, step.Script) + if err != nil { + return "", nil, err + } + sandbox, err := scriptSandbox(rc) + if err != nil { + return "", nil, err + } + path, err := writeScriptFile(sandbox, "step-"+step.ID, body) + if err != nil { + return "", nil, err + } + return runScriptFile(rc, runner, path) +} + +func scriptBody(rc RunCtx, path string) (string, error) { + src, ok := rc.(interface { + Workflow() (workflow.Launch, bool) + }) + if !ok { + return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) + } + launch, ok := src.Workflow() + if !ok { + return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) + } + body, ok := launch.Scripts[path] + if !ok || strings.TrimSpace(body) == "" { + return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) + } + return body, nil +} + +func registeredWorkflowLaunch(rc RunCtx, def core.WorkflowDef, slots map[string]string) workflow.Launch { + launch := workflow.Launch{Def: def, Slots: slots, Scripts: map[string]string{}} + lookup, ok := rc.(interface { + WorkflowScript(id, path string) (string, bool) + }) + if !ok { + return launch + } + for _, step := range def.Steps { + if step.Kind != "script" || strings.TrimSpace(step.Script) == "" { + continue + } + if body, found := lookup.WorkflowScript(def.ID, step.Script); found { + launch.Scripts[step.Script] = body + } + } + return launch +} + +func runModelScript(rc RunCtx, runner WorkflowRunner, body string) (string, []*core.ArtifactRef, error) { + if !scriptsEnabled(rc) { + return "", nil, fmt.Errorf("workflow: 工作流脚本未开启") + } + preview := body + if len(preview) > 4000 { + preview = preview[:4000] + "\n…（脚本超长，已截断展示）" + } + resp, err := rc.AskHITL(core.HITLCard{ + Question: "运行这段脚本？\n" + preview, Kind: core.HITLKindConfirm, AllowFreeInput: false, + Choices: []core.HITLChoice{ + {ID: "yes", Label: "运行", Description: "在会话沙箱里执行", Impact: "会调用工具并可能写入工件"}, + {ID: "no", Label: "取消", Description: "不执行", Impact: "不会改动任何东西"}, + }, + }) + if err != nil { + return "", nil, err + } + if resp.ChoiceID != "yes" { + return "未确认，未执行", nil, nil + } + sandbox, err := scriptSandbox(rc) + if err != nil { + return "", nil, err + } + path, err := writeScriptFile(sandbox, "model", body) + if err != nil { + return "", nil, err + } + return runScriptFile(rc, runner, path) +} + +func writeScriptFile(sandbox, prefix, body string) (string, error) { + dir := filepath.Join(sandbox, ".agent") + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + path := filepath.Join(dir, fmt.Sprintf("wf-%s-%d.py", prefix, time.Now().UnixNano())) + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + return "", err + } + return path, nil +} + +func runScriptFile(rc RunCtx, runner WorkflowRunner, path string) (string, []*core.ArtifactRef, error) { + if !scriptsEnabled(rc) { + return "", nil, fmt.Errorf("workflow: 工作流脚本未开启") + } + sandbox, err := scriptSandbox(rc) + if err != nil { + return "", nil, err + } + timeout := runner.ScriptTimeout + if timeout <= 0 { + timeout = scriptDefaultTimeout + } + if timeout > scriptMaxTimeout { + timeout = scriptMaxTimeout + } + helper, err := agentwfHelperDir() + if err != nil { + return "", nil, err + } + parent := rc.Context() + if parent == nil { + parent = context.Background() + } + ctx, cancel := context.WithTimeout(parent, timeout) + defer cancel() + cmd, err := tools.ScriptCommand(ctx, sandbox, path) + if err != nil { + return "", nil, err + } + cmd.Env = prependPythonPath(cmd.Env, helper) + reqR, reqW, err := os.Pipe() + if err != nil { + return "", nil, err + } + respR, respW, err := os.Pipe() + if err != nil { + reqR.Close() + reqW.Close() + return "", nil, err + } + cmd.ExtraFiles = []*os.File{reqW, respR} + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + if err := cmd.Start(); err != nil { + reqR.Close() + reqW.Close() + respR.Close() + respW.Close() + return "", nil, err + } + reqW.Close() + respR.Close() + defer reqR.Close() + defer respW.Close() + arts, callErr := serveScript(ctx, rc, reqR, respW) + if callErr != nil { + cancel() + } + waitErr := cmd.Wait() + text := clipScriptOutput(&stdout, &stderr) + if callErr != nil { + return text, arts, callErr + } + if ctx.Err() == context.DeadlineExceeded { + return text, arts, fmt.Errorf("workflow: 脚本超时") + } + if ctx.Err() == context.Canceled { + return text, arts, fmt.Errorf("workflow: 脚本已取消") + } + if waitErr != nil { + return text, arts, fmt.Errorf("workflow: 脚本退出：%v\n%s", waitErr, strings.TrimSpace(stderr.String())) + } + return text, arts, nil +} + +func clipScriptOutput(stdout, stderr *bytes.Buffer) string { + text := strings.TrimSpace(stdout.String()) + errText := strings.TrimSpace(stderr.String()) + if errText != "" { + if text != "" { + text += "\n" + } + text += errText + } + if len(text) > 16000 { + text = text[:16000] + "…（超长截断）" + } + return text +} + +func prependPythonPath(env []string, dir string) []string { + for i, kv := range env { + if strings.HasPrefix(kv, "PYTHONPATH=") { + env[i] = "PYTHONPATH=" + dir + string(os.PathListSeparator) + strings.TrimPrefix(kv, "PYTHONPATH=") + return env + } + } + return append(env, "PYTHONPATH="+dir) +} + +type scriptRequest struct { + ID string `json:"id"` + Op string `json:"op"` + Name string `json:"name"` + Args json.RawMessage `json:"args"` + Question string `json:"question"` + Choices []core.HITLChoice `json:"choices"` + Agent string `json:"agent"` + Task string `json:"task"` + Content string `json:"content"` + Title string `json:"title"` + Kind string `json:"kind"` + Status string `json:"status"` + StepID string `json:"stepId"` +} + +type scriptResponse struct { + ID string `json:"id"` + OK bool `json:"ok"` + Output string `json:"output,omitempty"` + Error string `json:"error,omitempty"` +} + +func serveScript(ctx context.Context, rc RunCtx, reqR, respW *os.File) ([]*core.ArtifactRef, error) { + sc := bufio.NewScanner(reqR) + sc.Buffer(make([]byte, 64*1024), scriptMaxLine) + var arts []*core.ArtifactRef + calls := 0 + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + calls++ + var req scriptRequest + if err := json.Unmarshal([]byte(line), &req); err != nil { + _ = writeScriptResponse(respW, scriptResponse{OK: false, Error: "请求不是合法 JSON"}) + continue + } + if calls > scriptMaxCalls { + _ = writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: false, Error: "脚本回调超过 50 次"}) + return arts, fmt.Errorf("workflow: 脚本回调超过 50 次") + } + if ctx.Err() != nil { + return arts, ctx.Err() + } + output, got, err := dispatchScript(rc, req) + arts = append(arts, got...) + if err != nil { + _ = writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: false, Error: err.Error()}) + if ctx.Err() != nil { + return arts, ctx.Err() + } + continue + } + if err := writeScriptResponse(respW, scriptResponse{ID: req.ID, OK: true, Output: output}); err != nil { + return arts, err + } + } + if err := sc.Err(); err != nil { + if err == bufio.ErrTooLong { + return arts, fmt.Errorf("workflow: 单行请求超过 1MB") + } + if ctx.Err() != nil { + return arts, ctx.Err() + } + return arts, err + } + return arts, nil +} + +func writeScriptResponse(w io.Writer, resp scriptResponse) error { + raw, err := json.Marshal(resp) + if err != nil { + return err + } + _, err = w.Write(append(raw, '\n')) + return err +} + +func dispatchScript(rc RunCtx, req scriptRequest) (string, []*core.ArtifactRef, error) { + switch req.Op { + case "tool.call": + if _, denied := workflow.DeniedTools[req.Name]; denied { + return "", nil, fmt.Errorf("不能调用 %s", req.Name) + } + res := rc.ExecuteTool(core.ToolCallRequest{Name: req.Name, Arguments: req.Args}) + if res.Status == "error" { + return "", nil, fmt.Errorf("%s", res.Output) + } + var arts []*core.ArtifactRef + if res.Artifact != nil { + arts = append(arts, &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind}) + } + arts = append(arts, artifactsIn(rc, res.Output)...) + return res.Output, arts, nil + case "hitl.ask": + card := core.HITLCard{Question: req.Question, Kind: core.HITLKindInput, AllowFreeInput: true} + if len(req.Choices) > 0 { + card.Kind = core.HITLKindChoice + card.AllowFreeInput = false + card.Choices = req.Choices + for i := range card.Choices { + if card.Choices[i].Impact == "" { + card.Choices[i].Impact = "会按这个选择继续" + } + } + } + resp, err := rc.AskHITL(card) + if err != nil { + return "", nil, err + } + if strings.TrimSpace(resp.FreeText) != "" { + return strings.TrimSpace(resp.FreeText), nil, nil + } + return resp.ChoiceID, nil, nil + case "agent.spawn": + agent := req.Agent + if strings.TrimSpace(agent) == "" { + agent = "general" + } + var res core.AgentResultPayload + var err error + if sp, ok := rc.(interface { + SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) + }); ok { + res, err = sp.SpawnAgent(subagentProfileID(agent), req.Title, req.Task, "") + } else { + res, err = rc.SpawnSubagent(req.Title, req.Task, "") + } + if err != nil { + return "", nil, err + } + text := res.Text + if text == "" { + text = res.Summary + } + return text, artifactsIn(rc, text), nil + case "artifact.register": + name := req.Name + if name == "" { + name = "script.md" + } + art, err := rc.WriteArtifact(name, "document", "text/markdown", []byte(req.Content)) + if err != nil { + return "", nil, err + } + return string(art.ID), []*core.ArtifactRef{{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}}, nil + case "card.create": + kind := core.CardKind(req.Kind) + if kind == "" { + kind = core.CardKindResult + } + title := req.Title + if title == "" { + title = req.Name + } + created, err := rc.CreateCard(core.ActivityCard{Title: title, Kind: kind}) + if err != nil { + return "", nil, err + } + return string(created.ID), nil, nil + case "step": + applyScriptPlan(rc, req.StepID, req.Title, req.Status) + return req.Status, nil, nil + default: + return "", nil, fmt.Errorf("未知回调 %s", req.Op) + } +} + +func applyScriptPlan(rc RunCtx, id, title, status string) { + switch status { + case "pending", "in_progress", "done", "skipped": + default: + status = "done" + } + snap := core.PlanSnapshot{} + if src, ok := rc.(interface { + Plan() (core.PlanSnapshot, bool) + }); ok { + if cur, found := src.Plan(); found { + snap = cur + snap.Steps = append([]core.PlanStep(nil), cur.Steps...) + } + } + found := false + for i := range snap.Steps { + if snap.Steps[i].ID == id { + snap.Steps[i].Status = status + if title != "" { + snap.Steps[i].Text = title + } + found = true + break + } + } + if !found { + snap.Steps = append(snap.Steps, core.PlanStep{ID: id, Text: title, Status: status}) + } + _ = rc.UpdatePlan(snap) +} diff --git a/internal/runtime/workflow_script_test.go b/internal/runtime/workflow_script_test.go new file mode 100644 index 0000000..d14a456 --- /dev/null +++ b/internal/runtime/workflow_script_test.go @@ -0,0 +1,230 @@ +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 TestModelScriptNeedsSwitchAndConfirm(t *testing.T) { + f := &wfFake{sandbox: t.TempDir()} + out := (&directRun{rc: f}).runRegisteredWorkflow(`{"script":"print('pong')"}`) + if !strings.Contains(out, "未开启") || len(f.hitlQ) != 0 { + t.Fatalf("%s questions %v", out, f.hitlQ) + } + + f2 := &wfFake{sandbox: t.TempDir(), settings: scriptSettings(), hitl: []core.HITLResponse{{ChoiceID: "no"}}} + out = (&directRun{rc: f2}).runRegisteredWorkflow(`{"script":"print('pong')"}`) + if out != "未确认，未执行" || !strings.Contains(f2.hitlQ[0], "print('pong')") { + t.Fatalf("%s questions %v", out, f2.hitlQ) + } + + f3 := &wfFake{sandbox: t.TempDir(), settings: scriptSettings(), hitl: []core.HITLResponse{{ChoiceID: "yes"}}} + out = (&directRun{rc: f3}).runRegisteredWorkflow(`{"script":"print('pong')"}`) + if !strings.Contains(out, "pong") { + t.Fatalf("%s", out) + } +} diff --git a/internal/tools/exec_interpreter.go b/internal/tools/exec_interpreter.go index c79e3df..4de4138 100644 --- a/internal/tools/exec_interpreter.go +++ b/internal/tools/exec_interpreter.go @@ -211,6 +211,34 @@ func execShell(rc ToolExecCtx, args json.RawMessage) (string, error) { return runInterpreter(execParentCtx(rc), "bash", p, sandbox, firstPos(a.TimeoutS, a.TimeoutSec)) } +// ScriptCommand 准备一条以当前用户身份、工作目录为会话沙箱的 python3 命令。 +// 调用方在 Start 之前设置 ExtraFiles、Stdout 和 Stderr。取消 ctx 会杀掉整个进程组。 +func ScriptCommand(ctx context.Context, sandbox, path string) (*exec.Cmd, error) { + if ctx == nil { + ctx = context.Background() + } + bin, err := resolveInterpreter("python3") + if err != nil { + return nil, err + } + cmd := exec.CommandContext(ctx, bin, path) + cmd.Dir = sandbox + cmd.Env = execEnv() + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.WaitDelay = 2 * time.Second + cmd.Cancel = func() error { + if cmd.Process == nil { + return os.ErrProcessDone + } + err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + if errors.Is(err, syscall.ESRCH) { + return nil + } + return err + } + return cmd, nil +} + func execPython(rc ToolExecCtx, args json.RawMessage) (string, error) { if err := execGate(rc); err != nil { return "", err diff --git a/internal/workflow/registry.go b/internal/workflow/registry.go index 8da01a1..da77d32 100644 --- a/internal/workflow/registry.go +++ b/internal/workflow/registry.go @@ -88,6 +88,7 @@ type Launch struct { Def core.WorkflowDef Slots map[string]string Missing []SlotQuestion + Scripts map[string]string // 键是步骤里的相对路径，值是当时读到的脚本正文 } // Clone 拷贝定义，调用方之后的修改不影响副本。