Agent-Squad/squad/graph.go

package squad

import (
	"fmt"
	"hash/fnv"
	"strings"
)

// graph 是一次执行图：由权威 Steps 现场编译（Steps 仍是唯一真相源，
// 动态调整改 Steps 后下一轮自然生效）。只携带调度所需的派生信息：
// 依赖、写集与可执行分层。
type graph struct {
	needs  map[string][]string // step id → 依赖 id（含相邻 Parallel 兼容映射产生的）
	order  []string            // 声明序
	writes map[string][]string // step id → 写集
}

// compileGraph 编译步骤图为调度图。规则：
//   - 声明 Needs 的步骤：依赖即其 Needs（未知引用/环 fail-closed）。
//   - 未声明 Needs 的步骤：保持现状语义——相邻 Parallel=true 的步骤同层
//     （依赖 = 上一个「非本组」步骤），非 Parallel 步骤依赖前一个步骤。
//   - 混合图允许：Needs 提供剪枝，未声明处保持线性。
func compileGraph(steps []StepSpec) (*graph, error) {
	g := &graph{
		needs:  map[string][]string{},
		order:  make([]string, 0, len(steps)),
		writes: map[string][]string{},
	}
	ids := map[string]bool{}
	for _, s := range steps {
		if s.ID == "" {
			return nil, fmt.Errorf("graph: 存在空 id 的步骤")
		}
		if ids[s.ID] {
			return nil, fmt.Errorf("graph: 步骤 id 重复：%s", s.ID)
		}
		ids[s.ID] = true
	}
	prevNonGroup := -1 // 上一个「非本 Parallel 组」的步骤索引
	for i, s := range steps {
		g.order = append(g.order, s.ID)
		g.writes[s.ID] = append([]string(nil), s.Writes...)
		needsDecl := len(s.Needs) > 0
		switch {
		case needsDecl:
			for _, n := range s.Needs {
				if !ids[n] {
					return nil, fmt.Errorf("graph: 步骤 %s 的 needs 引用未知步骤 %s", s.ID, n)
				}
				if n == s.ID {
					return nil, fmt.Errorf("graph: 步骤 %s 的 needs 引用自身", s.ID)
				}
			}
			g.needs[s.ID] = append([]string(nil), s.Needs...)
			prevNonGroup = i
		case s.Parallel:
			// 相邻 Parallel 组：依赖上一个组外步骤（没有则无依赖）。
			if prevNonGroup >= 0 {
				g.needs[s.ID] = []string{steps[prevNonGroup].ID}
			}
			// Parallel 步骤不推进 prevNonGroup：相邻同伴同层。
		default:
			// 非 Parallel：依赖紧邻的前一步（与旧线性扫描的执行序一致——
			// 整组 Parallel 完成后才轮到它，而不是只等组外前一步）。
			if i > 0 {
				g.needs[s.ID] = []string{steps[i-1].ID}
			}
			prevNonGroup = i
		}
	}
	if cycle := g.findCycle(); cycle != nil {
		return nil, fmt.Errorf("graph: needs 依赖成环：%s", strings.Join(cycle, " → "))
	}
	// 循环步与路由步校验（Q2）：预算必须显式，引用必须存在，两者互斥
	//（同一步既循环又分支的语义未定义，fail-closed）。
	for _, s := range steps {
		if s.Loop != nil && len(s.Branches) > 0 {
			return nil, fmt.Errorf("graph: 步骤 %s 不得同时声明 loop 与 branches", s.ID)
		}
		if s.Loop != nil {
			if s.Loop.MaxIterations <= 0 {
				return nil, fmt.Errorf("graph: 步骤 %s 的 loop.max_iterations 必须 > 0", s.ID)
			}
			for _, b := range s.Loop.Body {
				if !ids[b] {
					return nil, fmt.Errorf("graph: 步骤 %s 的 loop.body 引用未知步骤 %s", s.ID, b)
				}
				if b == s.ID {
					return nil, fmt.Errorf("graph: 步骤 %s 的 loop.body 引用自身", s.ID)
				}
			}
		}
		for _, br := range s.Branches {
			if br.When.Field == "" {
				return nil, fmt.Errorf("graph: 步骤 %s 的分支缺少 when.field", s.ID)
			}
			if br.When.Equals == "" && br.When.NotEquals == "" {
				return nil, fmt.Errorf("graph: 步骤 %s 的分支 when 需要 equals 或 not_equals", s.ID)
			}
			if len(br.Then) == 0 {
				return nil, fmt.Errorf("graph: 步骤 %s 的分支 then 为空", s.ID)
			}
			for _, t := range br.Then {
				if !ids[t] {
					return nil, fmt.Errorf("graph: 步骤 %s 的分支 then 引用未知步骤 %s", s.ID, t)
				}
				if t == s.ID {
					return nil, fmt.Errorf("graph: 步骤 %s 的分支 then 引用自身", s.ID)
				}
			}
		}
	}
	return g, nil
}

// findCycle 报告一条依赖环（没有则 nil）。
func (g *graph) findCycle() []string {
	const (
		white = iota
		gray
		black
	)
	color := map[string]int{}
	var stack []string
	var visit func(id string) []string
	visit = func(id string) []string {
		if color[id] == gray {
			return append(stack, id)
		}
		if color[id] == black {
			return nil
		}
		color[id] = gray
		stack = append(stack, id)
		for _, n := range g.needs[id] {
			if c := visit(n); c != nil {
				return c
			}
		}
		stack = stack[:len(stack)-1]
		color[id] = black
		return nil
	}
	for _, id := range g.order {
		if c := visit(id); c != nil {
			return c
		}
	}
	return nil
}

// layer 是一个 superstep 的可执行集合：seq 为有写集冲突需按声明序串行的
// 步骤索引，par 为可并行的步骤索引。先执行 par 组（runParallel 语义），
// 再逐个执行 seq（含动态调整），与现状「并行组先于后续顺序步骤」一致。
type layer struct {
	par []int
	seq []int
}

// stepPending 报告步骤是否仍待执行。蓝图登记的步骤不带 status（空串），
// 旧线性扫描将其视为可执行——这里与之一致：空串等同 pending。
func stepPending(status string) bool {
	return status == "" || status == "pending" || status == "running"
}

// nextLayer 计算当前可执行层：状态为 pending/running 且全部依赖已「满足」。
// 依赖满足 = completed/skipped/failed——failed 视为满足与现状「非关键失败
// 降级继续」的线性扫描一致（关键失败时运行已终止，不会走到这里）。
// 无可执行步骤时返回 (layer{}, false)：可能全部完成，也可能调度卡死
// （由调用方用 pendingCount 区分）。
func (g *graph) nextLayer(steps []StepSpec) (layer, bool) {
	idxOf := map[string]int{}
	for i, s := range steps {
		idxOf[s.ID] = i
	}
	satisfied := func(id string) bool {
		st := steps[idxOf[id]].Status
		return st == "completed" || st == "skipped" || st == "failed"
	}
	var out layer
	for i, s := range steps {
		if !stepPending(s.Status) {
			continue
		}
		ready := true
		for _, n := range g.needs[s.ID] {
			if !satisfied(n) {
				ready = false
				break
			}
		}
		if !ready {
			continue
		}
		if g.conflictsWithEarlier(out, i, steps) {
			out.seq = append(out.seq, i)
		} else {
			out.par = append(out.par, i)
		}
	}
	return out, len(out.par)+len(out.seq) > 0
}

// conflictsWithEarlier 报告步骤 i 的写集与本层中更早的可执行步骤相交。
func (g *graph) conflictsWithEarlier(l layer, i int, steps []StepSpec) bool {
	wi := g.writes[steps[i].ID]
	if len(wi) == 0 {
		return false
	}
	earlier := append(append([]int(nil), l.par...), l.seq...)
	for _, j := range earlier {
		for _, w := range g.writes[steps[j].ID] {
			for _, v := range wi {
				if w == v {
					return true
				}
			}
		}
	}
	return false
}

// pendingCount 统计仍未终结（pending/running/空串）的步骤数。
func pendingCount(steps []StepSpec) int {
	n := 0
	for _, s := range steps {
		if stepPending(s.Status) {
			n++
		}
	}
	return n
}

// hash 是图结构指纹：声明序、依赖与写集的稳定散列。
// 恢复连续性校验用它比对崩溃前后的图形态（Q1 WP4；WP1 的图版本裁定落点）。
func (g *graph) hash() string {
	h := fnv.New64a()
	for _, id := range g.order {
		h.Write([]byte(id))
		h.Write([]byte{0})
		for _, n := range g.needs[id] {
			h.Write([]byte(">"))
			h.Write([]byte(n))
		}
		h.Write([]byte{1})
		for _, w := range g.writes[id] {
			h.Write([]byte("w"))
			h.Write([]byte(w))
		}
		h.Write([]byte{2})
	}
	return fmt.Sprintf("%016x", h.Sum64())
}