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 <cursoragent@cursor.com>
This commit is contained in:
2026-10-01 16:18:59 +00:00
co-authored by Cursor
parent 9a563df970
commit fb5f071761
10 changed files with 919 additions and 14 deletions
+25
View File
@@ -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()
+50
View File
@@ -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")
}
}
+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)
+26 -3
View File
@@ -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)
+6 -7
View File
@@ -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 {
+38 -4
View File
@@ -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 }
+464
View File
@@ -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)
}
+230
View File
@@ -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)
}
}
+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
+1
View File
@@ -88,6 +88,7 @@ type Launch struct {
Def core.WorkflowDef
Slots map[string]string
Missing []SlotQuestion
Scripts map[string]string // 键是步骤里的相对路径,值是当时读到的脚本正文
}
// Clone 拷贝定义,调用方之后的修改不影响副本。