Cyrene_v2/internal/capability/skills/steps.go

package skills

import (
	"context"
	"fmt"
	"log/slog"
	"sort"
	"strings"

	"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
)

// 步骤调度器：把 design/09 §6.4-1/2/4/5 变成可测的行为。
//
//   - 按 depends 拓扑调度；无依赖（含依赖已就绪）的步骤可并行；
//   - 并行上限默认 2，硬上限 3（与 design/15 §2.5 的口径一致）；
//   - 每步输出后校验 guardrail：enforced 违反 ⇒ 中止并如实报「哪条、哪一步」，
//     非 enforced ⇒ WARN 后继续；
//   - 全部步骤完成后再做一次**累计输出的全量复核**（design/09 §6.4-4 的「末步
//     前再全量复核一次」：并行调度下没有唯一的「末步」，因此落在全部步骤产出之后、
//     返回之前，复核对象是累计输出）；
//   - 任一步失败 ⇒ 技能失败（不静默跳过）；父 ctx 取消 ⇒ 立即取消收敛。

const (
	// defaultStepParallel 是未指定时的步骤并行上限。
	defaultStepParallel = 2
	// maxStepParallel 是并行上限的硬顶：技能调用方给再大的值也夹到这里。
	maxStepParallel = 3
	// maxStepOutputChars 是单步输出传给下游步骤时的字符上限。步骤输出按有界摘要
	// 传递（design/09 §6.4-2），避免一步的长输出把后续步骤的上下文撑爆。
	maxStepOutputChars = 4000
	// stepTruncatedMarker 是步骤输出被预算截断时的显式标记（绝不静默截断）。
	stepTruncatedMarker = "…(truncated)"
)

// clampParallel 把并行上限夹进 [1, maxStepParallel]：非正取默认 2，超硬顶夹到 3。
func clampParallel(n int) int {
	if n <= 0 {
		return defaultStepParallel
	}
	if n > maxStepParallel {
		return maxStepParallel
	}
	return n
}

// runEnv 是一次技能执行的全部状态（每次 RunSkill 新建一个，因此没有跨执行的共享
// 可变状态）。只有调度器 goroutine 会读写它，步骤 goroutine 只读入参、经 channel 回传。
type runEnv struct {
	skill    sdk.Skill
	steps    []sdk.Step
	spec     Spec
	tools    []string
	input    string
	exec     StepExecutor
	checker  GuardrailChecker
	parallel int
	log      *slog.Logger
}

// stepDone 是步骤 goroutine 回传给调度器的结果。
type stepDone struct {
	index  int
	output string
	err    error
}

// stepGraph 是步骤依赖图：deps[i] 是步骤 i 依赖的索引，dependents 是反向边，
// remaining 是每个步骤还没完成的前序数，ready 是初始就绪队列（按声明序）。
type stepGraph struct {
	deps       [][]int
	dependents [][]int
	remaining  []int
	ready      []int
}

// buildStepGraph 从已校验的步骤建图。名字重复、依赖缺失都会在此二次报错（Validate
// 已经拦过一次，这里是执行路径自己的保底）。
func buildStepGraph(steps []sdk.Step) (stepGraph, error) {
	size := len(steps)
	index := make(map[string]int, size)
	for i, st := range steps {
		index[strings.TrimSpace(st.Name)] = i
	}
	g := stepGraph{
		deps:       make([][]int, size),
		dependents: make([][]int, size),
		remaining:  make([]int, size),
	}
	for i, st := range steps {
		for _, dep := range st.Depends {
			depIdx, ok := index[strings.TrimSpace(dep)]
			if !ok {
				return g, fmt.Errorf("skills: step %q depends on unknown step %q: %w",
					st.Name, dep, ErrStepDependsMissing)
			}
			g.deps[i] = append(g.deps[i], depIdx)
			g.dependents[depIdx] = append(g.dependents[depIdx], i)
			g.remaining[i]++
		}
	}
	for i := range steps {
		if g.remaining[i] == 0 {
			g.ready = append(g.ready, i)
		}
	}
	return g, nil
}

// run 执行整个技能。base 里已带上技能名与工具面，本函数补上步骤结果与违规清单。
func (e *runEnv) run(ctx context.Context, base RunResult) (RunResult, error) {
	e.steps = e.skill.Steps()
	if len(e.steps) == 0 {
		return e.result(base, nil, nil, nil), nil
	}
	graph, err := buildStepGraph(e.steps)
	if err != nil {
		return e.result(base, nil, nil, nil), fmt.Errorf("skills: runner: run %q: %w", base.Skill, err)
	}
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	outcomes := make([]StepOutcome, len(e.steps))
	outputs := make([]string, len(e.steps))
	finished := make([]bool, len(e.steps))
	doneCh := make(chan stepDone, e.parallel)
	violations := make([]Violation, 0)
	ready := append([]int(nil), graph.ready...)
	running, completed := 0, 0

	for completed < len(e.steps) {
		if cerr := ctx.Err(); cerr != nil {
			e.drain(doneCh, running)
			return e.result(base, outcomes, finished, violations), e.wrapCanceled(base.Skill, cerr)
		}
		for running < e.parallel && len(ready) > 0 {
			idx := ready[0]
			ready = ready[1:]
			running++
			go e.execStep(ctx, idx, outputs, graph.deps[idx], doneCh)
		}
		if running == 0 {
			// 已经没有在飞步骤、也没有就绪步骤，却仍有步骤没完成 ⇒ 依赖成环。
			// Validate 已拦过一次，这里兜底返回而不是死等（运行期绝不挂死）。
			return e.result(base, outcomes, finished, violations),
				fmt.Errorf("skills: runner: run %q: %w", base.Skill, ErrStepCycle)
		}
		select {
		case <-ctx.Done():
			e.drain(doneCh, running)
			return e.result(base, outcomes, finished, violations), e.wrapCanceled(base.Skill, ctx.Err())
		case res := <-doneCh:
			running--
			if res.err != nil {
				// 先取消（把信号发给在飞步骤），再收它们 —— 反过来会一直等到
				// 各步自然结束，等于把失败当成了「等一下就好」。
				cancel()
				e.drain(doneCh, running)
				return e.result(base, outcomes, finished, violations), res.err
			}
			outcomes[res.index] = StepOutcome{Name: e.steps[res.index].Name, Output: res.output}
			outputs[res.index] = res.output
			finished[res.index] = true
			completed++
			found, verr := e.checkStepGuardrails(ctx, e.steps[res.index].Name, res.output)
			violations = append(violations, found...)
			if verr != nil {
				cancel()
				e.drain(doneCh, running)
				return e.result(base, outcomes, finished, violations), verr
			}
			e.release(graph, res.index, &ready)
		}
	}
	found, verr := e.finalRecheck(ctx, outputs)
	violations = append(violations, found...)
	result := e.result(base, outcomes, finished, violations)
	if verr != nil {
		return result, verr
	}
	e.log.Info("skill run completed",
		"skill", base.Skill, "steps", completed, "tools", len(base.Tools), "violations", len(violations))
	return result, nil
}

// execStep 跑一步并把结果回传。它由调度器起在独立 goroutine 里，只读共享输入。
func (e *runEnv) execStep(ctx context.Context, index int, outputs []string, deps []int, done chan<- stepDone) {
	step := e.steps[index]
	if err := ctx.Err(); err != nil {
		done <- stepDone{index: index, err: e.stepError(step.Name, err)}
		return
	}
	req := StepRequest{
		Skill:   e.skill.Name(),
		Step:    step,
		Prompt:  joinPrompt(e.skill.Prompt(), step.Prompt),
		Depends: e.depOutputs(deps, outputs),
		Input:   e.input,
		Tools:   append([]string(nil), e.tools...),
		Refs:    e.skill.Refs(),
		Spec:    e.spec,
	}
	res, err := e.exec.ExecuteStep(ctx, req)
	if err != nil {
		done <- stepDone{index: index, err: e.stepError(step.Name, err)}
		return
	}
	done <- stepDone{index: index, output: res.Output}
}

// depOutputs 取本步依赖的各前序步骤输出（有界摘要），键为步骤名。
func (e *runEnv) depOutputs(deps []int, outputs []string) map[string]string {
	if len(deps) == 0 {
		return nil
	}
	out := make(map[string]string, len(deps))
	for _, dep := range deps {
		out[e.steps[dep].Name] = truncateOutput(outputs[dep], maxStepOutputChars)
	}
	return out
}

// release 把 step 完成的事实传播给它的后继：前序计数归零的后继进入就绪队列。
func (e *runEnv) release(graph stepGraph, step int, ready *[]int) {
	for _, dependent := range graph.dependents[step] {
		graph.remaining[dependent]--
		if graph.remaining[dependent] == 0 {
			*ready = append(*ready, dependent)
		}
	}
}

// checkStepGuardrails 在某步输出后校验全部 guardrail。返回发现的违规（非 enforced
// 的会一并记下），以及**必须中止**时的错误（enforced 违规、或校验器本身失败）。
func (e *runEnv) checkStepGuardrails(ctx context.Context, stepName, output string) ([]Violation, error) {
	if e.checker == nil {
		return nil, nil
	}
	found := make([]Violation, 0, len(e.skill.Guardrails()))
	for _, g := range e.skill.Guardrails() {
		violated, err := e.checker.Check(ctx, g, stepName, output)
		if err != nil {
			return found, fmt.Errorf("skills: runner: run %q: guardrail %q at step %q: %w",
				e.skill.Name(), g.ID, stepName, fmt.Errorf("%w: %w", ErrGuardrailCheck, err))
		}
		if !violated {
			continue
		}
		v := Violation{GuardrailID: g.ID, Subject: stepName, Message: g.Message, Enforced: g.Enforced}
		found = append(found, v)
		if g.Enforced {
			return found, e.violationError(v)
		}
		e.log.Warn("skill guardrail violated (warn only)",
			"skill", e.skill.Name(), "guardrail", g.ID, "step", stepName, "message", g.Message)
	}
	return found, nil
}

// finalRecheck 对**全部步骤的累计输出**做一次全量复核（design/09 §6.4-4）。
func (e *runEnv) finalRecheck(ctx context.Context, outputs []string) ([]Violation, error) {
	guardrails := e.skill.Guardrails()
	if e.checker == nil || len(guardrails) == 0 {
		return nil, nil
	}
	joined := truncateOutput(strings.Join(nonEmptyOutputs(outputs, e.steps), "\n\n"), maxStepOutputChars*len(guardrails))
	found := make([]Violation, 0, len(guardrails))
	for _, g := range guardrails {
		violated, err := e.checker.Check(ctx, g, GuardrailFinalSubject, joined)
		if err != nil {
			return found, fmt.Errorf("skills: runner: run %q: guardrail %q at %s: %w",
				e.skill.Name(), g.ID, GuardrailFinalSubject, fmt.Errorf("%w: %w", ErrGuardrailCheck, err))
		}
		if !violated {
			continue
		}
		v := Violation{GuardrailID: g.ID, Subject: GuardrailFinalSubject, Message: g.Message, Enforced: g.Enforced}
		found = append(found, v)
		if g.Enforced {
			return found, e.violationError(v)
		}
		e.log.Warn("skill guardrail violated on final review (warn only)",
			"skill", e.skill.Name(), "guardrail", g.ID, "message", g.Message)
	}
	return found, nil
}

// nonEmptyOutputs 取已产出步骤的输出（按声明序），跳过后面的空输出。
func nonEmptyOutputs(outputs []string, steps []sdk.Step) []string {
	out := make([]string, 0, len(outputs))
	for i, text := range outputs {
		if text == "" || i >= len(steps) {
			continue
		}
		out = append(out, text)
	}
	return out
}

// result 组装如实的结果：只汇报**已完成**的步骤（按声明序），违规清单排序后输出，
// 因此同样的执行路径总是得到同样的汇报形状。
func (e *runEnv) result(base RunResult, outcomes []StepOutcome, finished []bool, violations []Violation) RunResult {
	out := base
	out.Steps = make([]StepOutcome, 0, len(outcomes))
	for i, oc := range outcomes {
		if finished != nil && i < len(finished) && !finished[i] {
			continue
		}
		out.Steps = append(out.Steps, oc)
	}
	out.Violations = sortViolations(violations)
	return out
}

// sortViolations 按（对象, guardrail id）排序，让并行执行下的汇报顺序确定。
func sortViolations(in []Violation) []Violation {
	out := append([]Violation(nil), in...)
	sort.Slice(out, func(i, j int) bool {
		if out[i].Subject != out[j].Subject {
			return out[i].Subject < out[j].Subject
		}
		return out[i].GuardrailID < out[j].GuardrailID
	})
	return out
}

// drain 等在飞的步骤回传，避免留下无主 goroutine。channel 的缓冲等于并行上限，
// 因此回传不会阻塞；调用方必须保证 exec 尊重 ctx（StepExecutor 的契约）。
func (e *runEnv) drain(done chan stepDone, running int) {
	for i := 0; i < running; i++ {
		<-done
	}
}

// violationError 把一条 enforced 违规变成可分类错误，并**如实带上哪条 guardrail、
// 在哪一步**（design/09 §6.4-4 的要求）。
func (e *runEnv) violationError(v Violation) error {
	return fmt.Errorf("skills: runner: run %q: guardrail %q violated at %s: %s: %w",
		e.skill.Name(), v.GuardrailID, v.Subject, v.Message, ErrGuardrailViolation)
}

// stepError 包装步骤失败：带步骤名，并用 %w 串上 ErrStepFailed 让调用方能分类。
func (e *runEnv) stepError(stepName string, err error) error {
	return fmt.Errorf("skills: runner: run %q: step %q: %w: %w",
		e.skill.Name(), stepName, ErrStepFailed, err)
}

// wrapCanceled 把父 ctx 的取消如实报出来（errors.Is 能命中 context.Canceled /
// context.DeadlineExceeded）。
func (e *runEnv) wrapCanceled(skillName string, err error) error {
	return fmt.Errorf("skills: runner: run %q: %w", skillName, err)
}

// joinPrompt 拼接技能基础提示词与本步提示词；任一段为空时不留孤立空行。
func joinPrompt(base, step string) string {
	parts := make([]string, 0, 2)
	if trimmed := strings.TrimSpace(base); trimmed != "" {
		parts = append(parts, trimmed)
	}
	if trimmed := strings.TrimSpace(step); trimmed != "" {
		parts = append(parts, trimmed)
	}
	return strings.Join(parts, "\n\n")
}

// truncateOutput 把步骤输出截到 n 个字符以内（按 rune 计，不切断多字节字符），
// 超长时追加显式标记 —— 绝不静默截断。
func truncateOutput(s string, n int) string {
	if n <= 0 {
		return s
	}
	runes := []rune(s)
	if len(runes) <= n {
		return s
	}
	return string(runes[:n]) + stepTruncatedMarker
}