Add the WorkflowDef-to-squad compiler with host intent steps (Q3 WP1)
ci / vet-and-test (pull_request) Successful in 44s

Why: Q3 WP1 requires core.WorkflowDef workflows to run as compiled squad
graphs (P4 limits and DeniedTools enforced at the compile entry) so the
legacy SpawnSubagent-compression executor can be retired in WP2.

Evidence: internal/wfcompile conformance (zero model, zero network; script
step runs the real codeorch host like the legacy tests): demo.brief shape
(hitl/prompt+critic/artifact/card mapping, evaluator-optimizer loop budget
= maxRetries+1, needs chain), P4 limits (dynamic >8 rejected, dynamic
script rejected, DeniedTools, critic >3, workflow retries >6), demo.brief
E2E (critic fail->retry->pass, drops record, artifact from step-ref file,
card, hitl sequence without plan confirmation for registered workflows,
BB out_length visible in the draft task brief), critic exhaustion fails
closed, demo.script E2E (plugin script exempt from confirmation, wf.tool/
ask/artifact/step callbacks), workflowScripts off fails, model script
double confirmation with reject paths. Full backend suite green (SELF);
CI authoritative (GATE).

Impact: new package internal/wfcompile (compiler + host IntentRunner +
RunCompiled driver); nested deps/agent-squad 06934bc -> 633e853 (StepSpec
.Brief). No existing backend behavior changes (legacy executor untouched
until WP2).

Refs: Q3 plan WP1 (2026-10-07-agent-stack-q2q3-plans.md); task card
.task/feature-agent-stack-q3.md
This commit is contained in:
2026-10-08 02:36:39 +00:00
parent 06934bccab
commit 017c96baa5
5 changed files with 1923 additions and 1 deletions
+389
View File
@@ -0,0 +1,389 @@
// Package wfcompile 把工作流定义(core.WorkflowDef)编译为 Agent-Squad 计划
// (Q3 WP1):六类步骤映射到 squad 图步/宿主 intent 步,StepCritic 映射为
// evaluator-optimizer 循环对,P4 全部上限与 DeniedTools 经 internal/workflow
// 的 Validate 在编译入口强制(声明式 ≤20 步;模型生成 ≤8 步、禁 script、
// 执行前必须确认;critic 每步 ≤3 缺省 1、整条 ≤6;脚本回调 ≤50、单行 ≤1MB、
// 超时 600/1800s 由宿主脚本执行器持有)。
//
// 步骤映射(方案 Q3-WP1):
// - prompt → 角色步(Brief 携带渲染后指令;agent 字段映射角色键)
// - hitl → 宿主 intent 步 wf.hitl(AskHITL 卡片,答案落 output.md + 黑板 out_<id>)
// - tool → 宿主 intent 步 wf.tool(ExecuteTool;DeniedTools 编译期已拦)
// - artifact → 宿主 intent 步 wf.artifact(WriteArtifact)
// - card → 宿主 intent 步 wf.card(CreateCard)
// - script → 宿主 intent 步 wf.script(codeorch 宿主执行;插件脚本免确认,
// 模型脚本必须确认;runtime.workflowScripts 开关默认关)
// - critic → 追加 wf.critic 评审步 + Loop{Body:[原步]}(evaluator-optimizer
// 形态):verdict 黑板字段驱动再武装,预算 = 步重试上限 + 1;无法解析的
// 判定按不通过处理(fail-closed,与 legacy judgeStep 同款)。
//
// 模板占位符:{{input}}/{{slots.*}} 在编译期渲染(槽位已在启动前收齐);
// {{steps.X.output}} 编译为文件指引(宿主步全文落 tasks/NN_X/output.md,
// 宿主步要点回填黑板 out_X)——语义桥接,非逐字节等价,差异由 WP5 A/B 记录。
package wfcompile
import (
"encoding/json"
"fmt"
"regexp"
"strings"
"agent-squad/squad"
"agent-system-backend/internal/workflow"
core "agent-system-core"
)
// 宿主 intent 步的意图 ID(目录由本包合成,处理函数在 host.go)。
const (
IntentHitl = "wf.hitl"
IntentTool = "wf.tool"
IntentArtifact = "wf.artifact"
IntentCard = "wf.card"
IntentScript = "wf.script"
IntentCritic = "wf.critic"
)
// CriticPair 记录一个「原步 ↔ 评审步」映射(收口后校验最终判定用)。
type CriticPair struct {
StepID string // 被评审的原步 ID
StepTitle string
EvalID string // 评审步 ID(<StepID>_critic)
VerdictField string // 黑板判定字段名
}
// Compiled 是一次编译产物。
type Compiled struct {
Plan *squad.SquadPlan
Catalog *squad.IntentCatalog
Critics []CriticPair
// SubOf 步骤 ID → 子工作区相对路径(宿主处理器解析跨步文件引用)。
SubOf map[string]string
Def core.WorkflowDef
}
// CompileOptions 编译口径。
type CompileOptions struct {
// DynamicLimits 模型生成工作流:≤8 步、禁 script(P4)。
DynamicLimits bool
// ModelScript 模型提供的脚本(run_workflow script= 分支):脚本步执行前
// 必须 HITL 确认(P4:模型脚本必须确认;插件脚本免确认)。
ModelScript bool
}
// Compile 把 def 编译为 squad 计划。vars 提供 {{input}}/{{slots.*}} 的编译期
// 实值(缺失槽位须已由调用方收齐)。knownTools 报告工具是否已注册
// (nil = 只查禁用名单)。
func Compile(def core.WorkflowDef, vars workflow.Vars, opts CompileOptions, knownTools func(string) bool) (*Compiled, error) {
lim := workflow.DeclarativeLimits()
if opts.DynamicLimits {
lim = workflow.DynamicLimits()
}
if err := workflow.Validate(def, lim, knownTools); err != nil {
return nil, err
}
c := &Compiled{SubOf: map[string]string{}, Def: def}
var steps []squad.StepSpec
var agents []squad.RoleSpec
agentSeen := map[string]bool{}
prevDef := "" // 上一个 def 步 ID(artifact 空内容的 prev 回退)
prevChain := "" // 图链上的前驱(含评审步)
for i, raw := range def.Steps {
// {{steps.X.output}} 是运行期引用:先用哨兵保护,避免 RenderStep 按
// 空值吞掉;渲染后按字段语义还原(人读字段 → 文件指引文本;机器字段
// → 原 token,由宿主处理器运行期解析)。
raw = protectStepRefs(raw)
rendered, err := workflow.RenderStep(raw, vars)
if err != nil {
return nil, err
}
sub := fmt.Sprintf("tasks/%02d_%s", i+1, raw.ID)
c.SubOf[raw.ID] = sub
var needs []string
if prevChain != "" {
needs = []string{prevChain}
}
var sp squad.StepSpec
switch raw.Kind {
case "prompt":
role := agentRole(raw.Agent)
if !agentSeen[role] {
agentSeen[role] = true
agents = append(agents, squad.RoleSpec{Role: role, Name: role})
}
sp = squad.StepSpec{
ID: raw.ID, Role: role, Title: rendered.Title,
SubWorkspace: sub, Needs: needs, Targets: []string{"output.md"},
Brief: c.refsToText(rendered.Prompt),
}
case "hitl":
question := c.refsToText(rendered.Title)
kind := string(core.HITLKindInput)
choicesJSON := ""
allowFree := "true"
hint := ""
if rendered.Hitl != nil {
if strings.TrimSpace(rendered.Hitl.Question) != "" {
question = c.refsToText(rendered.Hitl.Question)
}
if rendered.Hitl.Kind != "" {
kind = string(rendered.Hitl.Kind)
}
if len(rendered.Hitl.Choices) > 0 {
b, _ := json.Marshal(rendered.Hitl.Choices)
choicesJSON = string(b)
}
allowFree = boolSlot(rendered.Hitl.AllowFreeInput)
hint = rendered.Hitl.FreeInputHint
}
sp = intentStep(raw.ID, rendered.Title, sub, needs, IntentHitl, map[string]string{
"step_id": raw.ID, "question": question, "kind": kind,
"choices": choicesJSON, "allow_free_input": allowFree, "free_input_hint": hint,
}, []string{"output.md", "fields.json"})
case "tool":
sp = intentStep(raw.ID, rendered.Title, sub, needs, IntentTool, map[string]string{
"step_id": raw.ID, "tool": strings.TrimSpace(rendered.Tool), "args": restoreStepRefs(string(rendered.Args)),
}, []string{"output.md", "fields.json"})
case "artifact":
name := rendered.Artifact
if strings.TrimSpace(name) == "" {
name = raw.ID + ".md"
}
content := restoreStepRefs(rendered.Prompt)
slots := map[string]string{"step_id": raw.ID, "name": name, "content": content}
if strings.TrimSpace(content) == "" && prevDef != "" {
// legacy 语义:artifact 无正文时取上一步产出。
slots["content_ref"] = prevDef
}
sp = intentStep(raw.ID, rendered.Title, sub, needs, IntentArtifact, slots,
[]string{"output.md", "fields.json"})
case "card":
tmpl := core.ActivityCardTemplate{Kind: core.CardKindResult, Title: c.refsToText(rendered.Title)}
if rendered.Card != nil {
tmpl = *rendered.Card
tmpl.Title = c.refsToText(tmpl.Title)
if tmpl.Title == "" {
tmpl.Title = c.refsToText(rendered.Title)
}
}
b, _ := json.Marshal(tmpl)
sp = intentStep(raw.ID, rendered.Title, sub, needs, IntentCard, map[string]string{
"step_id": raw.ID, "template": string(b),
}, []string{"output.md", "fields.json"})
case "script":
sp = intentStep(raw.ID, rendered.Title, sub, needs, IntentScript, map[string]string{
"step_id": raw.ID, "path": strings.TrimSpace(raw.Script), "model": boolSlot(opts.ModelScript),
}, []string{"output.md", "fields.json"})
default:
return nil, fmt.Errorf("workflow: 未知步骤类型 %s", raw.Kind)
}
steps = append(steps, sp)
prevDef = raw.ID
prevChain = raw.ID
// critic → evaluator-optimizer 循环对:原步产出后评审步判定,
// verdict != pass 且预算未尽 → 原步与评审步一起再武装。
if raw.Critic != nil && strings.TrimSpace(raw.Critic.Criteria) != "" {
retries := 1
if raw.Critic.MaxRetries > 0 {
retries = raw.Critic.MaxRetries
}
if retries > workflow.MaxStepRetries {
retries = workflow.MaxStepRetries
}
field := "verdict_" + raw.ID
evalID := raw.ID + "_critic"
evalSub := fmt.Sprintf("tasks/%02d_%s_critic", i+1, raw.ID)
c.SubOf[evalID] = evalSub
steps = append(steps, intentStep(evalID, "验收:"+raw.Title, evalSub,
[]string{raw.ID}, IntentCritic, map[string]string{
"step_id": evalID,
"target_step": raw.ID,
"criteria": raw.Critic.Criteria,
"goal": def.Title,
"verdict_field": field,
"max_attempts": fmt.Sprintf("%d", retries+1),
}, []string{"verdict.md", "fields.json"}))
last := &steps[len(steps)-1]
last.Loop = &squad.LoopSpec{
MaxIterations: retries + 1,
UntilField: field,
UntilEquals: "pass",
Body: []string{raw.ID},
}
c.Critics = append(c.Critics, CriticPair{
StepID: raw.ID, StepTitle: raw.Title, EvalID: evalID, VerdictField: field,
})
prevChain = evalID
}
}
agents = append(agents, squad.DefaultIntentRoleSpec())
c.Plan = &squad.SquadPlan{
Squad: squad.SquadConfig{SquadName: "wf_" + def.ID, Version: "1", Agents: agents},
Workflow: squad.WorkflowPlan{
WorkflowID: squad.NewPlanID(),
Title: def.Title,
PlanningMode: squad.PlanModeBlueprint,
Status: "pending",
Steps: steps,
},
}
c.Catalog = buildCatalog()
return c, nil
}
// PlanPreview 动态工作流的确认预览(legacy planPreview 同款文案)。
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")
}
// ParseWorkflowDef 宽松解析模型输出的 WorkflowDef(legacy parseWorkflowDef 同款)。
func ParseWorkflowDef(raw string) (core.WorkflowDef, error) {
s := stripFenceBlock(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 stripFenceBlock(s string) string {
s = strings.TrimSpace(s)
if !strings.HasPrefix(s, "```") {
return s
}
s = strings.TrimPrefix(s, "```")
if i := strings.IndexByte(s, '\n'); i >= 0 {
s = s[i+1:]
}
return strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(s), "```"))
}
// intentStep 组装一个宿主 intent 步。
func intentStep(id, title, sub string, needs []string, intentID string, slots map[string]string, targets []string) squad.StepSpec {
return squad.StepSpec{
ID: id, Role: squad.DefaultIntentRole, Title: title,
SubWorkspace: sub, Needs: needs, Targets: targets,
IntentID: intentID, IntentSlots: slots,
}
}
// agentRole 把 WorkflowStep.Agent 规整为角色键(空 = general)。
func agentRole(agent string) string {
a := strings.TrimSpace(agent)
if a == "" {
return "general"
}
var b strings.Builder
for _, r := range strings.ToLower(a) {
switch {
case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '_':
b.WriteRune(r)
default:
b.WriteRune('_')
}
}
return b.String()
}
func boolSlot(v bool) string {
if v {
return "true"
}
return ""
}
// stepTokenRe 匹配 {{steps.<id>.output}};sentinelRe 匹配编译期哨兵。
var (
stepTokenRe = regexp.MustCompile(`\{\{steps\.([A-Za-z_][A-Za-z0-9_]*)\.output\}\}`)
sentinelRe = regexp.MustCompile(`@@stepsref:([A-Za-z_][A-Za-z0-9_]*)@@`)
)
// protectStepRefs 把步骤定义里的运行期引用换成哨兵(RenderStep 只认识
// {{input}}/{{slots.*}}/{{steps.*}},不保护会被空值替换吞掉)。
func protectStepRefs(step core.WorkflowStep) core.WorkflowStep {
step.Prompt = stepTokenRe.ReplaceAllString(step.Prompt, "@@stepsref:$1@@")
step.Artifact = stepTokenRe.ReplaceAllString(step.Artifact, "@@stepsref:$1@@")
if step.Hitl != nil {
q := stepTokenRe.ReplaceAllString(step.Hitl.Question, "@@stepsref:$1@@")
hitl := *step.Hitl
hitl.Question = q
step.Hitl = &hitl
}
if step.Card != nil {
card := *step.Card
card.Title = stepTokenRe.ReplaceAllString(card.Title, "@@stepsref:$1@@")
step.Card = &card
}
if len(step.Args) > 0 {
step.Args = json.RawMessage(stepTokenRe.ReplaceAllString(string(step.Args), "@@stepsref:$1@@"))
}
return step
}
// refsToText 哨兵 → 文件指引文本(人读/LLM 读字段:Brief、问题、卡标题)。
func (c *Compiled) refsToText(s string) string {
return sentinelRe.ReplaceAllStringFunc(s, func(m string) string {
id := sentinelRe.FindStringSubmatch(m)[1]
return fmt.Sprintf("〈步骤 %s 的产出〉(全文见运行根目录 %s/output.md;宿主步要点见黑板字段 out_%s)",
id, c.SubOf[id], id)
})
}
// restoreStepRefs 哨兵 → 原 token(机器字段:工具实参、工件正文,由宿主
// 处理器运行期按文件解析实值)。
func restoreStepRefs(s string) string {
return sentinelRe.ReplaceAllStringFunc(s, func(m string) string {
id := sentinelRe.FindStringSubmatch(m)[1]
return "{{steps." + id + ".output}}"
})
}
// buildCatalog 合成宿主 intent 目录(validateIntentSteps 的确定性校验面)。
func buildCatalog() *squad.IntentCatalog {
spec := func(id, desc string, slots ...squad.IntentSlotView) squad.IntentSpecView {
v := squad.IntentSpecView{ID: id, Description: desc, Handler: "wfcompile", Slots: slots}
for _, sl := range slots {
if sl.Required {
v.RequiredSlots = append(v.RequiredSlots, sl.Key)
}
}
return v
}
req := func(key string) squad.IntentSlotView {
return squad.IntentSlotView{Key: key, Required: true, Question: "缺少必填槽位 " + key}
}
opt := func(key string) squad.IntentSlotView { return squad.IntentSlotView{Key: key} }
catalog := &squad.IntentCatalog{
Revision: 1,
Markdown: "# 工作流宿主能力\n\n- wf.hitl 人工问答\n- wf.tool 宿主工具\n- wf.artifact 工件落库\n- wf.card 收件箱卡\n- wf.script 脚本宿主\n- wf.critic 步骤验收\n",
Specs: map[string]squad.IntentSpecView{
IntentHitl: spec(IntentHitl, "人工问答步", req("question"), opt("step_id"), opt("kind"), opt("choices"), opt("allow_free_input"), opt("free_input_hint")),
IntentTool: spec(IntentTool, "宿主工具步", req("tool"), opt("step_id"), opt("args")),
IntentArtifact: spec(IntentArtifact, "工件步", req("name"), opt("step_id"), opt("content"), opt("content_ref")),
IntentCard: spec(IntentCard, "收件箱卡步", req("template"), opt("step_id")),
IntentScript: spec(IntentScript, "脚本步", req("path"), opt("step_id"), opt("model")),
IntentCritic: spec(IntentCritic, "步骤验收", req("target_step"), req("criteria"), req("verdict_field"),
opt("step_id"), opt("goal"), opt("max_attempts")),
},
}
return catalog
}
+586
View File
@@ -0,0 +1,586 @@
package wfcompile
import (
"context"
"encoding/json"
"fmt"
"os"
"os/exec"
"path/filepath"
"strings"
"sync"
"time"
"agent-native/runtime/codeorch"
"agent-native/runtime/gates"
agentllm "agent-native/runtime/llm"
"agent-squad/squad"
"agent-system-backend/internal/tools"
"agent-system-backend/internal/workflow"
core "agent-system-core"
)
// Host 是编译图执行所需的宿主窄端口(结构性接口;runtime.RunCtx 天然满足,
// 避免 wfcompile 反向依赖 runtime 形成环)。
type Host interface {
Context() context.Context
Run() core.Run
Settings() core.EffectiveSettings
Emit(typ core.StreamEventType, payload any) error
AskHITL(card core.HITLCard) (core.HITLResponse, error)
UpdatePlan(p core.PlanSnapshot) error
WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error)
ExecuteTool(call core.ToolCallRequest) core.ToolCallResult
CreateCard(card core.ActivityCard) (core.ActivityCard, error)
ConversationSandbox() string
Tools() []core.ToolDesc
}
// Spawner 可选子代理派发能力(脚本回调 agent.spawn 用)。
type Spawner interface {
SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error)
SpawnSubagent(name, task string, hint string) (core.AgentResultPayload, error)
}
// PlanGetter 可选当前计划读取(脚本回调 step 用)。
type PlanGetter interface {
Plan() (core.PlanSnapshot, bool)
}
// Deps 是编译图执行的装配依赖。
type Deps struct {
// DataDir 小队运行数据根(运行目录 = DataDir/wf/<runID>)。
DataDir string
// Backends 步骤后端解析器(WF1 期由调用方装配 agent-system;WP2 切 agent-native)。
Backends *squad.BackendResolver
// DefaultBackend 角色未声明后端时的缺省(空 = agent-system)。
DefaultBackend string
// JudgeStream 验收模型流(nil = 验证器不可用,判定 fail-closed 不通过)。
JudgeStream agentllm.StreamFunc
// ScriptTimeout 覆盖脚本墙钟超时(0 = 600s;上限 1800s 由 codeorch 持有)。
ScriptTimeout time.Duration
}
// 脚本宿主限额(P4:回调 ≤50、单行 ≤1MB、超时 600/1800s)。
const (
scriptDefaultTimeout = 600 * time.Second
scriptMaxTimeout = 1800 * time.Second
scriptMaxCalls = 50
scriptMaxLine = 1 << 20
scriptModelPreview = 4000
// outFieldCap 宿主步产出回填黑板 out_<id> 字段的截断上限。
outFieldCap = 2000
)
// hostRunner 宿主 intent 步执行器(squad.IntentRunner 实现)。
type hostRunner struct {
rc Host
deps Deps
scripts map[string]string
subOf map[string]string
mu sync.Mutex
verdicts map[string]string // verdict 字段 → 最近一次判定
fails map[string]int // verdict 字段 → 累计不通过次数(触顶即败)
lastWhy map[string]string // 原步 ID → 最近一次不通过原因
drops []string // 丢弃记录(legacy drops 同款 "stepID:why")
arts []*core.ArtifactRef // 收集到的工件引用
}
func (r *hostRunner) Run(ctx context.Context, step squad.IntentStep) (*squad.ExecResult, error) {
var content string
var fields map[string]string
var err error
switch step.IntentID {
case IntentHitl:
content, fields, err = r.doHitl(step)
case IntentTool:
content, fields, err = r.doTool(ctx, step)
case IntentArtifact:
content, fields, err = r.doArtifact(step)
case IntentCard:
content, fields, err = r.doCard(step)
case IntentScript:
content, fields, err = r.doScript(ctx, step)
case IntentCritic:
content, fields, err = r.doCritic(ctx, step)
default:
err = fmt.Errorf("workflow: 未知宿主意图 %s", step.IntentID)
}
if err != nil {
return &squad.ExecResult{Success: false, ErrorReason: err.Error()}, nil
}
patches := map[string][]byte{}
main := "output.md"
if len(step.Targets) > 0 {
main = step.Targets[0]
}
// squad 交付物契约拒收空白文件(MaterializeTargets);legacy 允许空产出,
// 编译图用占位文本保持流转(out_ 字段仍按原始内容判空)。
body := content
if strings.TrimSpace(body) == "" {
body = "(本步无文本产出)"
}
patches[main] = []byte(body + "\n")
// 宿主步产出要点回填黑板(out_<step_id>),供后续步骤 TASK.md 快照引用。
if fields == nil {
fields = map[string]string{}
}
if sid := step.Slots["step_id"]; sid != "" && strings.TrimSpace(content) != "" {
fields["out_"+sid] = truncate(content, outFieldCap)
}
// fields.json 是声明目标之一,恒写(空对象也写,避免交付物校验失败)。
b, _ := json.Marshal(fields)
patches["fields.json"] = b
return &squad.ExecResult{Success: true, Patches: patches}, nil
}
// doHitl 人工问答步(legacy runWorkflowStep case "hitl" 同款)。
func (r *hostRunner) doHitl(step squad.IntentStep) (string, map[string]string, error) {
kind := core.HITLKind(step.Slots["kind"])
if kind == "" {
kind = core.HITLKindInput
}
card := core.HITLCard{
Question: step.Slots["question"],
Kind: kind,
AllowFreeInput: step.Slots["allow_free_input"] == "true",
FreeInputHint: step.Slots["free_input_hint"],
}
if raw := step.Slots["choices"]; strings.TrimSpace(raw) != "" {
_ = json.Unmarshal([]byte(raw), &card.Choices)
}
resp, err := r.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
}
// doTool 宿主工具步(DeniedTools 已在编译期 Validate 拦截)。
func (r *hostRunner) doTool(_ context.Context, step squad.IntentStep) (string, map[string]string, error) {
name := step.Slots["tool"]
args := r.renderRefs(step.Slots["args"], step.WorkDir)
res := r.rc.ExecuteTool(core.ToolCallRequest{Name: name, Arguments: json.RawMessage(args)})
if res.Status == "error" {
return "", nil, fmt.Errorf("workflow: 工具 %s 失败:%s", name, res.Output)
}
if res.Artifact != nil {
r.track(&core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind})
}
return res.Output, nil, nil
}
// doArtifact 工件步(空正文回退 content_ref 指向步骤的产出,legacy prev 语义)。
func (r *hostRunner) doArtifact(step squad.IntentStep) (string, map[string]string, error) {
content := r.renderRefs(step.Slots["content"], step.WorkDir)
if strings.TrimSpace(content) == "" {
if ref := step.Slots["content_ref"]; ref != "" {
content = r.readStepOutput(step.WorkDir, ref)
}
}
name := step.Slots["name"]
if strings.TrimSpace(name) == "" {
name = step.Slots["step_id"] + ".md"
}
art, err := r.rc.WriteArtifact(name, "document", "text/markdown", []byte(content))
if err != nil {
return "", nil, err
}
r.track(&core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind})
return string(art.ID), nil, nil
}
// doCard 收件箱卡步。
func (r *hostRunner) doCard(step squad.IntentStep) (string, map[string]string, error) {
var tmpl core.ActivityCardTemplate
if err := json.Unmarshal([]byte(step.Slots["template"]), &tmpl); err != nil {
return "", nil, fmt.Errorf("workflow: 卡片模板不合法: %v", err)
}
card := core.ActivityCard{Title: tmpl.Title, Kind: tmpl.Kind, Tone: tmpl.Tone, Body: tmpl.Body, Actions: tmpl.Actions}
if card.Kind == "" {
card.Kind = core.CardKindResult
}
created, err := r.rc.CreateCard(card)
if err != nil {
return "", nil, err
}
return string(created.ID), nil, nil
}
// doScript 脚本步:插件脚本免确认;模型脚本(model=true)必须先过 HITL 确认;
// runtime.workflowScripts 开关默认关(P4)。执行走 codeorch 宿主(限额同款)。
func (r *hostRunner) doScript(ctx context.Context, step squad.IntentStep) (string, map[string]string, error) {
path := step.Slots["path"]
body := r.scripts[path]
if strings.TrimSpace(body) == "" {
return "", nil, fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path)
}
if step.Slots["model"] == "true" {
if !r.scriptsEnabled() {
return "", nil, fmt.Errorf("workflow: 工作流脚本未开启")
}
preview := body
if len(preview) > scriptModelPreview {
preview = preview[:scriptModelPreview] + "\n…(脚本超长,已截断展示)"
}
resp, err := r.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 := strings.TrimSpace(r.rc.ConversationSandbox())
if sandbox == "" {
return "", nil, fmt.Errorf("workflow: 没有会话沙箱")
}
file, err := r.writeScriptFile(sandbox, "step-"+step.Slots["step_id"], body)
if err != nil {
return "", nil, err
}
return r.runScriptFile(ctx, sandbox, file)
}
func (r *hostRunner) scriptsEnabled() bool {
return r.rc.Settings().GetBool("runtime.workflowScripts", false)
}
func (r *hostRunner) 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
}
// runScriptFile codeorch 宿主执行(legacy runScriptFile 同款限额与回调分发)。
func (r *hostRunner) runScriptFile(ctx context.Context, sandbox, path string) (string, map[string]string, error) {
if !r.scriptsEnabled() {
return "", nil, fmt.Errorf("workflow: 工作流脚本未开启")
}
timeout := r.deps.ScriptTimeout
if timeout <= 0 {
timeout = scriptDefaultTimeout
}
out := codeorch.RunFile(ctx, sandbox, path, codeorch.Limits{
Timeout: timeout,
MaxTimeout: scriptMaxTimeout,
MaxCalls: scriptMaxCalls,
MaxResultBytes: scriptMaxLine,
MaxModelChars: 16000,
FullDir: sandbox,
}, func(c context.Context, dir, scriptPath string) (*exec.Cmd, error) {
return tools.ScriptCommand(c, dir, scriptPath)
}, func(_ context.Context, call codeorch.Call) (string, error) {
output, got, err := r.dispatchScript(call)
r.trackAll(got)
return output, err
})
return out.ModelText(), nil, out.Err
}
// dispatchScript 脚本回调分发(legacy dispatchScript 同款语义与文案)。
func (r *hostRunner) dispatchScript(req codeorch.Call) (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 := r.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})
}
return res.Output, arts, nil
case "hitl.ask":
var choices []core.HITLChoice
if len(req.Choices) > 0 {
_ = json.Unmarshal(req.Choices, &choices)
}
card := core.HITLCard{Question: req.Question, Kind: core.HITLKindInput, AllowFreeInput: true}
if len(choices) > 0 {
card.Kind = core.HITLKindChoice
card.AllowFreeInput = false
card.Choices = choices
for i := range card.Choices {
if card.Choices[i].Impact == "" {
card.Choices[i].Impact = "会按这个选择继续"
}
}
}
resp, err := r.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":
sp, ok := r.rc.(Spawner)
if !ok {
return "", nil, fmt.Errorf("当前运行不能派发子代理")
}
agent := req.Agent
if strings.TrimSpace(agent) == "" {
agent = "general"
}
res, err := sp.SpawnAgent("agent.sub."+agent, req.Title, req.Task, "")
if err != nil {
return "", nil, err
}
text := res.Text
if text == "" {
text = res.Summary
}
return text, nil, nil
case "artifact.register":
name := req.Name
if name == "" {
name = "script.md"
}
art, err := r.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 := r.rc.CreateCard(core.ActivityCard{Title: title, Kind: kind})
if err != nil {
return "", nil, err
}
return string(created.ID), nil, nil
case "step":
r.applyScriptPlan(req.StepID, req.Title, req.Status)
return req.Status, nil, nil
default:
return "", nil, fmt.Errorf("未知回调 %s", req.Op)
}
}
// applyScriptPlan 脚本回调 step:更新右栏计划步态(legacy 同款)。
func (r *hostRunner) applyScriptPlan(id, title, status string) {
switch status {
case "pending", "in_progress", "done", "skipped":
default:
status = "done"
}
snap := core.PlanSnapshot{}
if src, ok := r.rc.(PlanGetter); 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})
}
_ = r.rc.UpdatePlan(snap)
}
// doCritic 步骤验收(legacy judgeStep 同款提示词与 fail-closed 判定):
// verdict 字段写 pass/fail 驱动 Loop 再武装;不通过记 drops 与原因。
func (r *hostRunner) doCritic(ctx context.Context, step squad.IntentStep) (string, map[string]string, error) {
target := step.Slots["target_step"]
field := step.Slots["verdict_field"]
output := r.readStepOutput(step.WorkDir, target)
verdict := r.judge(ctx, step.Slots["goal"], step.Slots["criteria"], output)
fields := map[string]string{}
if verdict.Passed {
fields[field] = "pass"
r.recordVerdict(field, target, "pass", "")
return "验收通过。" + verdict.Reason, fields, nil
}
why := verdict.Reason
if verdict.NextAction != "" {
why += "。下一步:" + verdict.NextAction
}
fields[field] = "fail"
fields["reason_"+target] = truncate(why, outFieldCap)
fails := r.recordVerdict(field, target, "fail", why)
// 评审预算耗尽仍不通过 = 步骤失败(legacy runStepWithCritic 同款语义;
// Loop 触顶只停循环不失败,须由宿主判定 fail-closed)。
if maxAtt := parseIntSlot(step.Slots["max_attempts"], 1); fails >= maxAtt {
return "", nil, fmt.Errorf("步骤 %s 未通过验收:%s", target, why)
}
return "验收不通过:" + why, fields, nil
}
func parseIntSlot(s string, def int) int {
n := 0
if strings.TrimSpace(s) == "" {
return def
}
for _, c := range s {
if c < '0' || c > '9' {
return def
}
n = n*10 + int(c-'0')
}
if n <= 0 {
return def
}
return n
}
// recordVerdict 记录判定;返回该字段的累计不通过次数。
func (r *hostRunner) recordVerdict(field, target, verdict, why string) int {
r.mu.Lock()
defer r.mu.Unlock()
if r.verdicts == nil {
r.verdicts = map[string]string{}
r.lastWhy = map[string]string{}
r.fails = map[string]int{}
}
r.verdicts[field] = verdict
if verdict != "pass" {
r.lastWhy[target] = why
r.drops = append(r.drops, target+":"+why)
r.fails[field]++
return r.fails[field]
}
return r.fails[field]
}
// judge 验收判定(legacy judgeStep 移植;JudgeStream 缺席 = 验证器不可用,
// 无法解析 = 不通过,均 fail-closed)。
func (r *hostRunner) judge(ctx context.Context, goal, criteria, output string) gates.Verdict {
stream := r.deps.JudgeStream
if stream == nil {
return gates.Verdict{Passed: false, Reason: "验证器不可用", NextAction: "检查轻量模型配置"}
}
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(ctx, "verify", []agentllm.Message{{Role: agentllm.RoleUser, Content: prompt}}, nil, agentllm.StreamOptions{}) {
switch ev.Kind {
case agentllm.KindTextDelta:
b.WriteString(ev.Text)
case agentllm.KindError:
return gates.Verdict{Passed: false, Reason: "验证器不可用", NextAction: "稍后再试"}
case agentllm.KindMessageEnd:
if b.Len() == 0 {
b.WriteString(ev.Message.Content)
}
}
}
verdict, err := gates.ParseVerdict(b.String())
if err != nil {
return gates.Verdict{Passed: false, Reason: "无法解析", NextAction: "按验收标准重做"}
}
return verdict
}
// readStepOutput 读取某步的主产出(相对本步子工作区回到运行根)。
func (r *hostRunner) readStepOutput(workDir, stepID string) string {
sub, ok := r.subOf[stepID]
if !ok {
return ""
}
wsDir := filepath.Dir(filepath.Dir(workDir))
data, err := os.ReadFile(filepath.Join(wsDir, filepath.FromSlash(sub), "output.md"))
if err != nil {
return ""
}
return string(data)
}
// renderRefs 运行期解析残留的 {{steps.X.output}} 占位符(宿主步实参/正文;
// workDir 为本步子工作区,经其上两级回到运行根再定位目标步产出)。
func (r *hostRunner) renderRefs(s, workDir string) string {
if !strings.Contains(s, "{{steps.") {
return s
}
var b strings.Builder
for {
i := strings.Index(s, "{{steps.")
if i < 0 {
b.WriteString(s)
return b.String()
}
j := strings.Index(s[i:], "}}")
if j < 0 {
b.WriteString(s)
return b.String()
}
token := s[i : i+j+2]
id := strings.TrimSuffix(strings.TrimPrefix(token, "{{steps."), ".output}}")
b.WriteString(s[:i])
b.WriteString(r.readStepOutput(workDir, id))
s = s[i+j+2:]
}
}
func (r *hostRunner) track(art *core.ArtifactRef) {
r.mu.Lock()
r.arts = append(r.arts, art)
r.mu.Unlock()
}
func (r *hostRunner) trackAll(arts []*core.ArtifactRef) {
if len(arts) == 0 {
return
}
r.mu.Lock()
r.arts = append(r.arts, arts...)
r.mu.Unlock()
}
func (r *hostRunner) snapshot() (arts []*core.ArtifactRef, drops []string, verdicts map[string]string, why map[string]string) {
r.mu.Lock()
defer r.mu.Unlock()
arts = append([]*core.ArtifactRef(nil), r.arts...)
drops = append([]string(nil), r.drops...)
verdicts = map[string]string{}
for k, v := range r.verdicts {
verdicts[k] = v
}
why = map[string]string{}
for k, v := range r.lastWhy {
why[k] = v
}
return
}
func truncate(s string, n int) string {
if len(s) <= n {
return s
}
return s[:n] + "…(截断)"
}
+364
View File
@@ -0,0 +1,364 @@
package wfcompile
import (
"context"
"fmt"
"path/filepath"
"strings"
"time"
"agent-squad/squad"
"agent-system-backend/internal/workflow"
core "agent-system-core"
)
// run.go — 编译图执行驱动(Q3 WP1):槽位补齐 → 编译 → squad 引擎装配 →
// 步进循环(挂起 → HITL 卡片 → Respond;执行态 → Resume)→ critic 终判 →
// legacy 同款 outcome 汇总。
// Outcome 是一次编译图执行的结果(legacy workflowOutcome 同款形态)。
type Outcome struct {
Summary string
Parts []core.Part
Artifacts []*core.ArtifactRef
}
// RunOptions 编译图执行口径。
type RunOptions struct {
// DynamicLimits 模型生成上限(≤8 步、禁 script)。
DynamicLimits bool
// RequireConfirm 执行前强制计划确认(模型生成工作流与模型脚本必须确认;
// 注册的声明式工作流免确认)。
RequireConfirm bool
// ModelScript 脚本步为模型提供(执行前单独 HITL 确认)。
ModelScript bool
}
// RunCompiled 编译并执行一个工作流。vars.Input 提供 {{input}} 实值;
// launch.Missing 的槽位在执行前经 HITL 收齐(legacy run() 同款)。
func RunCompiled(rc Host, deps Deps, launch workflow.Launch, vars workflow.Vars, opts RunOptions) (Outcome, error) {
def := launch.Def
if vars.Slots == nil {
vars.Slots = map[string]string{}
}
if vars.Steps == nil {
vars.Steps = map[string]string{}
}
for k, v := range launch.Slots {
vars.Slots[k] = v
}
// 缺失槽位补齐(legacy 同款:逐个 HITL input 卡片)。
for _, missing := range launch.Missing {
if rc.Context().Err() != nil {
return Outcome{}, rc.Context().Err()
}
resp, err := rc.AskHITL(core.HITLCard{
Question: missing.Question, Kind: core.HITLKindInput, AllowFreeInput: true,
})
if err != nil {
return Outcome{}, err
}
answer := strings.TrimSpace(resp.FreeText)
if answer == "" {
answer = resp.ChoiceID
}
vars.Slots[missing.Key] = answer
}
known := map[string]bool{}
for _, tool := range rc.Tools() {
known[tool.Name] = true
}
compiled, err := Compile(def, vars, CompileOptions{
DynamicLimits: opts.DynamicLimits,
ModelScript: opts.ModelScript,
}, func(name string) bool { return known[name] })
if err != nil {
return Outcome{}, err
}
registry := squad.NewRoleRegistry()
for _, role := range compiled.Plan.Squad.Agents {
if err := registry.Register(role); err != nil {
return Outcome{}, fmt.Errorf("workflow: 注册角色 %s: %w", role.Role, err)
}
}
backends := deps.Backends
if backends == nil {
backends = squad.NewBackendResolver()
}
runner := &hostRunner{
rc: rc, deps: deps, scripts: launch.Scripts, subOf: compiled.SubOf,
verdicts: map[string]string{}, lastWhy: map[string]string{}, fails: map[string]int{},
}
engine, err := squad.NewEngine(squad.Options{
DataDir: filepath.Join(deps.DataDir, "wf", string(rc.Run().ID)),
Registry: registry,
Planner: &compiledPlanner{plan: compiled.Plan, needsConfirm: opts.RequireConfirm, reason: PlanPreview(def)},
Backends: backends,
DefaultBackend: orDefault(deps.DefaultBackend, "agent-system"),
StepTimeout: -1, // 负数 = 不设单步截止(引擎只在 0 时回落 10 分钟)
HitlPolicy: squad.HitlPolicy{Mode: squad.HitlInteractive},
Checker: passChecker{},
IntentCatalog: compiled.Catalog,
IntentRunner: runner,
Sink: squad.NopSink{},
})
if err != nil {
return Outcome{}, fmt.Errorf("workflow: 引擎装配: %w", err)
}
// 初始计划快照(legacy 同款:执行前 UpdatePlan 全 pending)。
_ = rc.UpdatePlan(planSnapshot(compiled, def, nil))
res, err := engine.Run(rc.Context(), squad.Goal{Text: def.Title}, squad.RunOptions{})
rejected := false
for err == nil && !squad.RunStatus(res.Status).Terminal() {
if isHoldStatus(string(res.Status)) {
req := res.Pending
if req == nil {
if req, err = engine.Pending(res.RunID); err != nil || req == nil {
break
}
}
// 动态工作流的计划确认被拒 → legacy 语义「未确认,未执行」。
decision, aerr := askHuman(rc, req)
if aerr != nil {
return Outcome{}, aerr
}
if req.Kind == squad.HitlConfirmation && decision.Choice != "" &&
decision.Choice != "approve" && decision.Choice != "yes" {
rejected = true
break
}
res, err = engine.Respond(rc.Context(), res.RunID, decision)
} else {
res, err = engine.Resume(rc.Context(), res.RunID)
}
if err == nil && res.Plan != nil {
_ = rc.UpdatePlan(planSnapshot(compiled, def, res.Plan))
}
}
if rejected {
return Outcome{Summary: "未确认,未执行",
Parts: []core.Part{{Type: core.PartText, Text: "未确认,未执行"}}}, nil
}
finalPlan := res.Plan
if finalPlan == nil {
if insp, ierr := engine.Inspect(res.RunID); ierr == nil {
finalPlan = insp.Plan
}
}
_ = rc.UpdatePlan(planSnapshot(compiled, def, finalPlan))
// critic 终判(P4:评审预算耗尽仍不通过 = 工作流失败)。
stopped := ""
arts, drops, verdicts, why := runner.snapshot()
if err == nil && (res.Status == squad.RunCompleted || res.Status == squad.RunDegraded) {
for _, cp := range compiled.Critics {
if verdicts[cp.VerdictField] != "pass" {
stopped = fmt.Sprintf("步骤 %s 未通过验收:%s", cp.StepTitle, orDefault(why[cp.StepID], "验收未通过"))
err = fmt.Errorf("workflow: 步骤 %s 未通过验收", cp.StepID)
break
}
}
}
if err == nil && res.Status != squad.RunCompleted && res.Status != squad.RunDegraded {
// 引擎级失败(步骤失败/取消)→ legacy 的停止语义。
msg := firstNonEmpty(res.Error, string(res.Status))
stopped = msg
err = fmt.Errorf("workflow: %s", msg)
}
statuses := stepStatuses(compiled, def, finalPlan)
out := buildOutcome(def, statuses, arts, drops, stopped)
if err != nil && rc.Context().Err() != nil {
err = rc.Context().Err() // 取消优先(legacy 同款)
}
return out, err
}
// compiledPlanner 一次性规划器:直接返回编译好的计划(确定性,零模型)。
type compiledPlanner struct {
plan *squad.SquadPlan
needsConfirm bool
reason string
}
func (p *compiledPlanner) Name() string { return "compiled" }
func (p *compiledPlanner) Plan(_ context.Context, _ squad.PlanInput) (*squad.PlanOutput, error) {
steps := make([]squad.StepSpec, len(p.plan.Workflow.Steps))
copy(steps, p.plan.Workflow.Steps)
for i := range steps {
steps[i].Status = "pending"
}
plan := &squad.SquadPlan{
Squad: p.plan.Squad,
Template: p.plan.Template,
Workflow: squad.WorkflowPlan{
WorkflowID: p.plan.Workflow.WorkflowID, Title: p.plan.Workflow.Title,
PlanningMode: p.plan.Workflow.PlanningMode, Status: "pending", Steps: steps,
},
}
return &squad.PlanOutput{Plan: plan, NeedsConfirmation: p.needsConfirm, ConfirmReason: p.reason}, nil
}
// passChecker 编译图收口不做交付物质检(legacy 工作流没有收口质检门;
// 质量把关在 critic 层,由 wf.critic + Loop 承担)。
type passChecker struct{}
func (passChecker) Check(_ squad.CheckRequest) *squad.QualityReport {
return &squad.QualityReport{CheckedAt: time.Now(), Score: 100, Passed: true}
}
// planSnapshot 把编译计划(只列 def 步骤,评审步不单列)映射为右栏快照。
func planSnapshot(compiled *Compiled, def core.WorkflowDef, live *squad.WorkflowPlan) core.PlanSnapshot {
statusOf := map[string]string{}
if live != nil {
for _, s := range live.Steps {
statusOf[s.ID] = mapStepStatus(s.Status)
}
}
snap := core.PlanSnapshot{Title: def.Title, Steps: make([]core.PlanStep, len(def.Steps))}
for i, step := range def.Steps {
snap.Steps[i] = core.PlanStep{ID: step.ID, Text: step.Title,
Status: orDefault(statusOf[step.ID], "pending")}
}
return snap
}
func mapStepStatus(s string) string {
switch s {
case "running":
return "in_progress"
case "completed":
return "done"
case "skipped":
return "skipped"
case "failed":
return "failed"
default:
return "pending"
}
}
// stepStatuses 每个 def 步骤的终态(legacy statuses 语义:done/failed/skipped/pending)。
func stepStatuses(compiled *Compiled, def core.WorkflowDef, live *squad.WorkflowPlan) []string {
out := make([]string, len(def.Steps))
if live == nil {
return out
}
byID := map[string]string{}
for _, s := range live.Steps {
byID[s.ID] = s.Status
}
for i, step := range def.Steps {
switch byID[step.ID] {
case "completed":
out[i] = "done"
case "failed":
out[i] = "failed"
case "skipped":
out[i] = "skipped"
default:
out[i] = "pending"
}
}
return out
}
// buildOutcome legacy outcome() 同款汇总文案与部件。
func buildOutcome(def core.WorkflowDef, statuses []string, arts []*core.ArtifactRef, drops []string, stopped string) Outcome {
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")
}
text := strings.TrimRight(b.String(), "\n")
parts := []core.Part{{Type: core.PartText, Text: text}}
for _, art := range arts {
parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: art})
}
return Outcome{Summary: text, Parts: parts, Artifacts: arts}
}
// askHuman:HumanRequest → HITL 卡片 → RespondDecision(squadadapter 同款桥接)。
func askHuman(rc Host, req *squad.HumanRequest) (squad.RespondDecision, error) {
choices := make([]core.HITLChoice, 0, len(req.Options))
for _, o := range req.Options {
choices = append(choices, core.HITLChoice{
ID: o.Value, Label: o.Label,
Description: firstNonEmpty(req.Context, "工作流挂起"),
Impact: fmt.Sprintf("决策类型 %s:%s 将继续/终止本次执行", req.Kind, o.Value),
})
}
if len(choices) == 0 {
choices = append(choices,
core.HITLChoice{ID: "approve", Label: "继续", Description: "按当前计划继续执行", Impact: "工作流继续执行后续步骤", Recommended: true},
core.HITLChoice{ID: "abort", Label: "中止", Description: "终止执行", Impact: "执行取消,已产出保留"},
)
}
resp, err := rc.AskHITL(core.HITLCard{
Question: req.Question,
Kind: core.HITLKindChoice,
Choices: choices,
AllowFreeInput: true,
FreeInputHint: "可直接输入回答或修改指令",
})
if err != nil {
return squad.RespondDecision{RequestID: req.ID, Choice: "abort"}, err
}
d := squad.RespondDecision{RequestID: req.ID}
if resp.ChoiceID != "" {
d.Choice = resp.ChoiceID
} else {
d.Answer = resp.FreeText
}
return d, nil
}
func isHoldStatus(s string) bool {
return s == "waiting_human" || s == "awaiting_confirmation" || s == "awaiting_clarification" || s == "quality_review"
}
func orDefault(v, def string) string {
if strings.TrimSpace(v) != "" {
return v
}
return def
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if strings.TrimSpace(v) != "" {
return v
}
}
return ""
}
+583
View File
@@ -0,0 +1,583 @@
package wfcompile
import (
"context"
"encoding/json"
"fmt"
"os"
"strings"
"sync"
"testing"
agentllm "agent-native/runtime/llm"
"agent-squad/squad"
"agent-system-backend/internal/workflow"
core "agent-system-core"
)
// wfcompile_test.go — Q3 WP1 验收:demo.brief / demo.script 经编译图跑通
// (脚本化后端 + 桩验收流,零模型零网络零真 LLM;脚本步走真实 codeorch 宿主,
// 与 legacy workflow_script_test 同款环境假设)。
// demo 定义与 Dev 仓 plugins/example/workflows/ 逐字节同源(backend 独立 CI
// 无法读 Dev 仓文件,故内联;Dev 级 make test 以真实插件装载兜底)。
// ---- demo 定义(内联,同源 plugins/example/workflows/) ----
const demoBriefJSON = `{
"id": "demo.brief",
"title": "示例简报",
"description": "命中 demo.brief 后:选篇幅、起草、落成工件、发一张结果卡,再请用户确认。",
"intentId": "demo.brief",
"steps": [
{
"id": "length",
"title": "篇幅",
"kind": "hitl",
"hitl": {
"question": "要多长?",
"kind": "choice",
"allowFreeInput": false,
"choices": [
{"id": "short", "label": "短", "description": "一段", "impact": "只写一段"},
{"id": "long", "label": "长", "description": "几段", "impact": "会写得更长"}
]
}
},
{
"id": "draft",
"title": "起草",
"kind": "prompt",
"agent": "research",
"prompt": "写一份关于「{{slots.topic}}」的简报,篇幅:{{steps.length.output}}。",
"critic": {"criteria": "必须包含单独一行「核对标记:P4」。没有这一行就不通过,nextAction 请要求补上这一行。", "maxRetries": 1}
},
{
"id": "save",
"title": "成稿",
"kind": "artifact",
"artifact": "demo-brief.md",
"prompt": "{{steps.draft.output}}"
},
{
"id": "note",
"title": "收件",
"kind": "card",
"card": {"kind": "result", "title": "简报好了", "tone": "ok"}
},
{
"id": "confirm",
"title": "确认",
"kind": "hitl",
"hitl": {
"question": "这份简报可以用吗?",
"kind": "confirm",
"allowFreeInput": false,
"choices": [
{"id": "yes", "label": "可以用", "description": "结束", "impact": "不会再改"},
{"id": "no", "label": "先不用", "description": "结束", "impact": "不会再改"}
]
}
}
]
}`
const demoScriptJSON = `{
"id": "demo.script",
"title": "示例脚本",
"description": "插件自带脚本。需要打开 runtime.workflowScripts。不绑定意图,用 run_workflow 启动。",
"steps": [
{
"id": "run",
"title": "跑脚本",
"kind": "script",
"script": "workflows/scripts/demo.py"
}
]
}`
const demoPyBody = `import agentwf as wf
now = wf.tool("time_now", {})
answer = wf.ask("要记下来吗?", [{"id": "yes", "label": "记下", "impact": "会写入一份备忘"}])
wf.artifact("demo-script.md", "时间 " + now + "\n选择 " + answer)
wf.step("run", "跑脚本", "done")
`
func parseDef(t *testing.T, raw string) core.WorkflowDef {
t.Helper()
var def core.WorkflowDef
if err := json.Unmarshal([]byte(raw), &def); err != nil {
t.Fatal(err)
}
return def
}
// ---- 夹具 ----
// fakeHost 宿主桩(legacy wfFake 同款行为面)。
type fakeHost struct {
ctx context.Context
settings core.EffectiveSettings
sandbox string
toolNames []string
hitl []core.HITLResponse
hitlQ []string
writes []string
cards []core.ActivityCard
tools []core.ToolCallRequest
plans []core.PlanSnapshot
spawnText string
spawnIDs []string
artSeq int
planCurrent core.PlanSnapshot
}
func (f *fakeHost) Context() context.Context {
if f.ctx == nil {
return context.Background()
}
return f.ctx
}
func (f *fakeHost) Run() core.Run { return core.Run{ID: "run_wfcompile"} }
func (f *fakeHost) Settings() core.EffectiveSettings {
if f.settings == nil {
return core.EffectiveSettings{}
}
return f.settings
}
func (f *fakeHost) Emit(core.StreamEventType, any) error { return nil }
func (f *fakeHost) 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 *fakeHost) UpdatePlan(p core.PlanSnapshot) error {
p.Steps = append([]core.PlanStep(nil), p.Steps...)
f.plans = append(f.plans, p)
f.planCurrent = p
return nil
}
func (f *fakeHost) 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 *fakeHost) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
f.writes = append(f.writes, name+"="+string(content))
f.artSeq++
id := core.ID(fmt.Sprintf("art_%012d", f.artSeq))
return core.Artifact{ID: id, Name: name, Kind: kind}, nil
}
func (f *fakeHost) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
f.tools = append(f.tools, call)
if call.Name == "time_now" {
return core.ToolCallResult{Status: "ok", Output: "12:00"}
}
return core.ToolCallResult{Status: "ok", Output: "ok"}
}
func (f *fakeHost) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
card.ID = core.ID(fmt.Sprintf("card_%d", len(f.cards)+1))
f.cards = append(f.cards, card)
return card, nil
}
func (f *fakeHost) ConversationSandbox() string { return f.sandbox }
func (f *fakeHost) Tools() []core.ToolDesc {
var out []core.ToolDesc
for _, name := range f.toolNames {
out = append(out, core.ToolDesc{Name: name})
}
return out
}
func (f *fakeHost) SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) {
f.spawnIDs = append(f.spawnIDs, agentID)
text := f.spawnText
if text == "" {
text = "子代理正文"
}
return core.AgentResultPayload{Text: text, Status: core.RunCompleted}, nil
}
func (f *fakeHost) SpawnSubagent(name, task, hint string) (core.AgentResultPayload, error) {
return f.SpawnAgent("agent.sub.general", name, task, hint)
}
// seqBackend 脚本化步骤后端:按调用序回放文本(末条重复)。
type seqBackend struct {
mu sync.Mutex
calls int
texts []string
prompts []string
}
func (b *seqBackend) Name() string { return "seq" }
func (b *seqBackend) Execute(_ context.Context, req squad.ExecRequest) (*squad.ExecResult, error) {
b.mu.Lock()
idx := b.calls
b.calls++
b.prompts = append(b.prompts, req.Prompt)
b.mu.Unlock()
if idx >= len(b.texts) {
idx = len(b.texts) - 1
}
patches := map[string][]byte{}
for _, t := range req.Targets {
body := []byte(b.texts[idx] + "\n")
patches[t] = body
if err := os.WriteFile(req.WorkDir+"/"+t, body, 0o644); err != nil {
return nil, err
}
}
return &squad.ExecResult{Success: true, Patches: patches}, nil
}
// verdictStream 验收模型桩:按序回放判定 JSON(末条重复)。
func verdictStream(seq ...string) agentllm.StreamFunc {
var mu sync.Mutex
i := 0
return func(_ context.Context, _ string, _ []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event {
mu.Lock()
idx := i
i++
mu.Unlock()
if idx >= len(seq) {
idx = len(seq) - 1
}
ch := make(chan agentllm.Event, 1)
ch <- agentllm.Event{Kind: agentllm.KindTextDelta, Text: seq[idx]}
close(ch)
return ch
}
}
const (
verdictFail = `{"passed":false,"reason":"缺少核对标记行","nextAction":"补上「核对标记:P4」一行"}`
verdictPass = `{"passed":true,"reason":"包含核对标记行","nextAction":""}`
)
func seqResolver(t *testing.T, be squad.Executor) *squad.BackendResolver {
t.Helper()
r := squad.NewBackendResolver()
r.Register(be.Name(), func(squad.RoleSpec) (squad.Executor, error) { return be, nil })
return r
}
func scriptSettings() core.EffectiveSettings {
return core.EffectiveSettings{"runtime": map[string]any{"workflowScripts": true}}
}
// ---- 编译形状 ----
func TestCompileDemoBriefShape(t *testing.T) {
def := parseDef(t, demoBriefJSON)
c, err := Compile(def, workflow.Vars{Slots: map[string]string{"topic": "太空电梯"}}, CompileOptions{}, nil)
if err != nil {
t.Fatal(err)
}
var ids []string
for _, s := range c.Plan.Workflow.Steps {
ids = append(ids, s.ID)
}
want := "length,draft,draft_critic,save,note,confirm"
if strings.Join(ids, ",") != want {
t.Fatalf("步骤 = %v,期望 %s", ids, want)
}
byID := map[string]squad.StepSpec{}
for _, s := range c.Plan.Workflow.Steps {
byID[s.ID] = s
}
// hitl 步 → wf.hitl,choices 完整进槽位。
if byID["length"].IntentID != IntentHitl || byID["length"].IntentSlots["question"] != "要多长?" ||
!strings.Contains(byID["length"].IntentSlots["choices"], "short") {
t.Fatalf("length = %+v", byID["length"])
}
// prompt 步 → 角色步:Brief 渲染槽位、步骤引用编译为文件指引。
draft := byID["draft"]
if draft.Role != "research" || draft.IntentID != "" {
t.Fatalf("draft = %+v", draft)
}
if !strings.Contains(draft.Brief, "太空电梯") || strings.Contains(draft.Brief, "{{slots.") {
t.Fatalf("draft brief 槽位未渲染:%q", draft.Brief)
}
if !strings.Contains(draft.Brief, "〈步骤 length 的产出〉") || strings.Contains(draft.Brief, "{{steps.") {
t.Fatalf("draft brief 步骤引用未编译:%q", draft.Brief)
}
// critic → 评审步 + evaluator-optimizer 循环(预算 = maxRetries+1)。
critic := byID["draft_critic"]
if critic.IntentID != IntentCritic || critic.Loop == nil {
t.Fatalf("draft_critic = %+v", critic)
}
if critic.Loop.MaxIterations != 2 || critic.Loop.UntilField != "verdict_draft" ||
critic.Loop.UntilEquals != "pass" || strings.Join(critic.Loop.Body, ",") != "draft" {
t.Fatalf("loop = %+v", critic.Loop)
}
// 依赖链:length←draft←draft_critic←save←note←confirm。
chain := map[string]string{"draft": "length", "draft_critic": "draft", "save": "draft_critic", "note": "save", "confirm": "note"}
for id, need := range chain {
if strings.Join(byID[id].Needs, ",") != need {
t.Fatalf("%s needs = %v,期望 [%s]", id, byID[id].Needs, need)
}
}
// artifact 步:名字编译期定值;正文里的步骤引用还原为 token,
// 由宿主处理器运行期按文件解析实值。
if byID["save"].IntentSlots["name"] != "demo-brief.md" ||
byID["save"].IntentSlots["content"] != "{{steps.draft.output}}" {
t.Fatalf("save = %+v", byID["save"].IntentSlots)
}
// intent 角色在编制内。
found := false
for _, a := range c.Plan.Squad.Agents {
if a.Role == squad.DefaultIntentRole {
found = true
}
}
if !found {
t.Fatal("编制缺 intent 角色")
}
}
// P4 上限经编译入口强制(validate 同款文案)。
func TestCompileLimitsP4(t *testing.T) {
steps := make([]core.WorkflowStep, 9)
for i := range steps {
steps[i] = core.WorkflowStep{ID: fmt.Sprintf("s%d", i), Title: "步", Kind: "prompt", Prompt: "x"}
}
def9 := core.WorkflowDef{ID: "dyn", Title: "九步", Steps: steps}
if _, err := Compile(def9, workflow.Vars{}, CompileOptions{DynamicLimits: true}, nil); err == nil ||
!strings.Contains(err.Error(), "超出上限 8") {
t.Fatalf("动态 9 步应拒绝:%v", err)
}
// 动态禁 script。
defScript := core.WorkflowDef{ID: "dyn2", Title: "动态脚本", Steps: []core.WorkflowStep{
{ID: "s", Title: "跑", Kind: "script", Script: "a.py"},
}}
if _, err := Compile(defScript, workflow.Vars{}, CompileOptions{DynamicLimits: true}, nil); err == nil ||
!strings.Contains(err.Error(), "不能包含 script") {
t.Fatalf("动态 script 应拒绝:%v", err)
}
// DeniedTools。
defTool := core.WorkflowDef{ID: "dt", Title: "工具", Steps: []core.WorkflowStep{
{ID: "s", Title: "跑", Kind: "tool", Tool: "run_python"},
}}
if _, err := Compile(defTool, workflow.Vars{}, CompileOptions{}, nil); err == nil ||
!strings.Contains(err.Error(), "不能调用 run_python") {
t.Fatalf("DeniedTools 应拒绝:%v", err)
}
// critic 每步 ≤3;整条 ≤6。
defCritic := core.WorkflowDef{ID: "dc", Title: "评审", Steps: []core.WorkflowStep{
{ID: "s", Title: "写", Kind: "prompt", Prompt: "x", Critic: &core.StepCritic{Criteria: "c", MaxRetries: 4}},
}}
if _, err := Compile(defCritic, workflow.Vars{}, CompileOptions{}, nil); err == nil ||
!strings.Contains(err.Error(), "超过 3") {
t.Fatalf("critic>3 应拒绝:%v", err)
}
defRetry := core.WorkflowDef{ID: "dr", Title: "重试", MaxRetries: 7, Steps: []core.WorkflowStep{
{ID: "s", Title: "写", Kind: "prompt", Prompt: "x"},
}}
if _, err := Compile(defRetry, workflow.Vars{}, CompileOptions{}, nil); err == nil ||
!strings.Contains(err.Error(), "超过 6") {
t.Fatalf("maxRetries>6 应拒绝:%v", err)
}
}
// ---- demo.brief 端到端(critic 重试 + 工件 + 卡 + 确认) ----
func TestRunCompiledDemoBrief(t *testing.T) {
def := parseDef(t, demoBriefJSON)
be := &seqBackend{texts: []string{
"初稿正文:关于太空电梯的简报,篇幅短。(这一版故意缺少核对标记行)",
"终稿正文:关于太空电梯的简报,篇幅短。\n核对标记:P4",
}}
f := &fakeHost{
toolNames: []string{"time_now"},
hitl: []core.HITLResponse{
{ChoiceID: "short"}, // length
{ChoiceID: "yes"}, // confirm
},
}
out, err := RunCompiled(f, Deps{
DataDir: t.TempDir(), Backends: seqResolver(t, be), DefaultBackend: "seq",
JudgeStream: verdictStream(verdictFail, verdictPass),
}, workflow.Launch{Def: def}, workflow.Vars{Input: "写简报", Slots: map[string]string{"topic": "太空电梯"}}, RunOptions{})
if err != nil {
t.Fatalf("err = %v, summary = %s", err, out.Summary)
}
// critic 失败一次 → draft 重跑一次(evaluator-optimizer 循环)。
if be.calls != 2 {
t.Fatalf("draft 执行 %d 次,期望 2(critic 重试)", be.calls)
}
if !strings.Contains(out.Summary, "工作流「示例简报」完成。") || !strings.Contains(out.Summary, "- 起草:done") {
t.Fatalf("summary = %s", out.Summary)
}
// 失败判定进 drops(legacy 同款:重试成功也保留丢弃记录)。
if !strings.Contains(out.Summary, "已丢弃") || !strings.Contains(out.Summary, "draft:缺少核对标记行") {
t.Fatalf("drops 缺失:%s", out.Summary)
}
// hitl 序列 = length → confirm(无计划确认:注册声明式工作流免确认)。
if strings.Join(f.hitlQ, "|") != "要多长?|这份简报可以用吗?" {
t.Fatalf("hitlQ = %v", f.hitlQ)
}
// 工件:save 步取 draft 产出(content_ref 文件指引)。
if len(f.writes) != 1 || !strings.HasPrefix(f.writes[0], "demo-brief.md=") ||
!strings.Contains(f.writes[0], "核对标记:P4") {
t.Fatalf("writes = %v", f.writes)
}
if !strings.Contains(out.Summary, "工件:art_") {
t.Fatalf("summary 缺工件行:%s", out.Summary)
}
// 卡片。
if len(f.cards) != 1 || f.cards[0].Title != "简报好了" || f.cards[0].Kind != core.CardKindResult {
t.Fatalf("cards = %+v", f.cards)
}
// draft 的任务书:槽位已渲染 + hitl 答案经黑板 out_length 可见。
if len(be.prompts) < 1 {
t.Fatal("draft 未收到任务书")
}
p0 := be.prompts[0]
if !strings.Contains(p0, "太空电梯") || !strings.Contains(p0, "## 任务指令") {
t.Fatalf("任务书缺指令/槽位:%q", p0)
}
if !strings.Contains(p0, "out_length = short") {
t.Fatalf("任务书黑板缺 out_length:%q", p0)
}
// 重试任务书带失败反馈(黑板 reason_draft)。
if len(be.prompts) == 2 && !strings.Contains(be.prompts[1], "reason_draft") {
t.Fatalf("重试任务书缺验收反馈:%q", be.prompts[1])
}
}
// critic 预算耗尽仍不通过 → 工作流失败(P4 fail-closed)。
func TestRunCompiledCriticExhaustedFails(t *testing.T) {
def := parseDef(t, demoBriefJSON)
be := &seqBackend{texts: []string{"每一版都缺少核对标记行的正文,用于验证评审预算耗尽路径。"}}
f := &fakeHost{
hitl: []core.HITLResponse{{ChoiceID: "short"}}, // 只到 length;draft 失败后中止
}
_, err := RunCompiled(f, Deps{
DataDir: t.TempDir(), Backends: seqResolver(t, be), DefaultBackend: "seq",
JudgeStream: verdictStream(verdictFail),
}, workflow.Launch{Def: def}, workflow.Vars{Slots: map[string]string{"topic": "太空电梯"}}, RunOptions{})
if err == nil || !strings.Contains(err.Error(), "步骤 draft 未通过验收") {
t.Fatalf("应 fail-closed:%v", err)
}
if be.calls != 2 {
t.Fatalf("预算 = maxRetries+1 = 2 次执行,实际 %d", be.calls)
}
}
// ---- demo.script 端到端(插件脚本免确认 + codeorch 宿主回调) ----
func TestRunCompiledDemoScript(t *testing.T) {
def := parseDef(t, demoScriptJSON)
f := &fakeHost{
sandbox: t.TempDir(),
settings: scriptSettings(),
hitl: []core.HITLResponse{{ChoiceID: "yes"}}, // wf.ask
}
out, err := RunCompiled(f, Deps{DataDir: t.TempDir()},
workflow.Launch{Def: def, Scripts: map[string]string{"workflows/scripts/demo.py": demoPyBody}},
workflow.Vars{}, RunOptions{})
if err != nil {
t.Fatalf("err = %v, summary = %s", err, out.Summary)
}
if !strings.Contains(out.Summary, "工作流「示例脚本」完成。") {
t.Fatalf("summary = %s", out.Summary)
}
// 插件脚本免确认:唯一一次 HITL 是脚本内的 wf.ask。
if strings.Join(f.hitlQ, "|") != "要记下来吗?" {
t.Fatalf("hitlQ = %v(插件脚本不得追加确认门)", f.hitlQ)
}
if len(f.writes) != 1 || f.writes[0] != "demo-script.md=时间 12:00\n选择 yes" {
t.Fatalf("writes = %v", f.writes)
}
if len(f.tools) != 1 || f.tools[0].Name != "time_now" {
t.Fatalf("tools = %v", f.tools)
}
// wf.step 回调 → 计划步态 done。
sawDone := false
for _, p := range f.plans {
for _, s := range p.Steps {
if s.ID == "run" && s.Status == "done" {
sawDone = true
}
}
}
if !sawDone {
t.Fatal("wf.step 回调未落计划")
}
if !strings.Contains(out.Summary, "工件:art_") {
t.Fatalf("summary 缺工件行:%s", out.Summary)
}
}
// runtime.workflowScripts 默认关 → 脚本步失败收口(P4 开关语义)。
func TestRunCompiledScriptDisabled(t *testing.T) {
def := parseDef(t, demoScriptJSON)
f := &fakeHost{sandbox: t.TempDir()} // settings 空 = 开关关
out, err := RunCompiled(f, Deps{DataDir: t.TempDir()},
workflow.Launch{Def: def, Scripts: map[string]string{"workflows/scripts/demo.py": demoPyBody}},
workflow.Vars{}, RunOptions{})
if err == nil {
t.Fatal("开关关闭应失败")
}
if !strings.Contains(out.Summary, "停止") || !strings.Contains(out.Summary, "工作流脚本未开启") {
t.Fatalf("summary = %s", out.Summary)
}
}
// 模型脚本:计划确认 + 脚本确认双门;拒绝脚本确认 → 「未确认,未执行」。
func TestRunCompiledModelScriptConfirm(t *testing.T) {
def := core.WorkflowDef{ID: "model.script", Title: "模型脚本", Steps: []core.WorkflowStep{
{ID: "run", Title: "跑", Kind: "script", Script: "m.py"},
}}
body := "import agentwf as wf\nprint(wf.step('run', '跑', 'done'))\n"
sandbox := t.TempDir()
// 计划确认过、脚本确认拒 → 步骤产出「未确认,未执行」,工作流正常收口。
f := &fakeHost{
sandbox: sandbox,
settings: scriptSettings(),
hitl: []core.HITLResponse{
{ChoiceID: "approve"}, // 计划确认(engine 门)
{ChoiceID: "no"}, // 脚本确认
},
}
out, err := RunCompiled(f, Deps{DataDir: t.TempDir()},
workflow.Launch{Def: def, Scripts: map[string]string{"m.py": body}},
workflow.Vars{}, RunOptions{RequireConfirm: true, ModelScript: true})
if err != nil {
t.Fatalf("err = %v", err)
}
if len(f.hitlQ) != 2 || !strings.Contains(f.hitlQ[1], "运行这段脚本?") {
t.Fatalf("hitlQ = %v", f.hitlQ)
}
if !strings.Contains(out.Summary, "- 跑:done") {
t.Fatalf("summary = %s", out.Summary)
}
entries, _ := os.ReadDir(sandbox)
for _, e := range entries {
if e.Name() == ".agent" {
subs, _ := os.ReadDir(sandbox + "/.agent")
if len(subs) > 0 {
t.Fatalf("拒绝确认后不应落脚本文件:%v", subs)
}
}
}
// 计划确认被拒 → legacy 语义「未确认,未执行」。
f2 := &fakeHost{sandbox: t.TempDir(), settings: scriptSettings(),
hitl: []core.HITLResponse{{ChoiceID: "abort"}}}
out2, err2 := RunCompiled(f2, Deps{DataDir: t.TempDir()},
workflow.Launch{Def: def, Scripts: map[string]string{"m.py": body}},
workflow.Vars{}, RunOptions{RequireConfirm: true, ModelScript: true})
if err2 != nil {
t.Fatal(err2)
}
if out2.Summary != "未确认,未执行" {
t.Fatalf("summary = %s", out2.Summary)
}
}