Why: Goal-sensitive workflow verification persisted model prompts and prior output text in trusted receipt rows, while cross-step model inputs could reuse stale machine output. Evidence: RED receipt privacy test; full Go suite, targeted race, isolated PostgreSQL store/runmgr/storage-migrate tests, vet, and static build passed locally before commit. A known unrelated TempDir cleanup flake failed once and passed on targeted and serial full replay. Impact: Add optional same-provider source recheck, verify every included item and evidence body before reuse, gate cross-step outputs, reject unsettled attempts, remove raw scope columns, scrub isolated candidate v34 SQLite/PG schemas, and reject unsafe backups. Refs: Approved RDM stage6 W6 source-recheck design adjustment r1; W6 independent review.
713 lines
26 KiB
Go
713 lines
26 KiB
Go
package wfcompile
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/json"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
agentllm "agent-native/runtime/llm"
|
||
"agent-squad/squad"
|
||
"agent-system-backend/internal/projectcontext"
|
||
"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])
|
||
}
|
||
}
|
||
|
||
type projectSourceHost struct {
|
||
*fakeHost
|
||
boundID string
|
||
boundHash [32]byte
|
||
verified int
|
||
invalidated int
|
||
}
|
||
|
||
type stepContextProbe struct{ got string }
|
||
|
||
func (p *stepContextProbe) Name() string { return "probe" }
|
||
func (p *stepContextProbe) Execute(ctx context.Context, _ squad.ExecRequest) (*squad.ExecResult, error) {
|
||
p.got = projectcontext.WorkflowStepAttempt(ctx)
|
||
return &squad.ExecResult{Success: true}, nil
|
||
}
|
||
|
||
func TestProjectStepBackendCarriesEngineAttempt(t *testing.T) {
|
||
probe := &stepContextProbe{}
|
||
base := squad.NewBackendResolver()
|
||
base.RegisterInstance("probe", probe)
|
||
backend, _, err := projectStepBackends(base, nil, nil, nil).Resolve(squad.RoleSpec{Role: "writer", Backend: "probe"}, "")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := backend.Execute(context.Background(), squad.ExecRequest{ExecutionID: "engine-run/draft/2"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if probe.got != "engine-run/draft/2" {
|
||
t.Fatalf("model backend lost trusted step attempt: %q", probe.got)
|
||
}
|
||
}
|
||
|
||
func (h *projectSourceHost) Run() core.Run {
|
||
return core.Run{ID: "run_project_wf", ProjectID: "p1", Owner: "alice"}
|
||
}
|
||
func (h *projectSourceHost) BindProjectStepOutput(_ context.Context, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) error {
|
||
if kind != "prompt" || stepAttemptID == "" || startedAt.IsZero() {
|
||
return fmt.Errorf("untrusted step binding: %s %s %s", stepID, kind, startedAt)
|
||
}
|
||
h.boundID, h.boundHash = stepID, sha256.Sum256(output)
|
||
return nil
|
||
}
|
||
func (h *projectSourceHost) VerifyProjectStepOutput(_ context.Context, stepID, stepAttemptID, kind string, output []byte) error {
|
||
h.verified++
|
||
if stepID != h.boundID || stepAttemptID == "" || kind != "prompt" || sha256.Sum256(output) != h.boundHash {
|
||
return fmt.Errorf("step output provenance mismatch: bound=%q/%x got=%q/%x", h.boundID, h.boundHash, stepID, sha256.Sum256(output))
|
||
}
|
||
return nil
|
||
}
|
||
func (h *projectSourceHost) InvalidateProjectStepOutput(_ context.Context, stepID string) error {
|
||
if stepID == h.boundID {
|
||
h.boundID = ""
|
||
h.boundHash = [32]byte{}
|
||
}
|
||
h.invalidated++
|
||
return nil
|
||
}
|
||
|
||
func TestProjectOutputSinkRejectsConcurrentAndRetriedAttribution(t *testing.T) {
|
||
h := &projectSourceHost{fakeHost: &fakeHost{}}
|
||
root := t.TempDir()
|
||
sub := "tasks/01_draft"
|
||
path := filepath.Join(root, "runs", "engine-run", sub, "output.md")
|
||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := os.WriteFile(path, []byte("model body"), 0o600); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink := newProjectOutputSink(h, root, map[string]string{"draft": sub}, map[string]string{"draft": "prompt", "other": "tool"})
|
||
started := time.Now().Add(-time.Second)
|
||
if err := h.BindProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", started, []byte("old attempt")); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink.Emit(squad.Event{Type: squad.EvStepStarted, RunID: "engine-run", StepID: "draft", At: started})
|
||
if err := h.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("old attempt")); err == nil {
|
||
t.Fatal("retry retained previous attempt's binding")
|
||
}
|
||
sink.Emit(squad.Event{Type: squad.EvStepStarted, RunID: "engine-run", StepID: "other", At: started.Add(time.Millisecond)})
|
||
sink.Emit(squad.Event{Type: squad.EvStepCompleted, RunID: "engine-run", StepID: "draft", At: time.Now()})
|
||
if h.boundID != "" || h.invalidated == 0 {
|
||
t.Fatalf("ambiguous retry rebound output: id=%q invalidated=%d", h.boundID, h.invalidated)
|
||
}
|
||
}
|
||
|
||
func TestProjectPromptCriticBindsActualStepOutputBeforeJudge(t *testing.T) {
|
||
f := &projectSourceHost{fakeHost: &fakeHost{hitl: []core.HITLResponse{{ChoiceID: "short"}, {ChoiceID: "yes"}}}}
|
||
be := &seqBackend{texts: []string{"draft text\n核对标记:P4"}}
|
||
_, err := RunCompiled(f, Deps{DataDir: t.TempDir(), Backends: seqResolver(t, be), DefaultBackend: "seq", JudgeStream: verdictStream(verdictPass)},
|
||
workflow.Launch{Def: parseDef(t, demoBriefJSON)}, workflow.Vars{Input: "write", Slots: map[string]string{"topic": "topic"}}, RunOptions{})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if f.boundID != "draft" || f.boundHash != sha256.Sum256([]byte("draft text\n核对标记:P4\n")) || f.verified == 0 {
|
||
t.Fatalf("project workflow output lacked trusted step binding: id=%q verified=%d", f.boundID, f.verified)
|
||
}
|
||
}
|
||
|
||
func TestProjectToolCriticRejectsUnprovenOutputBeforeModel(t *testing.T) {
|
||
f := &projectSourceHost{fakeHost: &fakeHost{toolNames: []string{"time_now"}}}
|
||
def := core.WorkflowDef{ID: "tool-critic", Title: "Tool review", Steps: []core.WorkflowStep{{ID: "get", Title: "Read", Kind: "tool", Tool: "time_now", Critic: &core.StepCritic{Criteria: "check"}}}}
|
||
called := false
|
||
judge := func(ctx context.Context, model string, messages []agentllm.Message, tools []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event {
|
||
called = true
|
||
return verdictStream(verdictPass)(ctx, model, messages, tools, opts)
|
||
}
|
||
_, err := RunCompiled(f, Deps{DataDir: t.TempDir(), JudgeStream: judge}, workflow.Launch{Def: def}, workflow.Vars{Input: "read"}, RunOptions{})
|
||
if err == nil || called || (!strings.Contains(err.Error(), "产物依据不可用") && !strings.Contains(err.Error(), "unproven machine output")) {
|
||
t.Fatalf("unproven tool output reached judge: err=%v called=%v", err, called)
|
||
}
|
||
}
|
||
|
||
func TestProjectModelStepRejectsEarlierToolOutputInTaskBrief(t *testing.T) {
|
||
f := &projectSourceHost{fakeHost: &fakeHost{toolNames: []string{"time_now"}}}
|
||
be := &seqBackend{texts: []string{"should not run"}}
|
||
def := core.WorkflowDef{ID: "tool-then-model", Title: "Tool then model", Steps: []core.WorkflowStep{
|
||
{ID: "get", Title: "Read", Kind: "tool", Tool: "time_now"},
|
||
{ID: "draft", Title: "Draft", Kind: "prompt", Agent: "writer", Prompt: "use {{steps.get.output}}"},
|
||
}}
|
||
_, err := RunCompiled(f, Deps{DataDir: t.TempDir(), Backends: seqResolver(t, be), DefaultBackend: "seq"}, workflow.Launch{Def: def}, workflow.Vars{Input: "write"}, RunOptions{})
|
||
if err == nil || be.calls != 0 || !strings.Contains(err.Error(), "unproven machine output") {
|
||
t.Fatalf("tool output entered project model step: err=%v modelCalls=%d", err, be.calls)
|
||
}
|
||
}
|
||
|
||
// 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)
|
||
}
|
||
}
|