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) } else { deps.JudgeStream = GuardProjectStream(rc, "", deps.JudgeStream) } 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 := openRouteClient(route, rc.Settings()) model := route.Model base := 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) } return GuardProjectStream(rc, "", base) } // 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 "" } func (h workflowHost) InvalidateProjectStepOutput(ctx context.Context, stepID string) error { invalidate, ok := h.rc.(interface { InvalidateProjectStepOutput(context.Context, string) error }) if !ok { return fmt.Errorf("project workflow step invalidation unavailable") } return invalidate.InvalidateProjectStepOutput(ctx, stepID) } func (h workflowHost) BindProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) error { binder, ok := h.rc.(interface { BindProjectStepOutput(context.Context, string, string, string, time.Time, []byte) error }) if !ok { return fmt.Errorf("project workflow step binding unavailable") } return binder.BindProjectStepOutput(ctx, stepID, stepAttemptID, kind, startedAt, output) } func (h workflowHost) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, output []byte) error { verifier, ok := h.rc.(interface { VerifyProjectStepOutput(context.Context, string, string, string, []byte) error }) if !ok { return fmt.Errorf("project workflow step verification unavailable") } return verifier.VerifyProjectStepOutput(ctx, stepID, stepAttemptID, kind, output) }