AskaEth
f75e7a9c9b
- 加载器/校验：目录式 configs/skills/<id>/skill.yaml（兼容旧单文件 <id>.yaml，先目录后文件）； 新字段 version/when-to-use/invocation(三态)/triggers/paths/allowed-tools/disallowed-tools/ context(ine|fork)/agent/model/effort/metadata；**依赖环 ⇒ 加载期拒**（运行期绝不挂死）； depends 指向不存在步骤 ⇒ 拒；同一工具同时在 allowed 与 disallowed ⇒ 拒；错误可分类（哨兵 + %w） - Runner.Run 真执行（替换 ErrRunPending）：步骤拓扑（无依赖可并行、上限 2）+ guardrail 强制点 （enforced 违反 ⇒ 中止并如实报「哪条、哪一步」；否则 WARN 继续）+ 失败不静默跳过 + ctx 取消传播 - 🔴 能力收缩只收不放：该轮可用工具 = allowed-tools ∩ 安全闸 ∩ 注册表，再减 disallowed-tools - 另有渐进披露相关的 listing/surface/activation（清单投影与激活判定） ⚠️ 未接线：本片是库与执行器，尚未把技能清单接进模型面（见 report 的分阶段）。
378 lines
13 KiB
Go
378 lines
13 KiB
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
|
||
}
|