From 70afe40b00ee33cb63d5fa26f3a889f7829111b4 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 8 Oct 2026 03:24:30 +0000 Subject: [PATCH] Retire the legacy workflow executor and switch squad steps to agent-native (Q3 WP2) Why: Q3 WP2 retires the SpawnSubagent-compression execution path: all WorkflowDef execution now goes through the wfcompile compiled graph, and squad steps run on the agent-native kernel backend instead of the agent-system step backend. Evidence: runtime-level P4 parity tests (workflow_compiled_test.go): registered branch lookup/errors, planned branch confirmation gate + dynamic 8-step cap fail-closed before any HITL, model script switch default-off, missing-slot HITL collection with compile-time rendering, judge stream routed to the configured lightweight model (nil = fail-closed verifier-unavailable). squadadapter TestOrchestrateRunsOnKernelStub: four-phase skeleton runs end-to-end on the kernel stub (write_file materialization, seeded skills, exec allowlist). server TestSquadOrchestration reworked: kernel steps run in-process (no subagent child runs). Full backend suite green; wfcompile/squadadapter/runmgr green with -race (pre-existing llm_test fake race on main recorded as debt, CI does not gate -race). Impact: DELETED runtime/workflow.go + workflow_script.go (legacy step loop, critic, script host) and their tests; squadadapter stepBackend 'agent-system' + profileForRole deleted; BuildBackends exported and registers only agent-native; Orchestrate DefaultBackend -> agent-native; builtin roles declare agent-native with write_file added to allowlists and writer run_shell -> run_command (kernel tool name); runmgr default workflow runner -> CompiledWorkflowRunner; mock #workflow demo runs compiled (host-only steps). Model script branch keeps the single script-level confirmation (no extra plan confirmation), matching legacy. Refs: Q3 plan WP2 (2026-10-07-agent-stack-q2q3-plans.md); task card .task/feature-agent-stack-q3.md --- internal/runmgr/runmgr.go | 4 +- internal/runtime/llm_direct.go | 35 +- internal/runtime/mock.go | 4 +- internal/runtime/workflow.go | 437 ------------------ internal/runtime/workflow_compiled.go | 170 +++++++ ...exec_test.go => workflow_compiled_test.go} | 242 ++++------ internal/runtime/workflow_script.go | 297 ------------ internal/runtime/workflow_script_test.go | 243 ---------- internal/server/e2e_test.go | 91 +++- internal/squadadapter/adapter.go | 107 +---- internal/squadadapter/adapter_test.go | 64 +-- internal/squadadapter/entry.go | 2 +- .../squadadapter/orchestrate_native_test.go | 99 ++++ 13 files changed, 502 insertions(+), 1293 deletions(-) delete mode 100644 internal/runtime/workflow.go create mode 100644 internal/runtime/workflow_compiled.go rename internal/runtime/{workflow_exec_test.go => workflow_compiled_test.go} (66%) delete mode 100644 internal/runtime/workflow_script.go delete mode 100644 internal/runtime/workflow_script_test.go create mode 100644 internal/squadadapter/orchestrate_native_test.go diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index c9fe329..d8279f0 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -46,7 +46,7 @@ type Manager struct { mock runtime.Runner // 会话脚本回放（演示/测试与 mock 模式） mockSub runtime.Runner // 子代理脚本回放 squad runtime.Runner // runtime.entry=squad 时的对话入口 - wfRunner runtime.Runner // 意图或工具启动的工作流；nil 时用 runtime.WorkflowRunner + wfRunner runtime.Runner // 意图或工具启动的工作流；nil 时用 runtime.CompiledWorkflowRunner workflows *workflow.Registry // run_workflow 按 id 查找已注册定义 gate IntentGate cards *cardsvc.Service // 可选意图闸门（nil = 未启用，行为回退） @@ -920,7 +920,7 @@ func selectConversationRunner(m *Manager, eff core.EffectiveSettings, launch *wo if m.wfRunner != nil { return m.wfRunner } - return runtime.WorkflowRunner{} + return runtime.CompiledWorkflowRunner{} } if eff.GetBool("runtime.mock", false) { return m.mock diff --git a/internal/runtime/llm_direct.go b/internal/runtime/llm_direct.go index ffcc6a2..6bd4a8b 100644 --- a/internal/runtime/llm_direct.go +++ b/internal/runtime/llm_direct.go @@ -18,6 +18,7 @@ import ( agenttools "agent-native/runtime/tools" "agent-system-backend/internal/ctxasm" "agent-system-backend/internal/squadadapter" + "agent-system-backend/internal/wfcompile" core "agent-system-core" ) @@ -628,7 +629,9 @@ func (st *directRun) runRegisteredWorkflow(args string) string { if !ok { return "未注册的工作流 " + a.Workflow } - res, err := (WorkflowRunner{}).run(st.rc, registeredWorkflowLaunch(st.rc, def, a.Inputs)) + // Q3 WP2：注册工作流经编译图执行（P4 上限在编译入口强制）。 + res, err := (CompiledWorkflowRunner{}).runLaunch(st.rc, + registeredWorkflowLaunch(st.rc, def, a.Inputs), wfcompile.RunOptions{}) st.pinSubagentArtifacts(res.Summary) if err != nil { if res.Summary != "" { @@ -640,31 +643,25 @@ func (st *directRun) runRegisteredWorkflow(args string) string { } func (st *directRun) runScriptWorkflow(body string) string { - text, arts, err := runModelScript(st.rc, WorkflowRunner{}, body) - var ids []string - for _, art := range arts { - if art != nil { - ids = append(ids, string(art.ArtifactID)) - } - } - if len(ids) > 0 { - if text != "" { - text += "\n" - } - text += "工件：" + strings.Join(ids, "、") - } - st.pinSubagentArtifacts(text) + // Q3 WP2：模型脚本包成单步工作流走编译图——runtime.workflowScripts + // 开关与执行前确认由 wfcompile 宿主处理器持有（P4/legacy 同款：模型脚本 + // 只有脚本级确认门，不叠加计划确认）。 + res, err := (CompiledWorkflowRunner{}).runLaunch(st.rc, modelScriptLaunch(body), + wfcompile.RunOptions{ModelScript: true}) + st.pinSubagentArtifacts(res.Summary) if err != nil { - if text != "" { - return text + "\n" + err.Error() + if res.Summary != "" { + return res.Summary + "\n" + err.Error() } return err.Error() } - return text + return res.Summary } func (st *directRun) runPlannedWorkflow(goal string) string { - res, err := planWorkflow(st.rc, goal, WorkflowRunner{}) + // Q3 WP2：规划分支——规划子代理产出 WorkflowDef，动态上限编译， + // 执行前强制计划确认（legacy planWorkflow 同款流）。 + res, err := planCompiledWorkflow(st.rc, goal, CompiledWorkflowRunner{}) st.pinSubagentArtifacts(res.Summary) if err != nil { if res.Summary != "" { diff --git a/internal/runtime/mock.go b/internal/runtime/mock.go index 9e4f066..c98632a 100644 --- a/internal/runtime/mock.go +++ b/internal/runtime/mock.go @@ -7,6 +7,7 @@ import ( "agent-system-backend/internal/docgen" "agent-system-backend/internal/squadadapter" + "agent-system-backend/internal/wfcompile" "agent-system-backend/internal/workflow" core "agent-system-core" ) @@ -214,7 +215,8 @@ func (m *MockRunner) Execute(rc RunCtx) error { } case strings.Contains(user, "#workflow"): - res, err := (WorkflowRunner{}).run(rc, workflow.Launch{Def: mockWorkflow()}) + // Q3 WP2：演示工作流同走编译图（纯宿主步，不依赖模型后端）。 + res, err := (CompiledWorkflowRunner{}).runLaunch(rc, workflow.Launch{Def: mockWorkflow()}, wfcompile.RunOptions{}) if err != nil { if res.Summary != "" { addText(res.Summary) diff --git a/internal/runtime/workflow.go b/internal/runtime/workflow.go deleted file mode 100644 index a3cae49..0000000 --- a/internal/runtime/workflow.go +++ /dev/null @@ -1,437 +0,0 @@ -package runtime - -import ( - "encoding/json" - "fmt" - "strings" - "time" - - "agent-native/runtime/gates" - agentllm "agent-native/runtime/llm" - "agent-system-backend/internal/workflow" - core "agent-system-core" -) - -const workflowPlannerID = "agent.sub.workflow-plan" - -// WorkflowRunner 执行启动时拷贝下来的工作流定义。 -type WorkflowRunner struct { - // JudgeStream 注入验收模型。nil 时按轻量模型构造客户端。 - JudgeStream agentllm.StreamFunc - // ScriptTimeout 覆盖脚本墙钟超时。0 表示 600 秒，超过 1800 秒会被截断。 - ScriptTimeout time.Duration -} - -func (WorkflowRunner) Name() string { return "workflow" } - -type workflowSource interface { - Workflow() (workflow.Launch, bool) -} - -type workflowOutcome struct { - Summary string - Parts []core.Part - Artifacts []*core.ArtifactRef -} - -// planWorkflow 让规划子代理写出步骤，校验通过后必须经用户确认才执行。 -func planWorkflow(rc RunCtx, goal string, runner WorkflowRunner) (workflowOutcome, error) { - sp, ok := rc.(interface { - SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) - }) - if !ok { - return workflowOutcome{}, fmt.Errorf("workflow: 当前运行不能规划") - } - res, err := sp.SpawnAgent(workflowPlannerID, "规划", "把下面的目标规划成 WorkflowDef。只输出 JSON，不要完成目标本身。\n目标："+goal, "") - if err != nil { - return workflowOutcome{}, err - } - text := res.Text - if text == "" { - text = res.Summary - } - def, err := parseWorkflowDef(text) - if err != nil { - return workflowOutcome{}, err - } - known := map[string]bool{} - for _, tool := range rc.Tools() { - known[tool.Name] = true - } - if err := workflow.Validate(def, workflow.DynamicLimits(), func(name string) bool { return known[name] }); err != nil { - return workflowOutcome{}, err - } - resp, err := rc.AskHITL(core.HITLCard{ - Question: planPreview(def), Kind: core.HITLKindConfirm, AllowFreeInput: false, - Choices: []core.HITLChoice{ - {ID: "yes", Label: "执行", Description: "按这些步骤做", Impact: "会调用工具并可能写入工件"}, - {ID: "no", Label: "取消", Description: "不执行", Impact: "不会改动任何东西"}, - }, - }) - if err != nil { - return workflowOutcome{}, err - } - if resp.ChoiceID != "yes" { - return workflowOutcome{Summary: "未确认，未执行"}, nil - } - return runner.run(rc, workflow.Launch{Def: def}) -} - -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 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") -} - -func (w WorkflowRunner) Execute(rc RunCtx) error { - src, ok := rc.(workflowSource) - if !ok { - return fmt.Errorf("workflow: 当前运行没有工作流定义") - } - launch, ok := src.Workflow() - if !ok { - return fmt.Errorf("workflow: 当前运行没有工作流定义") - } - res, err := w.run(rc, launch) - if len(res.Parts) > 0 { - _ = rc.SetAssistantParts(res.Parts) - } - return err -} - -func (w WorkflowRunner) run(rc RunCtx, launch workflow.Launch) (workflowOutcome, error) { - def := launch.Def - vars := workflow.Vars{ - Input: currentUserText(rc), - Slots: map[string]string{}, - Steps: map[string]string{}, - } - for k, v := range launch.Slots { - vars.Slots[k] = v - } - for _, missing := range launch.Missing { - if rc.Context().Err() != nil { - return workflowOutcome{}, rc.Context().Err() - } - resp, err := rc.AskHITL(core.HITLCard{ - Question: missing.Question, Kind: core.HITLKindInput, AllowFreeInput: true, - }) - if err != nil { - return workflowOutcome{}, err - } - answer := strings.TrimSpace(resp.FreeText) - if answer == "" { - answer = resp.ChoiceID - } - vars.Slots[missing.Key] = answer - } - - plan := core.PlanSnapshot{Title: def.Title, Steps: make([]core.PlanStep, len(def.Steps))} - for i, step := range def.Steps { - plan.Steps[i] = core.PlanStep{ID: step.ID, Text: step.Title, Status: "pending"} - } - if err := rc.UpdatePlan(plan); err != nil { - return workflowOutcome{}, err - } - - wholeLeft := def.MaxRetries - if wholeLeft <= 0 || wholeLeft > workflow.MaxWorkflowRetries { - wholeLeft = workflow.MaxWorkflowRetries - } - var drops []string - var arts []*core.ArtifactRef - var prev string - statuses := make([]string, len(def.Steps)) - for i := range def.Steps { - if rc.Context().Err() != nil { - return outcome(def, statuses, arts, drops, ""), rc.Context().Err() - } - step := def.Steps[i] - plan.Steps[i].Status = "in_progress" - if err := rc.UpdatePlan(plan); err != nil { - return workflowOutcome{}, err - } - rendered, err := workflow.RenderStep(step, vars) - if err != nil { - return outcome(def, statuses, arts, drops, err.Error()), err - } - output, stepArts, reason, err := w.runStepWithCritic(rc, def, rendered, vars, prev, launch.Scripts, &wholeLeft, &drops) - arts = append(arts, stepArts...) - if err != nil { - statuses[i] = "failed" - plan.Steps[i].Status = "skipped" - _ = rc.UpdatePlan(plan) - res := outcome(def, statuses, arts, drops, reason) - return res, err - } - vars.Steps[step.ID] = output - prev = output - statuses[i] = "done" - plan.Steps[i].Status = "done" - if err := rc.UpdatePlan(plan); err != nil { - return workflowOutcome{}, err - } - } - res := outcome(def, statuses, arts, drops, "") - return res, nil -} - -func (w WorkflowRunner) runStepWithCritic(rc RunCtx, def core.WorkflowDef, step core.WorkflowStep, vars workflow.Vars, prev string, scripts map[string]string, wholeLeft *int, drops *[]string) (string, []*core.ArtifactRef, string, error) { - stepRetries := 1 - if step.Critic != nil && step.Critic.MaxRetries > 0 { - stepRetries = step.Critic.MaxRetries - } - if stepRetries > workflow.MaxStepRetries { - stepRetries = workflow.MaxStepRetries - } - used := 0 - note := "" - var arts []*core.ArtifactRef - for { - current := step - if note != "" && current.Prompt != "" { - current.Prompt = current.Prompt + "\n\n未通过验收：" + note - } else if note != "" { - current.Prompt = "未通过验收：" + note - } - output, got, err := w.runWorkflowStep(rc, current, vars, prev, scripts) - arts = append(arts, got...) - if err != nil { - return output, arts, err.Error(), err - } - if step.Critic == nil || strings.TrimSpace(step.Critic.Criteria) == "" { - return output, arts, "", nil - } - verdict := w.judgeStep(rc, def.Title, step.Critic.Criteria, output) - if verdict.Passed { - return output, arts, "", nil - } - why := verdict.Reason - if verdict.NextAction != "" { - why += "。下一步：" + verdict.NextAction - } - *drops = append(*drops, step.ID+"："+why) - if used >= stepRetries || *wholeLeft <= 0 { - return output, arts, "步骤 " + step.Title + " 未通过验收：" + why, fmt.Errorf("workflow: 步骤 %s 未通过验收", step.ID) - } - used++ - *wholeLeft-- - note = why - } -} - -func (w WorkflowRunner) runWorkflowStep(rc RunCtx, step core.WorkflowStep, vars workflow.Vars, prev string, scripts map[string]string) (string, []*core.ArtifactRef, error) { - switch step.Kind { - case "prompt": - return runPromptStep(rc, step) - case "tool": - res := rc.ExecuteTool(core.ToolCallRequest{Name: step.Tool, Arguments: step.Args}) - if res.Status == "error" { - return res.Output, nil, fmt.Errorf("workflow: 工具 %s 失败：%s", step.Tool, 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": - card := core.HITLCard{Question: step.Title, Kind: core.HITLKindInput, AllowFreeInput: true} - if step.Hitl != nil { - card.Question = step.Hitl.Question - card.Kind = step.Hitl.Kind - card.Choices = step.Hitl.Choices - card.AllowFreeInput = step.Hitl.AllowFreeInput - card.FreeInputHint = step.Hitl.FreeInputHint - } - resp, err := 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 "artifact": - content := step.Prompt - if strings.TrimSpace(content) == "" { - content = prev - } - name := step.Artifact - if name == "" { - name = step.ID + ".md" - } - art, err := rc.WriteArtifact(name, "document", "text/markdown", []byte(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": - card := core.ActivityCard{Title: step.Title, Kind: core.CardKindResult} - if step.Card != nil { - card.Kind = step.Card.Kind - card.Title = step.Card.Title - card.Tone = step.Card.Tone - card.Body = step.Card.Body - card.Actions = step.Card.Actions - } - created, err := rc.CreateCard(card) - if err != nil { - return "", nil, err - } - return string(created.ID), nil, nil - case "script": - return executeScriptStep(rc, w, step, scripts) - default: - return "", nil, fmt.Errorf("workflow: 未知步骤类型 %s", step.Kind) - } -} - -func runPromptStep(rc RunCtx, step core.WorkflowStep) (string, []*core.ArtifactRef, error) { - agent := step.Agent - if strings.TrimSpace(agent) == "" { - agent = "general" - } - var res core.AgentResultPayload - var err error - if sp, ok := rc.(interface { - SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) - }); ok { - res, err = sp.SpawnAgent(subagentProfileID(agent), step.Title, step.Prompt, "") - } else { - res, err = rc.SpawnSubagent(step.Title, step.Prompt, "") - } - if err != nil { - return "", nil, err - } - text := res.Text - if text == "" { - text = res.Summary - } - return text, artifactsIn(rc, text), nil -} - -func artifactsIn(rc RunCtx, text string) []*core.ArtifactRef { - var out []*core.ArtifactRef - seen := map[string]bool{} - for _, id := range gates.ArtifactIDPattern.FindAllString(text, -1) { - if seen[id] { - continue - } - art, _, ok := rc.Artifact(id) - if !ok { - continue - } - seen[id] = true - out = append(out, &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}) - } - return out -} - -func (w WorkflowRunner) judgeStep(rc RunCtx, goal, criteria, output string) gates.Verdict { - stream := w.JudgeStream - model := "verify" - if stream == nil { - route := ResolveLightweightModel(rc.Profile(), rc.Settings()) - if route.Model == "" || route.Base == "" { - return gates.Verdict{Passed: false, Reason: "验证器不可用", NextAction: "检查轻量模型配置"} - } - model = route.Model - client := openLLMClient(route.Base, route.Key) - stream = client.Stream - } - 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(rc.Context(), model, []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 -} - -func outcome(def core.WorkflowDef, statuses []string, arts []*core.ArtifactRef, drops []string, stopped string) workflowOutcome { - 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") - } - parts := []core.Part{{Type: core.PartText, Text: strings.TrimRight(b.String(), "\n")}} - for _, art := range arts { - parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: art}) - } - return workflowOutcome{Summary: parts[0].Text, Parts: parts, Artifacts: arts} -} - -// stripFenceBlock 剥模型输出的 ``` 围栏（工作流定义解析用；与门禁域无关，本地持有）。 -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), "```")) -} diff --git a/internal/runtime/workflow_compiled.go b/internal/runtime/workflow_compiled.go new file mode 100644 index 0000000..7af38f2 --- /dev/null +++ b/internal/runtime/workflow_compiled.go @@ -0,0 +1,170 @@ +package runtime + +import ( + "context" + "fmt" + "strings" + "time" + + agentllm "agent-native/runtime/llm" + "agent-system-backend/internal/squadadapter" + "agent-system-backend/internal/wfcompile" + "agent-system-backend/internal/workflow" + core "agent-system-core" +) + +// workflow_compiled.go — 工作流执行器（Q3 WP2）：WorkflowDef 一律经 +// internal/wfcompile 编译为 squad 图执行（legacy 逐步循环 + SpawnSubagent +// 压缩路径已退役）。P4 语义由编译入口与宿主处理器持有：步数上限、动态 +// 工作流禁 script 且必须确认、模型脚本必须确认、runtime.workflowScripts +// 默认关、critic 每步 ≤3 整条 ≤6、脚本回调限额、DeniedTools。 + +const workflowPlannerID = "agent.sub.workflow-plan" + +// workflowSource 由 runmgr 的 runCtx 实现：run 启动时拷贝的工作流定义。 +type workflowSource interface { + Workflow() (workflow.Launch, bool) +} + +// CompiledWorkflowRunner 以编译图执行启动时绑定的工作流（意图/演示入口）。 +type CompiledWorkflowRunner struct { + // JudgeStream 注入验收模型；nil 时按轻量模型配置构造，仍缺则验证器不可用。 + JudgeStream agentllm.StreamFunc + // ScriptTimeout 覆盖脚本墙钟超时（0 = 600s；1800s 上限由宿主持有）。 + ScriptTimeout time.Duration +} + +func (CompiledWorkflowRunner) Name() string { return "workflow" } + +func (w CompiledWorkflowRunner) Execute(rc RunCtx) error { + src, ok := rc.(workflowSource) + if !ok { + return fmt.Errorf("workflow: 当前运行没有工作流定义") + } + launch, ok := src.Workflow() + if !ok { + return fmt.Errorf("workflow: 当前运行没有工作流定义") + } + out, err := w.runLaunch(rc, launch, wfcompile.RunOptions{}) + if len(out.Parts) > 0 { + _ = rc.SetAssistantParts(out.Parts) + } + return err +} + +// runLaunch 装配编译图执行依赖并跑一次工作流（三个 run_workflow 分支与 +// 意图入口共用）。 +func (w CompiledWorkflowRunner) runLaunch(rc RunCtx, launch workflow.Launch, opts wfcompile.RunOptions) (wfcompile.Outcome, error) { + deps := wfcompile.Deps{ + DataDir: rc.Settings().GetString("storage.dataDir", "./data"), + Backends: squadadapter.BuildBackends(rc), + DefaultBackend: "agent-native", + JudgeStream: w.JudgeStream, + ScriptTimeout: w.ScriptTimeout, + } + if deps.JudgeStream == nil { + deps.JudgeStream = judgeStreamFor(rc) + } + return wfcompile.RunCompiled(workflowHost{rc: rc}, deps, launch, workflow.Vars{ + Input: currentUserText(rc), + Slots: map[string]string{}, + Steps: map[string]string{}, + }, opts) +} + +// judgeStreamFor 按轻量模型配置构造验收流（legacy judgeStep 回退路径同款： +// 模型名取路由结果；未配置 = nil，wfcompile 按「验证器不可用」fail-closed）。 +func judgeStreamFor(rc RunCtx) agentllm.StreamFunc { + route := ResolveLightweightModel(rc.Profile(), rc.Settings()) + if route.Model == "" || route.Base == "" { + return nil + } + client := openLLMClient(route.Base, route.Key) + model := route.Model + return func(ctx context.Context, _ string, msgs []agentllm.Message, tools []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event { + return client.Stream(ctx, model, msgs, tools, opts) + } +} + +// planCompiledWorkflow 规划分支（run_workflow goal=）：规划子代理产出 +// WorkflowDef → 动态上限编译 → 执行前强制确认（legacy planWorkflow 同款流）。 +func planCompiledWorkflow(rc RunCtx, goal string, runner CompiledWorkflowRunner) (wfcompile.Outcome, error) { + sp, ok := rc.(interface { + SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) + }) + if !ok { + return wfcompile.Outcome{}, fmt.Errorf("workflow: 当前运行不能规划") + } + res, err := sp.SpawnAgent(workflowPlannerID, "规划", "把下面的目标规划成 WorkflowDef。只输出 JSON，不要完成目标本身。\n目标："+goal, "") + if err != nil { + return wfcompile.Outcome{}, err + } + text := res.Text + if text == "" { + text = res.Summary + } + def, err := wfcompile.ParseWorkflowDef(text) + if err != nil { + return wfcompile.Outcome{}, err + } + return runner.runLaunch(rc, workflow.Launch{Def: def}, wfcompile.RunOptions{ + DynamicLimits: true, RequireConfirm: true, + }) +} + +// registeredWorkflowLaunch 组装注册工作流的启动副本（脚本正文取装载快照， +// legacy 同款：插件后续变更不影响已启动运行）。 +func registeredWorkflowLaunch(rc RunCtx, def core.WorkflowDef, slots map[string]string) workflow.Launch { + launch := workflow.Launch{Def: def, Slots: slots, Scripts: map[string]string{}} + lookup, ok := rc.(interface { + WorkflowScript(id, path string) (string, bool) + }) + if !ok { + return launch + } + for _, step := range def.Steps { + if step.Kind != "script" || strings.TrimSpace(step.Script) == "" { + continue + } + if body, found := lookup.WorkflowScript(def.ID, step.Script); found { + launch.Scripts[step.Script] = body + } + } + return launch +} + +// modelScriptLaunch 把模型提供的脚本正文包成单步工作流（script 分支； +// 确认与开关语义由 wfcompile 宿主处理器持有）。 +func modelScriptLaunch(body string) workflow.Launch { + def := core.WorkflowDef{ID: "model.script", Title: "模型脚本", Steps: []core.WorkflowStep{ + {ID: "run", Title: "跑脚本", Kind: "script", Script: "model.py"}, + }} + return workflow.Launch{Def: def, Scripts: map[string]string{"model.py": body}} +} + +// workflowHost 把 RunCtx 适配到 wfcompile.Host（ConversationSandbox 是 +// 运行侧可选能力，缺失时回退空串由宿主处理器 fail-closed）。 +type workflowHost struct{ rc RunCtx } + +func (h workflowHost) Context() context.Context { return h.rc.Context() } +func (h workflowHost) Run() core.Run { return h.rc.Run() } +func (h workflowHost) Settings() core.EffectiveSettings { return h.rc.Settings() } +func (h workflowHost) Emit(t core.StreamEventType, p any) error { return h.rc.Emit(t, p) } +func (h workflowHost) AskHITL(c core.HITLCard) (core.HITLResponse, error) { return h.rc.AskHITL(c) } +func (h workflowHost) UpdatePlan(p core.PlanSnapshot) error { return h.rc.UpdatePlan(p) } +func (h workflowHost) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) { + return h.rc.WriteArtifact(name, kind, mime, content) +} +func (h workflowHost) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult { + return h.rc.ExecuteTool(call) +} +func (h workflowHost) CreateCard(card core.ActivityCard) (core.ActivityCard, error) { + return h.rc.CreateCard(card) +} +func (h workflowHost) Tools() []core.ToolDesc { return h.rc.Tools() } +func (h workflowHost) ConversationSandbox() string { + if sb, ok := h.rc.(interface{ ConversationSandbox() string }); ok { + return sb.ConversationSandbox() + } + return "" +} diff --git a/internal/runtime/workflow_exec_test.go b/internal/runtime/workflow_compiled_test.go similarity index 66% rename from internal/runtime/workflow_exec_test.go rename to internal/runtime/workflow_compiled_test.go index 8e8d097..7b5ace4 100644 --- a/internal/runtime/workflow_exec_test.go +++ b/internal/runtime/workflow_compiled_test.go @@ -10,10 +10,16 @@ import ( "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 @@ -158,104 +164,9 @@ func (f *wfFake) WorkflowByID(id string) (core.WorkflowDef, bool) { return def, ok } -func TestWorkflowRunsFiveKindsAndAsksMissingSlot(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: "demo.brief", Title: "简报", Steps: []core.WorkflowStep{ - {ID: "len", Title: "篇幅", Kind: "hitl", Hitl: &core.HITLCardTemplate{Question: "篇幅？", Kind: core.HITLKindChoice, Choices: []core.HITLChoice{{ID: "short", Label: "短"}}}}, - {ID: "draft", Title: "起草", Kind: "prompt", Agent: "research", Prompt: "写 {{slots.topic}}"}, - {ID: "now", Title: "时间", Kind: "tool", Tool: "time_now", Args: []byte(`{"q":"{{slots.topic}}"}`)}, - {ID: "save", Title: "存稿", Kind: "artifact", Artifact: "brief.md", Prompt: "{{steps.draft.output}}"}, - {ID: "note", Title: "通知", Kind: "card", Card: &core.ActivityCardTemplate{Kind: core.CardKindResult, Title: "好了"}}, - }}, - } - if err := (WorkflowRunner{}).Execute(f); err != nil { - t.Fatal(err) - } - if f.hitlQ[0] != "主题是什么？" || f.hitlQ[1] != "篇幅？" { - t.Fatalf("questions %v", f.hitlQ) - } - if len(f.prompts) != 1 || !strings.Contains(f.prompts[0], "agent.sub.research|写 核能") { - t.Fatalf("prompt %v", f.prompts) - } - if len(f.tools) != 1 || string(f.tools[0].Arguments) == "" || !strings.Contains(string(f.tools[0].Arguments), "核能") { - t.Fatalf("tool %#v", f.tools) - } - if len(f.writes) != 1 || f.writes[0] != "brief.md=正文 art_0123456789ab" { - t.Fatalf("writes %v", f.writes) - } - if len(f.cards) != 1 || f.cards[0].Title != "好了" { - t.Fatalf("cards %+v", f.cards) - } - var sawProgress, sawDone bool - for _, p := range f.plans { - for _, s := range p.Steps { - if s.ID == "draft" && s.Status == "in_progress" { - sawProgress = true - } - if s.ID == "note" && s.Status == "done" { - sawDone = true - } - } - } - if !sawProgress || !sawDone { - t.Fatal("plan did not move from in_progress to done") - } - text := f.assistant[0].Text - if !strings.Contains(text, "art_0123456789ab") || !strings.Contains(text, "art_aaaaaaaaaaaa") { - t.Fatalf("summary %s", text) - } -} -func TestCriticRetriesThenStops(t *testing.T) { - f := &wfFake{} - f.launch = workflow.Launch{Def: core.WorkflowDef{ - ID: "a", Title: "验收", MaxRetries: 6, - Steps: []core.WorkflowStep{{ - ID: "draft", Title: "起草", Kind: "prompt", Prompt: "写", - Critic: &core.StepCritic{Criteria: "必须点明主题", MaxRetries: 1}, - }}, - }} - runner := WorkflowRunner{JudgeStream: passedStream( - `{"passed":false,"reason":"没点明主题","nextAction":"补上主题"}`, - `{"passed":true,"reason":"可以","nextAction":""}`, - )} - if err := runner.Execute(f); err != nil { - t.Fatal(err) - } - if len(f.prompts) != 2 || !strings.Contains(f.prompts[1], "没点明主题") || !strings.Contains(f.prompts[1], "补上主题") { - t.Fatalf("prompts %v", f.prompts) - } - if !strings.Contains(f.assistant[0].Text, "已丢弃") || !strings.Contains(f.assistant[0].Text, "没点明主题") { - t.Fatalf("summary %s", f.assistant[0].Text) - } - f2 := &wfFake{} - f2.launch = f.launch - bad := WorkflowRunner{JudgeStream: passedStream("不是 json", "不是 json")} - err := bad.Execute(f2) - if err == nil || len(f2.prompts) != 2 { - t.Fatalf("err %v prompts %d", err, len(f2.prompts)) - } - if !strings.Contains(f2.assistant[0].Text, "无法解析") { - t.Fatalf("summary %s", f2.assistant[0].Text) - } - - f3 := &wfFake{} - f3.launch = workflow.Launch{Def: core.WorkflowDef{ - ID: "a", Title: "验收", MaxRetries: 1, - Steps: []core.WorkflowStep{{ - ID: "draft", Title: "起草", Kind: "prompt", Prompt: "写", - Critic: &core.StepCritic{Criteria: "要过", MaxRetries: 3}, - }}, - }} - fail := `{"passed":false,"reason":"不行","nextAction":"重写"}` - stop := WorkflowRunner{JudgeStream: passedStream(fail, fail, fail)} - if err := stop.Execute(f3); err == nil || len(f3.prompts) != 2 { - t.Fatalf("whole cap err %v attempts %d", err, len(f3.prompts)) - } -} +// ---- run_workflow 注册分支：查找与错误面（legacy 同款文案） ---- func TestRunWorkflowToolUsesRegisteredDefinition(t *testing.T) { f := &wfFake{hitl: []core.HITLResponse{{ChoiceID: "yes"}}} @@ -277,7 +188,82 @@ func TestRunWorkflowToolUsesRegisteredDefinition(t *testing.T) { } } -func TestJudgeUsesConfiguredModel(t *testing.T) { +// ---- 规划分支：确认门（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 @@ -288,60 +274,22 @@ func TestJudgeUsesConfiguredModel(t *testing.T) { _, _ = 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", - }}}, - launch: workflow.Launch{Def: core.WorkflowDef{ - ID: "c", Title: "验收", - Steps: []core.WorkflowStep{{ - ID: "draft", Title: "起草", Kind: "prompt", Prompt: "写", - Critic: &core.StepCritic{Criteria: "要过", MaxRetries: 1}, - }}, - }}, + 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("已配置轻量模型时必须给出验收流") } - if err := (WorkflowRunner{}).Execute(f); err != nil { - t.Fatal(err) + 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) } -} - -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: "yes"}, {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) - } - - f2 := &wfFake{spawnText: plan, hitl: []core.HITLResponse{{ChoiceID: "no"}}} - out = (&directRun{rc: f2}).runRegisteredWorkflow(`{"goal":"写一份简报"}`) - if out != "未确认，未执行" || len(f2.hitlQ) != 1 { - t.Fatalf("out %s questions %v", out, f2.hitlQ) - } - - 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) - } - - f4 := &wfFake{spawnText: `{"id":"bad.script","title":"脚本","steps":[{"id":"run","title":"跑","kind":"script","script":"a.py"}]}`} - out = (&directRun{rc: f4}).runRegisteredWorkflow(`{"goal":"跑脚本"}`) - if !strings.Contains(out, "script") || len(f4.hitlQ) != 0 { - t.Fatalf("out %s questions %v", out, f4.hitlQ) + if judgeStreamFor(&wfFake{}) != nil { + t.Fatal("未配置模型应为 nil（验证器不可用 fail-closed）") } } diff --git a/internal/runtime/workflow_script.go b/internal/runtime/workflow_script.go deleted file mode 100644 index c6f9880..0000000 --- a/internal/runtime/workflow_script.go +++ /dev/null @@ -1,297 +0,0 @@ -package runtime - -import ( - "context" - "encoding/json" - "fmt" - "os" - "os/exec" - "path/filepath" - "strings" - "time" - - "agent-native/runtime/codeorch" - "agent-system-backend/internal/tools" - "agent-system-backend/internal/workflow" - core "agent-system-core" -) - -const ( - scriptDefaultTimeout = 600 * time.Second - scriptMaxTimeout = 1800 * time.Second - scriptMaxCalls = 50 - scriptMaxLine = 1 << 20 -) - -func scriptsEnabled(rc RunCtx) bool { - return rc.Settings().GetBool("runtime.workflowScripts", false) -} - -func scriptSandbox(rc RunCtx) (string, error) { - sandbox := strings.TrimSpace(conversationSandbox(rc)) - if sandbox == "" { - return "", fmt.Errorf("workflow: 没有会话沙箱") - } - return sandbox, nil -} - -func executeScriptStep(rc RunCtx, runner WorkflowRunner, step core.WorkflowStep, scripts map[string]string) (string, []*core.ArtifactRef, error) { - body := scripts[step.Script] - if strings.TrimSpace(body) == "" { - var err error - body, err = scriptBody(rc, step.Script) - if err != nil { - return "", nil, err - } - } - sandbox, err := scriptSandbox(rc) - if err != nil { - return "", nil, err - } - path, err := writeScriptFile(sandbox, "step-"+step.ID, body) - if err != nil { - return "", nil, err - } - return runScriptFile(rc, runner, path) -} - -func scriptBody(rc RunCtx, path string) (string, error) { - src, ok := rc.(interface { - Workflow() (workflow.Launch, bool) - }) - if !ok { - return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) - } - launch, ok := src.Workflow() - if !ok { - return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) - } - body, ok := launch.Scripts[path] - if !ok || strings.TrimSpace(body) == "" { - return "", fmt.Errorf("workflow: 脚本 %s 不在启动副本里", path) - } - return body, nil -} - -func registeredWorkflowLaunch(rc RunCtx, def core.WorkflowDef, slots map[string]string) workflow.Launch { - launch := workflow.Launch{Def: def, Slots: slots, Scripts: map[string]string{}} - lookup, ok := rc.(interface { - WorkflowScript(id, path string) (string, bool) - }) - if !ok { - return launch - } - for _, step := range def.Steps { - if step.Kind != "script" || strings.TrimSpace(step.Script) == "" { - continue - } - if body, found := lookup.WorkflowScript(def.ID, step.Script); found { - launch.Scripts[step.Script] = body - } - } - return launch -} - -func runModelScript(rc RunCtx, runner WorkflowRunner, body string) (string, []*core.ArtifactRef, error) { - if !scriptsEnabled(rc) { - return "", nil, fmt.Errorf("workflow: 工作流脚本未开启") - } - preview := body - if len(preview) > 4000 { - preview = preview[:4000] + "\n…（脚本超长，已截断展示）" - } - resp, err := 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, err := scriptSandbox(rc) - if err != nil { - return "", nil, err - } - path, err := writeScriptFile(sandbox, "model", body) - if err != nil { - return "", nil, err - } - return runScriptFile(rc, runner, path) -} - -func 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 -} - -func runScriptFile(rc RunCtx, runner WorkflowRunner, path string) (string, []*core.ArtifactRef, error) { - if !scriptsEnabled(rc) { - return "", nil, fmt.Errorf("workflow: 工作流脚本未开启") - } - sandbox, err := scriptSandbox(rc) - if err != nil { - return "", nil, err - } - timeout := runner.ScriptTimeout - if timeout <= 0 { - timeout = scriptDefaultTimeout - } - parent := rc.Context() - if parent == nil { - parent = context.Background() - } - var arts []*core.ArtifactRef - out := codeorch.RunFile(parent, sandbox, path, codeorch.Limits{ - Timeout: timeout, - MaxTimeout: scriptMaxTimeout, - MaxCalls: scriptMaxCalls, - MaxResultBytes: scriptMaxLine, - MaxModelChars: 16000, - FullDir: sandbox, - }, func(ctx context.Context, dir, scriptPath string) (*exec.Cmd, error) { - return tools.ScriptCommand(ctx, dir, scriptPath) - }, func(_ context.Context, call codeorch.Call) (string, error) { - output, got, err := dispatchScript(rc, call) - arts = append(arts, got...) - return output, err - }) - return out.ModelText(), arts, out.Err -} - -func dispatchScript(rc RunCtx, req codeorch.Call) (string, []*core.ArtifactRef, error) { - var choices []core.HITLChoice - if len(req.Choices) > 0 { - _ = json.Unmarshal(req.Choices, &choices) - } - switch req.Op { - case "tool.call": - if _, denied := workflow.DeniedTools[req.Name]; denied { - return "", nil, fmt.Errorf("不能调用 %s", req.Name) - } - res := 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}) - } - arts = append(arts, artifactsIn(rc, res.Output)...) - return res.Output, arts, nil - case "hitl.ask": - 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 := 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": - agent := req.Agent - if strings.TrimSpace(agent) == "" { - agent = "general" - } - var res core.AgentResultPayload - var err error - if sp, ok := rc.(interface { - SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) - }); ok { - res, err = sp.SpawnAgent(subagentProfileID(agent), req.Title, req.Task, "") - } else { - res, err = rc.SpawnSubagent(req.Title, req.Task, "") - } - if err != nil { - return "", nil, err - } - text := res.Text - if text == "" { - text = res.Summary - } - return text, artifactsIn(rc, text), nil - case "artifact.register": - name := req.Name - if name == "" { - name = "script.md" - } - art, err := 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 := rc.CreateCard(core.ActivityCard{Title: title, Kind: kind}) - if err != nil { - return "", nil, err - } - return string(created.ID), nil, nil - case "step": - applyScriptPlan(rc, req.StepID, req.Title, req.Status) - return req.Status, nil, nil - default: - return "", nil, fmt.Errorf("未知回调 %s", req.Op) - } -} - -func applyScriptPlan(rc RunCtx, id, title, status string) { - switch status { - case "pending", "in_progress", "done", "skipped": - default: - status = "done" - } - snap := core.PlanSnapshot{} - if src, ok := rc.(interface { - Plan() (core.PlanSnapshot, bool) - }); 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}) - } - _ = rc.UpdatePlan(snap) -} diff --git a/internal/runtime/workflow_script_test.go b/internal/runtime/workflow_script_test.go deleted file mode 100644 index c26ac10..0000000 --- a/internal/runtime/workflow_script_test.go +++ /dev/null @@ -1,243 +0,0 @@ -package runtime - -import ( - "context" - "os" - "path/filepath" - "strings" - "testing" - "time" - - "agent-system-backend/internal/workflow" - core "agent-system-core" -) - -func scriptSettings() core.EffectiveSettings { - return core.EffectiveSettings{"runtime": map[string]any{"workflowScripts": true}} -} - -func TestScriptStepRunsEachCallback(t *testing.T) { - body := `import agentwf as wf -print("tool", wf.tool("time_now", {})) -print("ask", wf.ask("继续？", [{"id":"yes","label":"好","impact":"继续"}])) -print("agent", wf.agent("research", "写一句")) -print("art", wf.artifact("note.md", "你好")) -print("card", wf.card("结果", "result")) -print("step", wf.step("s1", "一步", "done")) -print("file", wf.tool("make_file", {})) -` - f := &wfFake{ - sandbox: t.TempDir(), - settings: scriptSettings(), - hitl: []core.HITLResponse{{ChoiceID: "yes"}}, - launch: workflow.Launch{ - Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{ - {ID: "run", Title: "跑", Kind: "script", Script: "workflows/scripts/demo.py"}, - {ID: "save", Title: "留下", Kind: "artifact", Prompt: "{{steps.run.output}}"}, - }}, - Scripts: map[string]string{"workflows/scripts/demo.py": body}, - }, - } - if err := (WorkflowRunner{}).Execute(f); err != nil { - t.Fatal(err) - } - saved := f.writes[len(f.writes)-1] - for _, want := range []string{"tool 12:00", "ask yes", "agent 正文", "art art_aaaaaaaaaaaa", "card card_1", "file wrote"} { - if !strings.Contains(saved, want) { - t.Fatalf("missing %q in %s", want, saved) - } - } - if len(f.tools) != 2 || f.tools[0].Name != "time_now" || f.tools[1].Name != "make_file" { - t.Fatalf("tools %+v", f.tools) - } - if len(f.cards) != 1 || f.cards[0].Title != "结果" { - t.Fatalf("cards %+v", f.cards) - } - if len(f.spawnIDs) != 1 || f.spawnIDs[0] != "agent.sub.research" { - t.Fatalf("spawn %v", f.spawnIDs) - } - if f.writes[0] != "note.md=你好" { - t.Fatalf("writes %v", f.writes) - } - sawStep := false - var ids []string - for _, plan := range f.plans { - for _, step := range plan.Steps { - if step.ID == "s1" && step.Status == "done" { - sawStep = true - } - } - } - for _, part := range f.assistant { - if part.Artifact != nil { - ids = append(ids, string(part.Artifact.ArtifactID)) - } - } - if !sawStep { - t.Fatalf("plans %+v", f.plans) - } - joined := strings.Join(ids, " ") - if !strings.Contains(joined, "art_aaaaaaaaaaaa") || !strings.Contains(joined, "art_bbbbbbbbbbbb") { - t.Fatalf("deliverables %v", ids) - } -} - -func TestScriptRejectsDeniedTool(t *testing.T) { - dir := t.TempDir() - body := `import agentwf as wf -try: - wf.tool("run_python", {}) -except Exception as exc: - open("ERR", "w").write(str(exc)) -` - f := &wfFake{ - sandbox: dir, - settings: scriptSettings(), - launch: workflow.Launch{ - Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ - ID: "run", Title: "跑", Kind: "script", Script: "a.py", - }}}, - Scripts: map[string]string{"a.py": body}, - }, - } - if err := (WorkflowRunner{}).Execute(f); err != nil { - t.Fatal(err) - } - raw, err := os.ReadFile(filepath.Join(dir, "ERR")) - if err != nil { - t.Fatal(err) - } - if !strings.Contains(string(raw), "不能调用 run_python") || len(f.tools) != 0 { - t.Fatalf("err %s tools %+v", raw, f.tools) - } -} - -func TestScriptStopsAfterCallCap(t *testing.T) { - body := `import agentwf as wf -for i in range(60): - wf.tool("time_now", {}) -` - f := &wfFake{ - sandbox: t.TempDir(), - settings: scriptSettings(), - launch: workflow.Launch{ - Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ - ID: "run", Title: "跑", Kind: "script", Script: "a.py", - }}}, - Scripts: map[string]string{"a.py": body}, - }, - } - err := (WorkflowRunner{}).Execute(f) - if err == nil || !strings.Contains(err.Error(), "超过 50") || len(f.tools) != 50 { - t.Fatalf("err %v calls %d", err, len(f.tools)) - } -} - -func TestScriptCancelKillsProcess(t *testing.T) { - dir := t.TempDir() - body := `import os, time -open("PID", "w").write(str(os.getpid())) -time.sleep(30) -open("AFTER", "w").write("ran") -` - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - f := &wfFake{ - ctx: ctx, - sandbox: dir, - settings: scriptSettings(), - launch: workflow.Launch{ - Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ - ID: "run", Title: "跑", Kind: "script", Script: "a.py", - }}}, - Scripts: map[string]string{"a.py": body}, - }, - } - done := make(chan error, 1) - go func() { done <- (WorkflowRunner{}).Execute(f) }() - deadline := time.Now().Add(5 * time.Second) - for { - if _, err := os.Stat(filepath.Join(dir, "PID")); err == nil { - break - } - if time.Now().After(deadline) { - cancel() - t.Fatal("script did not start") - } - time.Sleep(20 * time.Millisecond) - } - raw, err := os.ReadFile(filepath.Join(dir, "PID")) - if err != nil { - t.Fatal(err) - } - cancel() - select { - case err = <-done: - if err == nil || !strings.Contains(err.Error(), "取消") { - t.Fatalf("%v", err) - } - case <-time.After(5 * time.Second): - t.Fatal("cancel did not stop the script") - } - if _, err := os.Stat("/proc/" + strings.TrimSpace(string(raw))); err == nil { - t.Fatal("script process still alive") - } - if _, err := os.Stat(filepath.Join(dir, "AFTER")); err == nil { - t.Fatal("script kept running after cancel") - } -} - -func TestScriptLineLimit(t *testing.T) { - body := `import os -f = os.fdopen(3, "w", encoding="utf-8", buffering=1) -f.write("x" * (1024 * 1024 + 32) + "\n") -f.flush() -` - f := &wfFake{ - sandbox: t.TempDir(), - settings: scriptSettings(), - launch: workflow.Launch{ - Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ - ID: "run", Title: "跑", Kind: "script", Script: "a.py", - }}}, - Scripts: map[string]string{"a.py": body}, - }, - } - err := (WorkflowRunner{}).Execute(f) - if err == nil || !strings.Contains(err.Error(), "1MB") { - t.Fatal(err) - } -} - -func TestRegisteredWorkflowUsesItsOwnScriptCopy(t *testing.T) { - f := &wfFake{sandbox: t.TempDir(), settings: scriptSettings()} - res, err := (WorkflowRunner{}).run(f, workflow.Launch{ - Def: core.WorkflowDef{ID: "demo.script", Title: "脚本", Steps: []core.WorkflowStep{{ - ID: "run", Title: "跑", Kind: "script", Script: "workflows/scripts/demo.py", - }}}, - Scripts: map[string]string{"workflows/scripts/demo.py": "print('from-launch')\n"}, - }) - if err != nil || !strings.Contains(res.Summary, "完成") { - t.Fatalf("err %v summary %s", err, res.Summary) - } -} - -func TestModelScriptNeedsSwitchAndConfirm(t *testing.T) { - f := &wfFake{sandbox: t.TempDir()} - out := (&directRun{rc: f}).runRegisteredWorkflow(`{"script":"print('pong')"}`) - if !strings.Contains(out, "未开启") || len(f.hitlQ) != 0 { - t.Fatalf("%s questions %v", out, f.hitlQ) - } - - f2 := &wfFake{sandbox: t.TempDir(), settings: scriptSettings(), hitl: []core.HITLResponse{{ChoiceID: "no"}}} - out = (&directRun{rc: f2}).runRegisteredWorkflow(`{"script":"print('pong')"}`) - if out != "未确认，未执行" || !strings.Contains(f2.hitlQ[0], "print('pong')") { - t.Fatalf("%s questions %v", out, f2.hitlQ) - } - - f3 := &wfFake{sandbox: t.TempDir(), settings: scriptSettings(), hitl: []core.HITLResponse{{ChoiceID: "yes"}}} - out = (&directRun{rc: f3}).runRegisteredWorkflow(`{"script":"print('pong')"}`) - if !strings.Contains(out, "pong") { - t.Fatalf("%s", out) - } -} diff --git a/internal/server/e2e_test.go b/internal/server/e2e_test.go index cd1b188..5cae00b 100644 --- a/internal/server/e2e_test.go +++ b/internal/server/e2e_test.go @@ -6,6 +6,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" agentcommand "agent-command" "agent-system-backend/internal/cardsvc" @@ -15,17 +16,23 @@ import ( "net/http" "net/http/httptest" "os" + osexec "os/exec" "path/filepath" "strings" "testing" "time" + agentllm "agent-native/runtime/llm" + nativetools "agent-native/runtime/tools" + nativeadp "agent-squad/adapters/native" "agent-system-backend/internal/bus" "agent-system-backend/internal/intentgate" "agent-system-backend/internal/plugins" "agent-system-backend/internal/profiles" "agent-system-backend/internal/runmgr" "agent-system-backend/internal/runtime" + skillseed "agent-system-backend/internal/skills" + "agent-system-backend/internal/squadadapter" "agent-system-backend/internal/settings" "agent-system-backend/internal/store" "agent-system-backend/internal/tools" @@ -65,6 +72,62 @@ func newTestServer(t *testing.T) (*Server, *store.Store) { return srv, st } +// stubNativeStream 内核桩模型流（零网络）：首轮按任务书「期望产出」逐个 +// write_file 落盘（agent-native 契约：workspace 档步骤自己写交付物）， +// 见到工具结果后给收尾文本。 +func stubNativeStream(text string) agentllm.StreamFunc { + return func(_ context.Context, _ string, msgs []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event { + ch := make(chan agentllm.Event, 8) + go func() { + defer close(ch) + finish := func() { + ch <- agentllm.Event{Kind: agentllm.KindTextDelta, Text: text} + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, + Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: text}} + } + prompt := "" + for _, m := range msgs { + if m.Role == agentllm.RoleTool { + finish() + return + } + if m.Role == agentllm.RoleUser { + prompt = m.Content + } + } + targets := stubTargets(prompt) + if len(targets) == 0 { + finish() + return + } + var calls []agentllm.ToolCall + for i, tgt := range targets { + args := fmt.Sprintf(`{"path":%q,"content":%q}`, tgt, text) + id := fmt.Sprintf("call_stub_%d", i+1) + ch <- agentllm.Event{Kind: agentllm.KindToolCallDelta, ToolIndex: i, + ToolID: id, ToolName: "write_file", ArgumentsDelta: args} + calls = append(calls, agentllm.ToolCall{ID: id, Name: "write_file", Arguments: args}) + } + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, + Message: agentllm.Message{Role: agentllm.RoleAssistant, ToolCalls: calls, StopReason: agentllm.StopToolCalls}} + }() + return ch + } +} + +// stubTargets 从任务书解析「期望产出：[a.md b.md]」。 +func stubTargets(prompt string) []string { + i := strings.Index(prompt, "期望产出：[") + if i < 0 { + return nil + } + j := strings.Index(prompt[i:], "]") + if j < 0 { + return nil + } + return strings.Fields(prompt[i+len("期望产出：[") : i+j]) +} + func doJSON(t *testing.T, h http.Handler, method, path string, body any) (int, map[string]any) { t.Helper() var buf bytes.Buffer @@ -853,9 +916,22 @@ func TestHITLIntentPivot(t *testing.T) { } // TestSquadOrchestration：Agent-Squad 引擎真实驱动——规则规划器四阶段计划、 -// 每步经 "agent-system" 后端派生子代理（任务树）、计划事件、交付物工件。 +// Q3 WP2 后：每步经 "agent-native" 内核后端在进程内执行（桩 Stream，零网络； +// SpawnSubagent 压缩路径已退役——步骤不再派生任务树子 Run）、计划事件、交付物工件。 func TestSquadOrchestration(t *testing.T) { srv, _ := newTestServer(t) + skillsRoot := t.TempDir() + if _, serr := skillseed.Seed(skillsRoot); serr != nil { + t.Fatal(serr) + } + savedNative := squadadapter.AgentNativeOptions + squadadapter.AgentNativeOptions = func(squadadapter.Ctx) (nativeadp.Config, bool) { + py, _ := osexec.LookPath("python3") + return nativeadp.Config{DefaultModel: "stub", SkillsRoots: []string{skillsRoot}, + Exec: &nativetools.ExecConfig{Bash: "/bin/bash", Python: py}, Stream: stubNativeStream( + "# 交付\n\n这是内核桩后端产出的步骤交付物正文，长度超过五十个字符以满足最小质检要求。")}, true + } + defer func() { squadadapter.AgentNativeOptions = savedNative }() h := srv.Handler() projectID, convID := setupConversation(t, h) runID := postMessage(t, h, convID, "#squad 生成一份示例目标简报") @@ -889,17 +965,12 @@ func TestSquadOrchestration(t *testing.T) { t.Fatalf("squad run status: %v (err: %v)", terminal["status"], terminal["error"]) } - // 1) 任务树：squad 步骤全部成为子代理 run。 + // 1) 任务树：内核后端进程内执行步骤，不再派生子代理 run + //（SpawnSubagent 压缩路径退役的行为面）。 _, out := doJSON(t, h, "GET", "/api/runs/"+runID, nil) children := out["children"].([]any) - if len(children) < 3 { - t.Fatalf("expected >=3 squad step subagents, got %d", len(children)) - } - for _, ci := range children { - c := ci.(map[string]any) - if c["kind"] != "subagent" || c["parentRunId"] != runID { - t.Errorf("bad squad child: %+v", c) - } + if len(children) != 0 { + t.Fatalf("kernel steps must not spawn subagent runs, got %d children", len(children)) } // 2) 计划快照：四阶段骨架步骤映射到右栏计划。 diff --git a/internal/squadadapter/adapter.go b/internal/squadadapter/adapter.go index 09a3771..51d1490 100644 --- a/internal/squadadapter/adapter.go +++ b/internal/squadadapter/adapter.go @@ -4,8 +4,8 @@ // 装配映射： // - RoleRegistry → 内置小队角色（规则规划器自主编排四阶段骨架） // - Planner → squad RulePlanner（确定性；LLM 规划器为后续档） -// - BackendResolver→ 自定义 "agent-system" 后端工厂：每步执行 = rc.SpawnSubagent -// （子 Run 进任务树，走既有 mock/LLM 子代理执行器与事件） +// - BackendResolver→ "agent-native" 内核后端工厂（Q3 WP2：SpawnSubagent +// 压缩路径已退役；每步执行 = agent-native 内核循环） // - ProgressSink → 步骤事件映射 plan.update / agent.progress // - HITL（挂起） → 引擎 Pending HumanRequest ↔ rc.AskHITL 卡片；RespondDecision 回填 package squadadapter @@ -13,7 +13,6 @@ package squadadapter import ( "context" "fmt" - "os" "path/filepath" "strings" @@ -69,14 +68,14 @@ func Orchestrate(rc Ctx, dataDir, goalText string) (string, error) { } } - backends := buildBackends(rc) + backends := BuildBackends(rc) engine, err := squad.NewEngine(squad.Options{ DataDir: runDir, Registry: registry, Planner: squad.NewRulePlanner(), Backends: backends, - DefaultBackend: "agent-system", + DefaultBackend: "agent-native", StepTimeout: -1, // 负数 = 不设单步截止（引擎只在 0 时回落到 10 分钟） HitlPolicy: squad.HitlPolicy{Mode: squad.HitlInteractive}, Checker: quality.Checker{}, @@ -147,70 +146,6 @@ func Orchestrate(rc Ctx, dataDir, goalText string) (string, error) { } } -// stepBackend：小队步骤 → 统一子 Run（任务树可见；执行走 runmgr 的子代理执行器）。 -type stepBackend struct { - rc Ctx - role squad.RoleSpec -} - -func (b *stepBackend) Name() string { return "agent-system" } - -// profileForRole 把小队角色键映射到子代理 profile。未列出的角色走 general。 -func profileForRole(role string) string { - switch role { - case "planner", "organizer": - return "agent.sub.plan" - case "researcher", "analyst": - return "agent.sub.research" - case "writer": - return "agent.sub.writer" - case "reviewer", "leader": - return "agent.sub.verify" - default: - return "agent.sub.general" - } -} - -func (b *stepBackend) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) { - task := strings.TrimSpace(req.Prompt) - if task == "" { - task = req.Instructions - } - hint := strings.TrimSpace(req.Instructions) - name := firstNonEmpty(b.role.Name, b.role.Role) - var out core.AgentResultPayload - var err error - if sp, ok := b.rc.(interface { - SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) - }); ok { - out, err = sp.SpawnAgent(profileForRole(b.role.Role), name, task, hint) - } else { - out, err = b.rc.SpawnSubagent(name, task, hint) - } - res := &squad.ExecResult{Success: err == nil, Logs: firstNonEmpty(out.Summary, "子代理完成")} - if err != nil { - res.ErrorReason = err.Error() - return res, nil - } - // 交付物：子代理全文写入步骤子工作区的目标文件（引擎契约：后端自行落盘， - // 检查器按 allTargetsPresent 校验；Patches 同时回传供引擎重试路径复用）。 - content := firstNonEmpty(out.Text, out.Summary, "（子代理无文本产出）") - targets := req.Targets - if len(targets) == 0 { - targets = []string{"output.md"} - } - res.Patches = make(map[string][]byte, len(targets)) - for _, t := range targets { - body := []byte(content + "\n") - res.Patches[t] = body - dest := filepath.Join(req.WorkDir, filepath.FromSlash(t)) - if merr := os.MkdirAll(filepath.Dir(dest), 0o755); merr == nil { - _ = os.WriteFile(dest, body, 0o644) - } - } - return res, nil -} - // askHuman：HumanRequest → HITL 卡片 → RespondDecision（选项直达 / 自由文本作答）。 func askHuman(rc Ctx, req *squad.HumanRequest) (squad.RespondDecision, error) { choices := make([]core.HITLChoice, 0, len(req.Options)) @@ -342,15 +277,12 @@ func SubagentFactory(rc Ctx) subagent.Factory { } } -// buildBackends 组装小队后端解析器："agent-system"（SpawnSubagent 路径）恒注册； -// 组合根注入 AgentNativeOptions 时额外注册 "agent-native" 内核工厂——角色显式 -// 声明 Backend:"agent-native" 才会走内核执行；默认后端仍是 agent-system， -// 不因工厂注册而切换（对齐 parity 渐进启用条款）。 -func buildBackends(rc Ctx) *squad.BackendResolver { +// BuildBackends 组装小队后端解析器（Q3 WP2）：只注册 "agent-native" 内核 +// 工厂（组合根注入 AgentNativeOptions 且模型已配置时）。legacy 的 +// "agent-system"（SpawnSubagent 压缩路径）已退役删除；工厂缺席时解析 +// fail-closed——步骤失败可见，不静默降级到别的执行面。 +func BuildBackends(rc Ctx) *squad.BackendResolver { backends := squad.NewBackendResolver() - backends.Register("agent-system", func(role squad.RoleSpec) (squad.Executor, error) { - return &stepBackend{rc: rc, role: role}, nil - }) if AgentNativeOptions != nil { if cfg, ok := AgentNativeOptions(rc); ok { nativeadp.RegisterFull(backends, cfg) @@ -372,23 +304,24 @@ func builtinRoles() []squad.RoleSpec { } return fallback } - // 执行配置显式化（agent-native 后端消费；agent-system 路径忽略不生效但保持声明一致）： + // 执行配置显式化（agent-native 后端消费）： // 规划/调研/审校一律 BashPermission=deny（无命令执行需求）； - // 撰稿员需要 run_python 生成文档——ask 档（每次运行首次使用前人工放行）。 - // 工具白名单按角色最小化；撰稿员技能指向启动种子的三套交付配方。 + // 撰稿员需要 run_python/run_command 生成文档——ask 档（每次运行首次使用前人工放行）。 + // 工具白名单按角色最小化（Q3 WP2：agent-native 下步骤须自己落盘交付物， + // 全员补 write_file）；撰稿员技能指向启动种子的三套交付配方。 return []squad.RoleSpec{ - {Role: "planner", Name: "规划员", Backend: "agent-system", BashPermission: squad.BashPermissionDeny, - Tools: []string{"read_file", "list_dir", "grep"}, + {Role: "planner", Name: "规划员", Backend: "agent-native", BashPermission: squad.BashPermissionDeny, + Tools: []string{"read_file", "list_dir", "grep", "write_file"}, Keywords: []string{"梳理", "规划", "分析"}, Instructions: instr("planner", "梳理目标与材料，产出结构化要点与执行建议。")}, - {Role: "researcher", Name: "调研员", Backend: "agent-system", BashPermission: squad.BashPermissionDeny, + {Role: "researcher", Name: "调研员", Backend: "agent-native", BashPermission: squad.BashPermissionDeny, Tools: []string{"web_search", "tavily_search", "fetch_url", "read_file", "write_file"}, Keywords: []string{"调研", "检索", "资料"}, Instructions: instr("researcher", "围绕目标调研关键问题，输出简明结论与依据。")}, - {Role: "writer", Name: "撰稿员", Backend: "agent-system", BashPermission: squad.BashPermissionAsk, - Tools: []string{"read_file", "write_file", "edit_file", "multi_edit", "list_dir", "grep", "glob", "load_skill", "run_python", "run_shell"}, + {Role: "writer", Name: "撰稿员", Backend: "agent-native", BashPermission: squad.BashPermissionAsk, + Tools: []string{"read_file", "write_file", "edit_file", "multi_edit", "list_dir", "grep", "glob", "load_skill", "run_python", "run_command"}, Skills: []string{"docx-author", "xlsx-author", "pptx-author"}, Keywords: []string{"撰写", "成文", "报告"}, Instructions: instr("writer", "把前序要点与调研结论整理为清晰成文的交付文档。")}, - {Role: "reviewer", Name: "审校员", Backend: "agent-system", BashPermission: squad.BashPermissionDeny, - Tools: []string{"read_file", "grep", "glob", "list_dir"}, + {Role: "reviewer", Name: "审校员", Backend: "agent-native", BashPermission: squad.BashPermissionDeny, + Tools: []string{"read_file", "grep", "glob", "list_dir", "write_file"}, Keywords: []string{"审校", "把关", "终审"}, Instructions: instr("reviewer", "终审交付文档：一致性、完整性与风险提示，给出通过或修改意见。")}, } } diff --git a/internal/squadadapter/adapter_test.go b/internal/squadadapter/adapter_test.go index b186571..64cf621 100644 --- a/internal/squadadapter/adapter_test.go +++ b/internal/squadadapter/adapter_test.go @@ -36,8 +36,8 @@ func TestBuiltinRolesValid(t *testing.T) { if err := reg.Register(r); err != nil { t.Errorf("role %s invalid: %v", r.Role, err) } - if r.Backend != "agent-system" { - t.Errorf("role %s backend: %s", r.Role, r.Backend) + if r.Backend != "agent-native" { + t.Errorf("role %s backend: %s", r.Backend, r.Backend) } } } @@ -115,53 +115,6 @@ func TestSubagentFactoryBridgesSpawnSubagent(t *testing.T) { } } -type agentFakeRc struct { - subagentFakeRc - agentID string -} - -func (f *agentFakeRc) SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) { - f.agentID = agentID - f.gotName, f.gotTask = name, task - return f.payload, f.err -} - -func TestStepRoleMapsToProfile(t *testing.T) { - dir := t.TempDir() - cases := []struct { - role string - want string - }{ - {"planner", "agent.sub.plan"}, - {"organizer", "agent.sub.plan"}, - {"researcher", "agent.sub.research"}, - {"analyst", "agent.sub.research"}, - {"writer", "agent.sub.writer"}, - {"reviewer", "agent.sub.verify"}, - {"leader", "agent.sub.verify"}, - {"specialist", "agent.sub.general"}, - } - for _, tc := range cases { - rc := &agentFakeRc{subagentFakeRc: subagentFakeRc{payload: core.AgentResultPayload{Text: "ok"}}} - b := &stepBackend{rc: rc, role: squad.RoleSpec{Role: tc.role, Name: "显示名"}} - res, err := b.Execute(context.Background(), squad.ExecRequest{Prompt: "做这件事", WorkDir: dir, Targets: []string{"output.md"}}) - if err != nil || !res.Success { - t.Fatalf("%s: %+v %v", tc.role, res, err) - } - if rc.agentID != tc.want || rc.gotName != "显示名" { - t.Fatalf("%s mapped to %s name %s", tc.role, rc.agentID, rc.gotName) - } - } - plain := &subagentFakeRc{payload: core.AgentResultPayload{Text: "ok"}} - b := &stepBackend{rc: plain, role: squad.RoleSpec{Role: "writer", Name: "撰稿"}} - if _, err := b.Execute(context.Background(), squad.ExecRequest{Prompt: "写", WorkDir: dir}); err != nil { - t.Fatal(err) - } - if plain.gotName != "撰稿" { - t.Fatalf("fallback name = %s", plain.gotName) - } -} - func TestBuiltinRolesExplicitExecConfig(t *testing.T) { roles := builtinRoles() byKey := map[string]squad.RoleSpec{} @@ -210,3 +163,16 @@ func containsStr(list []string, want string) bool { } return false } + + +// Q3 WP2：agent-system 退役后，未注入 AgentNativeOptions 时解析器为空—— +// 步骤后端解析 fail-closed，不再静默走 SpawnSubagent 压缩路径。 +func TestBuildBackendsWithoutNativeOptionsIsEmpty(t *testing.T) { + saved := AgentNativeOptions + AgentNativeOptions = nil + defer func() { AgentNativeOptions = saved }() + r := BuildBackends(&subagentFakeRc{}) + if _, _, err := r.Resolve(squad.RoleSpec{Role: "writer", Backend: "agent-native"}, "agent-native"); err == nil { + t.Fatal("无内核工厂时解析必须 fail-closed") + } +} diff --git a/internal/squadadapter/entry.go b/internal/squadadapter/entry.go index 920aa03..902c3e7 100644 --- a/internal/squadadapter/entry.go +++ b/internal/squadadapter/entry.go @@ -83,7 +83,7 @@ func RunEntry(rc Ctx, opt EntryOptions) (EntryResult, error) { return EntryResult{}, fmt.Errorf("squad entry: register %s: %w", role.Role, err) } } - backends := buildBackends(rc) + backends := BuildBackends(rc) nativeadp.RegisterFull(backends, cfg) checker := opt.Checker diff --git a/internal/squadadapter/orchestrate_native_test.go b/internal/squadadapter/orchestrate_native_test.go new file mode 100644 index 0000000..60906c0 --- /dev/null +++ b/internal/squadadapter/orchestrate_native_test.go @@ -0,0 +1,99 @@ +package squadadapter + +import ( + "context" + "fmt" + osexec "os/exec" + "strings" + "testing" + + nativetools "agent-native/runtime/tools" + + agentllm "agent-native/runtime/llm" + nativeadp "agent-squad/adapters/native" + "agent-system-backend/internal/skills" + core "agent-system-core" +) + +type orchFakeRc struct { + subagentFakeRc + dir string +} + +func (f *orchFakeRc) AskHITL(card core.HITLCard) (core.HITLResponse, error) { + if len(card.Choices) > 0 { + return core.HITLResponse{ChoiceID: card.Choices[0].ID}, nil + } + return core.HITLResponse{ChoiceID: "approve"}, nil +} + +// TestOrchestrateRunsOnKernelStub Q3 WP2 行为面：agent-system 退役后， +// Orchestrate 的四阶段骨架经 agent-native 内核（桩 Stream + write_file 落盘 +// 交付物 + 种子技能 + Exec 白名单）跑通并产出交付物工件。 +func TestOrchestrateRunsOnKernelStub(t *testing.T) { + skillsRoot := t.TempDir() + if _, serr := skills.Seed(skillsRoot); serr != nil { + t.Fatal(serr) + } + saved := AgentNativeOptions + AgentNativeOptions = func(Ctx) (nativeadp.Config, bool) { + py, _ := osexec.LookPath("python3") + return nativeadp.Config{DefaultModel: "stub", SkillsRoots: []string{skillsRoot}, + Exec: &nativetools.ExecConfig{Bash: "/bin/bash", Python: py}, Stream: stubStream("# 交付\n\n这是内核桩后端产出的步骤交付物正文，长度超过五十个字符以满足最小质检要求。")}, true + } + defer func() { AgentNativeOptions = saved }() + rc := &orchFakeRc{dir: t.TempDir()} + dataDir := t.TempDir() + summary, err := Orchestrate(rc, dataDir, "生成一份示例目标简报") + t.Logf("summary = %q", summary) + if err != nil { + t.Fatalf("err = %v", err) + } + if !strings.Contains(summary, "completed") || !strings.Contains(summary, "完成 4") { + t.Fatalf("summary = %q", summary) + } +} + + +func stubStream(text string) agentllm.StreamFunc { + return func(_ context.Context, _ string, msgs []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event { + ch := make(chan agentllm.Event, 8) + go func() { + defer close(ch) + finish := func() { + ch <- agentllm.Event{Kind: agentllm.KindTextDelta, Text: text} + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: text}} + } + prompt := "" + for _, m := range msgs { + if m.Role == agentllm.RoleTool { + finish() + return + } + if m.Role == agentllm.RoleUser { + prompt = m.Content + } + } + i := strings.Index(prompt, "期望产出：[") + if i < 0 { + finish() + return + } + j := strings.Index(prompt[i:], "]") + if j < 0 { + finish() + return + } + targets := strings.Fields(prompt[i+len("期望产出：[") : i+j]) + var calls []agentllm.ToolCall + for k, tgt := range targets { + args := fmt.Sprintf(`{"path":%q,"content":%q}`, tgt, text) + id := fmt.Sprintf("call_stub_%d", k+1) + ch <- agentllm.Event{Kind: agentllm.KindToolCallDelta, ToolIndex: k, ToolID: id, ToolName: "write_file", ArgumentsDelta: args} + calls = append(calls, agentllm.ToolCall{ID: id, Name: "write_file", Arguments: args}) + } + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, ToolCalls: calls, StopReason: agentllm.StopToolCalls}} + }() + return ch + } +}