Agent-System-Backend/internal/runtime/workflow_compiled.go

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, route.Model)
	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 ""
}