package runtime import ( "context" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "strings" "testing" agentllm "agent-native/runtime/llm" "agent-system-backend/internal/workflow" core "agent-system-core" ) // workflow_compiled_test.go — Q3 WP2 行为一致性(P4):run_workflow 三分支 // 改投编译图后的 runtime 级验证——注册工作流查找与错误面、规划分支确认门 // 与动态上限、模型脚本开关、缺失槽位 HITL 收齐、验收模型路由。 // (编译/宿主语义的细粒度断言在 internal/wfcompile 测试内。) type wfFake struct { ctx context.Context launch workflow.Launch hitl []core.HITLResponse hitlQ []string plans []core.PlanSnapshot tools []core.ToolCallRequest prompts []string writes []string cards []core.ActivityCard assistant []core.Part arts map[string]core.Artifact byID map[string]core.WorkflowDef spawnText string spawnIDs []string toolNames []string settings core.EffectiveSettings sandbox string scripts map[string]string } func (f *wfFake) Context() context.Context { if f.ctx == nil { return context.Background() } return f.ctx } func (f *wfFake) Run() core.Run { return core.Run{ID: "run_wf"} } func (f *wfFake) Profile() core.AgentProfile { return core.AgentProfile{} } func (f *wfFake) Settings() core.EffectiveSettings { if f.settings == nil { return core.EffectiveSettings{} } return f.settings } func (f *wfFake) ConversationSandbox() string { return f.sandbox } func (f *wfFake) Plan() (core.PlanSnapshot, bool) { if len(f.plans) == 0 { return core.PlanSnapshot{}, false } p := f.plans[len(f.plans)-1] p.Steps = append([]core.PlanStep(nil), p.Steps...) return p, true } func (f *wfFake) WorkflowScript(id, path string) (string, bool) { body, ok := f.scripts[workflow.ScriptKey(id, path)] return body, ok } func (f *wfFake) Emit(core.StreamEventType, any) error { return nil } func (f *wfFake) SetAssistantParts(parts []core.Part) error { f.assistant = parts return nil } func (f *wfFake) 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 *wfFake) SpawnSubagent(name, task, hint string) (core.AgentResultPayload, error) { return f.SpawnAgent("agent.sub.general", name, task, hint) } func (f *wfFake) SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) { f.prompts = append(f.prompts, agentID+"|"+task) f.spawnIDs = append(f.spawnIDs, agentID) text := f.spawnText if text == "" { text = "正文 art_0123456789ab" } return core.AgentResultPayload{Text: text, Status: core.RunCompleted}, nil } func (f *wfFake) UpdatePlan(p core.PlanSnapshot) error { p.Steps = append([]core.PlanStep(nil), p.Steps...) f.plans = append(f.plans, p) return nil } func (f *wfFake) UpdateTodos(core.TodoList) error { return nil } func (f *wfFake) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) { f.writes = append(f.writes, name+"="+string(content)) art := core.Artifact{ID: "art_aaaaaaaaaaaa", Name: name, Kind: kind} if f.arts == nil { f.arts = map[string]core.Artifact{} } f.arts[string(art.ID)] = art return art, nil } func (f *wfFake) Retrieve(core.RetrievalQuery) ([]core.RetrievalHit, error) { return nil, nil } func (f *wfFake) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult { f.tools = append(f.tools, call) if call.Name == "make_file" { art := core.Artifact{ID: "art_bbbbbbbbbbbb", Name: "made.md"} if f.arts == nil { f.arts = map[string]core.Artifact{} } f.arts[string(art.ID)] = art return core.ToolCallResult{Status: "ok", Output: "wrote", Artifact: &art} } return core.ToolCallResult{Status: "ok", Output: "12:00"} } func (f *wfFake) History() []core.Message { return nil } func (f *wfFake) Attachments() []core.ParsedDocument { return nil } func (f *wfFake) Tools() []core.ToolDesc { var out []core.ToolDesc for _, name := range f.toolNames { out = append(out, core.ToolDesc{Name: name}) } return out } func (f *wfFake) Subtask() string { return "" } func (f *wfFake) CreateCard(card core.ActivityCard) (core.ActivityCard, error) { card.ID = "card_1" f.cards = append(f.cards, card) return card, nil } func (f *wfFake) SetCardStatus(string, core.CardStatus, string, string) error { return nil } func (f *wfFake) SubmitCard(card core.ActivityCard) (core.ActivityCard, error) { return card, nil } func (f *wfFake) BeginStreamCard(card core.ActivityCard) (core.ActivityCard, error) { return card, nil } func (f *wfFake) AppendStreamCard(string) error { return nil } func (f *wfFake) CommitStreamCard() (core.ActivityCard, error) { return core.ActivityCard{}, nil } func (f *wfFake) Card(string) *core.ActivityCard { return nil } func (f *wfFake) Artifact(id string) (core.Artifact, []byte, bool) { if f.arts == nil { f.arts = map[string]core.Artifact{} } if id == "art_0123456789ab" { return core.Artifact{ID: core.ID(id), Name: "note.md"}, []byte("x"), true } art, ok := f.arts[id] return art, nil, ok } func (f *wfFake) SetContextObservation(core.ContextObservation) {} func (f *wfFake) Workflow() (workflow.Launch, bool) { return f.launch, true } func (f *wfFake) WorkflowByID(id string) (core.WorkflowDef, bool) { def, ok := f.byID[id] return def, ok } // ---- run_workflow 注册分支:查找与错误面(legacy 同款文案) ---- func TestRunWorkflowToolUsesRegisteredDefinition(t *testing.T) { f := &wfFake{hitl: []core.HITLResponse{{ChoiceID: "yes"}}} f.byID = map[string]core.WorkflowDef{ "demo.brief": {ID: "demo.brief", Title: "简报", Steps: []core.WorkflowStep{ {ID: "ask", Title: "确认", Kind: "hitl", Hitl: &core.HITLCardTemplate{Question: "可以吗?", Kind: core.HITLKindConfirm}}, }}, } st := &directRun{rc: f} out := st.runRegisteredWorkflow(`{"workflow":"demo.brief","inputs":{"topic":"核能"}}`) if !strings.Contains(out, "完成") || f.hitlQ[0] != "可以吗?" { t.Fatalf("out %s questions %v", out, f.hitlQ) } if got := st.runRegisteredWorkflow(`{"workflow":"missing"}`); !strings.Contains(got, "未注册") { t.Fatalf("%s", got) } if got := st.runRegisteredWorkflow(`{"goal":"做一件事","workflow":"demo.brief"}`); !strings.Contains(got, "只能") { t.Fatalf("%s", got) } } // ---- 规划分支:确认门(engine confirmation)+ 动态上限 fail-closed ---- func TestPlannedWorkflowNeedsConfirm(t *testing.T) { plan := `{"id":"plan.brief","title":"简报","steps":[{"id":"ask","title":"确认","kind":"hitl","hitl":{"question":"可以吗?","kind":"confirm","allowFreeInput":false}}]}` f := &wfFake{ spawnText: plan, hitl: []core.HITLResponse{{ChoiceID: "approve"}, {ChoiceID: "ok"}}, } st := &directRun{rc: f} out := st.runRegisteredWorkflow(`{"goal":"写一份简报"}`) if f.spawnIDs[0] != "agent.sub.workflow-plan" || !strings.Contains(f.hitlQ[0], "确认") { t.Fatalf("spawn %v questions %v", f.spawnIDs, f.hitlQ) } if !strings.Contains(out, "完成") || f.hitlQ[1] != "可以吗?" { t.Fatalf("out %s questions %v", out, f.hitlQ) } // 计划确认被拒 → legacy 语义「未确认,未执行」。 f2 := &wfFake{spawnText: plan, hitl: []core.HITLResponse{{ChoiceID: "abort"}}} out = (&directRun{rc: f2}).runRegisteredWorkflow(`{"goal":"写一份简报"}`) if out != "未确认,未执行" || len(f2.hitlQ) != 1 { t.Fatalf("out %s questions %v", out, f2.hitlQ) } // 动态上限:9 步 > 8,编译入口拒绝,不进任何 HITL。 var steps []string for i := 0; i < 9; i++ { steps = append(steps, fmt.Sprintf(`{"id":"s%d","title":"步","kind":"hitl"}`, i)) } f3 := &wfFake{spawnText: `{"id":"too.big","title":"太多","steps":[` + strings.Join(steps, ",") + `]}`} out = (&directRun{rc: f3}).runRegisteredWorkflow(`{"goal":"做很多事"}`) if !strings.Contains(out, "超出上限") || len(f3.hitlQ) != 0 { t.Fatalf("out %s questions %v", out, f3.hitlQ) } } // ---- 脚本分支:runtime.workflowScripts 开关(默认关) ---- func TestScriptBranchRespectsSwitch(t *testing.T) { st := &directRun{rc: &wfFake{sandbox: t.TempDir()}} out := st.runScriptWorkflow("import agentwf as wf\nwf.step('run','跑','done')\n") if !strings.Contains(out, "工作流脚本未开启") { t.Fatalf("开关默认关应失败:%s", out) } } // ---- 缺失槽位:执行前 HITL 收齐并参与编译期渲染 ---- func TestCompiledRunnerAsksMissingSlot(t *testing.T) { f := &wfFake{hitl: []core.HITLResponse{{FreeText: "核能"}, {ChoiceID: "short"}}} f.launch = workflow.Launch{ Missing: []workflow.SlotQuestion{{Key: "topic", Question: "主题是什么?"}}, Def: core.WorkflowDef{ID: "slots", Title: "槽位", Steps: []core.WorkflowStep{ {ID: "len", Title: "篇幅", Kind: "hitl", Hitl: &core.HITLCardTemplate{Question: "篇幅?", Kind: core.HITLKindChoice, Choices: []core.HITLChoice{{ID: "short", Label: "短"}}}}, {ID: "save", Title: "存稿", Kind: "artifact", Artifact: "brief.md", Prompt: "主题:{{slots.topic}},篇幅:{{steps.len.output}}"}, }}, } if err := (CompiledWorkflowRunner{}).Execute(f); err != nil { t.Fatal(err) } if f.hitlQ[0] != "主题是什么?" || f.hitlQ[1] != "篇幅?" { t.Fatalf("questions %v", f.hitlQ) } if len(f.writes) != 1 || !strings.HasPrefix(f.writes[0], "brief.md=") || !strings.Contains(f.writes[0], "主题:核能") || !strings.Contains(f.writes[0], "short") { t.Fatalf("writes = %v", f.writes) } text := f.assistant[0].Text if !strings.Contains(text, "工作流「槽位」完成。") { t.Fatalf("summary %s", text) } } // ---- 验收模型路由:轻量模型配置 → JudgeStream;未配置 → nil(fail-closed) ---- func TestJudgeStreamUsesConfiguredModel(t *testing.T) { var got string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var body map[string]any _ = json.NewDecoder(r.Body).Decode(&body) got, _ = body["model"].(string) w.Header().Set("Content-Type", "text/event-stream") _, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"{\\\"passed\\\":true,\\\"reason\\\":\\\"ok\\\",\\\"nextAction\\\":\\\"\\\"}\"}}]}\n\n") _, _ = io.WriteString(w, "data: [DONE]\n\n") })) t.Cleanup(srv.Close) f := &wfFake{settings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{ "baseUrl": srv.URL, "apiKey": "test-key", "model": "MiniMax-test", }}}} stream := judgeStreamFor(f) if stream == nil { t.Fatal("已配置轻量模型时必须给出验收流") } for ev := range stream(context.Background(), "verify", []agentllm.Message{{Role: agentllm.RoleUser, Content: "x"}}, nil, agentllm.StreamOptions{}) { if ev.Kind == agentllm.KindError { t.Fatal(ev.Text) } } if got != "MiniMax-test" { t.Fatalf("model %q", got) } if judgeStreamFor(&wfFake{}) != nil { t.Fatal("未配置模型应为 nil(验证器不可用 fail-closed)") } }