Agent-System-Backend/internal/wfcompile/run.go

package wfcompile

import (
	"context"
	"fmt"
	"path/filepath"
	"strings"
	"time"

	"agent-squad/squad"
	"agent-system-backend/internal/workflow"
	core "agent-system-core"
)

// run.go — 编译图执行驱动（Q3 WP1）：槽位补齐 → 编译 → squad 引擎装配 →
// 步进循环（挂起 → HITL 卡片 → Respond；执行态 → Resume）→ critic 终判 →
// legacy 同款 outcome 汇总。

// Outcome 是一次编译图执行的结果（legacy workflowOutcome 同款形态）。
type Outcome struct {
	Summary   string
	Parts     []core.Part
	Artifacts []*core.ArtifactRef
}

// RunOptions 编译图执行口径。
type RunOptions struct {
	// DynamicLimits 模型生成上限（≤8 步、禁 script）。
	DynamicLimits bool
	// RequireConfirm 执行前强制计划确认（模型生成工作流与模型脚本必须确认；
	// 注册的声明式工作流免确认）。
	RequireConfirm bool
	// ModelScript 脚本步为模型提供（执行前单独 HITL 确认）。
	ModelScript bool
}

// RunCompiled 编译并执行一个工作流。vars.Input 提供 {{input}} 实值；
// launch.Missing 的槽位在执行前经 HITL 收齐（legacy run() 同款）。
func RunCompiled(rc Host, deps Deps, launch workflow.Launch, vars workflow.Vars, opts RunOptions) (Outcome, error) {
	def := launch.Def
	if vars.Slots == nil {
		vars.Slots = map[string]string{}
	}
	if vars.Steps == nil {
		vars.Steps = map[string]string{}
	}
	for k, v := range launch.Slots {
		vars.Slots[k] = v
	}
	// 缺失槽位补齐（legacy 同款：逐个 HITL input 卡片）。
	for _, missing := range launch.Missing {
		if rc.Context().Err() != nil {
			return Outcome{}, rc.Context().Err()
		}
		resp, err := rc.AskHITL(core.HITLCard{
			Question: missing.Question, Kind: core.HITLKindInput, AllowFreeInput: true,
		})
		if err != nil {
			return Outcome{}, err
		}
		answer := strings.TrimSpace(resp.FreeText)
		if answer == "" {
			answer = resp.ChoiceID
		}
		vars.Slots[missing.Key] = answer
	}

	known := map[string]bool{}
	for _, tool := range rc.Tools() {
		known[tool.Name] = true
	}
	compiled, err := Compile(def, vars, CompileOptions{
		DynamicLimits: opts.DynamicLimits,
		ModelScript:   opts.ModelScript,
	}, func(name string) bool { return known[name] })
	if err != nil {
		return Outcome{}, err
	}
	stepKinds := make(map[string]string, len(def.Steps))
	for _, step := range def.Steps {
		stepKinds[step.ID] = step.Kind
	}
	for _, pair := range compiled.Critics {
		stepKinds[pair.EvalID] = "critic"
	}
	if rc.Run().ProjectID != "" {
		if _, ok := rc.(ProjectStepProvenance); !ok {
			return Outcome{}, fmt.Errorf("项目工作流缺少可信步骤产物绑定能力")
		}
	}

	registry := squad.NewRoleRegistry()
	for _, role := range compiled.Plan.Squad.Agents {
		if err := registry.Register(role); err != nil {
			return Outcome{}, fmt.Errorf("workflow: 注册角色 %s: %w", role.Role, err)
		}
	}
	backends := deps.Backends
	if backends == nil {
		backends = squad.NewBackendResolver()
	}
	if rc.Run().ProjectID != "" {
		backends = projectStepBackends(backends, rc.(ProjectStepProvenance), compiled.SubOf, stepKinds)
	}
	runner := &hostRunner{
		rc: rc, deps: deps, scripts: launch.Scripts, subOf: compiled.SubOf,
		stepKinds: stepKinds,
		verdicts:  map[string]string{}, lastWhy: map[string]string{}, fails: map[string]int{},
	}
	var progress squad.ProgressSink = squad.NopSink{}
	if rc.Run().ProjectID != "" {
		progress = newProjectOutputSink(rc, filepath.Join(deps.DataDir, "wf", string(rc.Run().ID)), compiled.SubOf, stepKinds)
	}
	engine, err := squad.NewEngine(squad.Options{
		DataDir:        filepath.Join(deps.DataDir, "wf", string(rc.Run().ID)),
		Registry:       registry,
		Planner:        &compiledPlanner{plan: compiled.Plan, needsConfirm: opts.RequireConfirm, reason: PlanPreview(def)},
		Backends:       backends,
		DefaultBackend: orDefault(deps.DefaultBackend, "agent-system"),
		StepTimeout:    -1, // 负数 = 不设单步截止（引擎只在 0 时回落 10 分钟）
		HitlPolicy:     squad.HitlPolicy{Mode: squad.HitlInteractive},
		Checker:        passChecker{},
		IntentCatalog:  compiled.Catalog,
		IntentRunner:   runner,
		Sink:           progress,
	})
	if err != nil {
		return Outcome{}, fmt.Errorf("workflow: 引擎装配: %w", err)
	}
	// 初始计划快照（legacy 同款：执行前 UpdatePlan 全 pending）。
	_ = rc.UpdatePlan(planSnapshot(compiled, def, nil))

	res, err := engine.Run(rc.Context(), squad.Goal{Text: def.Title}, squad.RunOptions{})
	rejected := false
	for err == nil && !squad.RunStatus(res.Status).Terminal() {
		if isHoldStatus(string(res.Status)) {
			req := res.Pending
			if req == nil {
				if req, err = engine.Pending(res.RunID); err != nil || req == nil {
					break
				}
			}
			// 动态工作流的计划确认被拒 → legacy 语义「未确认，未执行」。
			decision, aerr := askHuman(rc, req)
			if aerr != nil {
				return Outcome{}, aerr
			}
			if req.Kind == squad.HitlConfirmation && decision.Choice != "" &&
				decision.Choice != "approve" && decision.Choice != "yes" {
				rejected = true
				break
			}
			res, err = engine.Respond(rc.Context(), res.RunID, decision)
		} else {
			res, err = engine.Resume(rc.Context(), res.RunID)
		}
		if err == nil && res.Plan != nil {
			_ = rc.UpdatePlan(planSnapshot(compiled, def, res.Plan))
		}
	}
	if rejected {
		return Outcome{Summary: "未确认，未执行",
			Parts: []core.Part{{Type: core.PartText, Text: "未确认，未执行"}}}, nil
	}
	finalPlan := res.Plan
	if finalPlan == nil {
		if insp, ierr := engine.Inspect(res.RunID); ierr == nil {
			finalPlan = insp.Plan
		}
	}
	_ = rc.UpdatePlan(planSnapshot(compiled, def, finalPlan))

	// critic 终判（P4：评审预算耗尽仍不通过 = 工作流失败）。
	stopped := ""
	arts, drops, verdicts, why := runner.snapshot()
	if err == nil && (res.Status == squad.RunCompleted || res.Status == squad.RunDegraded) {
		for _, cp := range compiled.Critics {
			if verdicts[cp.VerdictField] != "pass" {
				stopped = fmt.Sprintf("步骤 %s 未通过验收：%s", cp.StepTitle, orDefault(why[cp.StepID], "验收未通过"))
				err = fmt.Errorf("workflow: 步骤 %s 未通过验收", cp.StepID)
				break
			}
		}
	}
	if err == nil && res.Status != squad.RunCompleted && res.Status != squad.RunDegraded {
		// 引擎级失败（步骤失败/取消）→ legacy 的停止语义。
		msg := firstNonEmpty(res.Error, string(res.Status))
		stopped = msg
		err = fmt.Errorf("workflow: %s", msg)
	}
	statuses := stepStatuses(compiled, def, finalPlan)
	out := buildOutcome(def, statuses, arts, drops, stopped)
	if err != nil && rc.Context().Err() != nil {
		err = rc.Context().Err() // 取消优先（legacy 同款）
	}
	return out, err
}

// compiledPlanner 一次性规划器：直接返回编译好的计划（确定性，零模型）。
type compiledPlanner struct {
	plan         *squad.SquadPlan
	needsConfirm bool
	reason       string
}

func (p *compiledPlanner) Name() string { return "compiled" }

func (p *compiledPlanner) Plan(_ context.Context, _ squad.PlanInput) (*squad.PlanOutput, error) {
	steps := make([]squad.StepSpec, len(p.plan.Workflow.Steps))
	copy(steps, p.plan.Workflow.Steps)
	for i := range steps {
		steps[i].Status = "pending"
	}
	plan := &squad.SquadPlan{
		Squad:    p.plan.Squad,
		Template: p.plan.Template,
		Workflow: squad.WorkflowPlan{
			WorkflowID: p.plan.Workflow.WorkflowID, Title: p.plan.Workflow.Title,
			PlanningMode: p.plan.Workflow.PlanningMode, Status: "pending", Steps: steps,
		},
	}
	return &squad.PlanOutput{Plan: plan, NeedsConfirmation: p.needsConfirm, ConfirmReason: p.reason}, nil
}

// passChecker 编译图收口不做交付物质检（legacy 工作流没有收口质检门；
// 质量把关在 critic 层，由 wf.critic + Loop 承担）。
type passChecker struct{}

func (passChecker) Check(_ squad.CheckRequest) *squad.QualityReport {
	return &squad.QualityReport{CheckedAt: time.Now(), Score: 100, Passed: true}
}

// planSnapshot 把编译计划（只列 def 步骤，评审步不单列）映射为右栏快照。
func planSnapshot(compiled *Compiled, def core.WorkflowDef, live *squad.WorkflowPlan) core.PlanSnapshot {
	statusOf := map[string]string{}
	if live != nil {
		for _, s := range live.Steps {
			statusOf[s.ID] = mapStepStatus(s.Status)
		}
	}
	snap := core.PlanSnapshot{Title: def.Title, Steps: make([]core.PlanStep, len(def.Steps))}
	for i, step := range def.Steps {
		snap.Steps[i] = core.PlanStep{ID: step.ID, Text: step.Title,
			Status: orDefault(statusOf[step.ID], "pending")}
	}
	return snap
}

func mapStepStatus(s string) string {
	switch s {
	case "running":
		return "in_progress"
	case "completed":
		return "done"
	case "skipped":
		return "skipped"
	case "failed":
		return "failed"
	default:
		return "pending"
	}
}

// stepStatuses 每个 def 步骤的终态（legacy statuses 语义：done/failed/skipped/pending）。
func stepStatuses(compiled *Compiled, def core.WorkflowDef, live *squad.WorkflowPlan) []string {
	out := make([]string, len(def.Steps))
	if live == nil {
		return out
	}
	byID := map[string]string{}
	for _, s := range live.Steps {
		byID[s.ID] = s.Status
	}
	for i, step := range def.Steps {
		switch byID[step.ID] {
		case "completed":
			out[i] = "done"
		case "failed":
			out[i] = "failed"
		case "skipped":
			out[i] = "skipped"
		default:
			out[i] = "pending"
		}
	}
	return out
}

// buildOutcome legacy outcome() 同款汇总文案与部件。
func buildOutcome(def core.WorkflowDef, statuses []string, arts []*core.ArtifactRef, drops []string, stopped string) Outcome {
	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")
	}
	text := strings.TrimRight(b.String(), "\n")
	parts := []core.Part{{Type: core.PartText, Text: text}}
	for _, art := range arts {
		parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: art})
	}
	return Outcome{Summary: text, Parts: parts, Artifacts: arts}
}

// askHuman：HumanRequest → HITL 卡片 → RespondDecision（squadadapter 同款桥接）。
func askHuman(rc Host, req *squad.HumanRequest) (squad.RespondDecision, error) {
	choices := make([]core.HITLChoice, 0, len(req.Options))
	for _, o := range req.Options {
		choices = append(choices, core.HITLChoice{
			ID: o.Value, Label: o.Label,
			Description: firstNonEmpty(req.Context, "工作流挂起"),
			Impact:      fmt.Sprintf("决策类型 %s：%s 将继续/终止本次执行", req.Kind, o.Value),
		})
	}
	if len(choices) == 0 {
		choices = append(choices,
			core.HITLChoice{ID: "approve", Label: "继续", Description: "按当前计划继续执行", Impact: "工作流继续执行后续步骤", Recommended: true},
			core.HITLChoice{ID: "abort", Label: "中止", Description: "终止执行", Impact: "执行取消，已产出保留"},
		)
	}
	resp, err := rc.AskHITL(core.HITLCard{
		Question:       req.Question,
		Kind:           core.HITLKindChoice,
		Choices:        choices,
		AllowFreeInput: true,
		FreeInputHint:  "可直接输入回答或修改指令",
	})
	if err != nil {
		return squad.RespondDecision{RequestID: req.ID, Choice: "abort"}, err
	}
	d := squad.RespondDecision{RequestID: req.ID}
	if resp.ChoiceID != "" {
		d.Choice = resp.ChoiceID
	} else {
		d.Answer = resp.FreeText
	}
	return d, nil
}

func isHoldStatus(s string) bool {
	return s == "waiting_human" || s == "awaiting_confirmation" || s == "awaiting_clarification" || s == "quality_review"
}

func orDefault(v, def string) string {
	if strings.TrimSpace(v) != "" {
		return v
	}
	return def
}

func firstNonEmpty(vals ...string) string {
	for _, v := range vals {
		if strings.TrimSpace(v) != "" {
			return v
		}
	}
	return ""
}