Merge pull request 'Add the WorkflowDef-to-squad compiler with host intent steps (Q3 WP1)' (#99) from feature/q3-wfcompile into main
ci / vet-and-test (push) Successful in 42s
ci / vet-and-test (push) Successful in 42s
Merge pull request 'Add the WorkflowDef-to-squad compiler with host intent steps (Q3 WP1)' (#99) from feature/q3-wfcompile into main
This commit was merged in pull request #99.
This commit is contained in:
Vendored
+1
-1
Submodule deps/agent-squad updated: bcc8d080b0...633e853bfc
@@ -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
|
||||
}
|
||||
@@ -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] + "…(截断)"
|
||||
}
|
||||
@@ -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 ""
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user