weichen
f1302ed5b3
Why: Allow consumers to download an immutable module revision without nested source checkouts or local path replacements. Evidence: SELF package tests and the prior canonical-path rejection are recorded in the Dev dependency-workspace task card; remote consumption is replayed against the committed candidate. Impact: Change the module identity and its Go imports while preserving API behavior. Existing historical submodule revisions remain buildable. Refs: Versioned dependencies U1/U3; feature/dependency-workspace
296 lines
11 KiB
Go
296 lines
11 KiB
Go
package runtime
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"strings"
|
||
"testing"
|
||
|
||
agentllm "git-ssh.u1680832.nyat.app/weichen/Agent-Native.git/runtime/llm"
|
||
"git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git/internal/workflow"
|
||
core "git-ssh.u1680832.nyat.app/weichen/Agent-System-Core.git"
|
||
)
|
||
|
||
// 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）")
|
||
}
|
||
}
|
||
|
||
func (*wfFake) Surfaces() []core.SurfaceSnapshot { return nil }
|