From f90abafe7908aff93982b209b5b948db5826b0ef Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 7 Oct 2026 09:06:52 +0000 Subject: [PATCH] Drive workflow execution as supersteps over a compiled dependency graph Why: Q1 WP1/WP2 - replace the linear step scan with per-round graph compilation and layer advancement so Needs drives scheduling and write-conflicting steps never run concurrently. Evidence: go vet clean in all five modules; go test ok everywhere, including the full compatibility net (engine, leader, recovery, r2 regression, intent, evaluator suites) and new graph tests. Impact: new squad/graph.go (compile/layer/conflict); StepSpec gains Writes; advance advances by superstep (parallel run of conflict-free members, serial for write-conflicts, single-member groups keep dynamic adjustment). Legacy semantics preserved: adjacent-Parallel mapping, degraded-continue on non-critical failure (failed counts as satisfied), empty status = pending, cycle/unknown/dup needs fail-closed. Refs: Documents proposal 2026-10-07-agent-stack-n1q1i1-plans.md Q1 WP1/WP2 --- squad/graph.go | 192 ++++++++++++++++++++++++++++++++++++++++++++ squad/graph_test.go | 139 ++++++++++++++++++++++++++++++++ squad/models.go | 3 + squad/workflow.go | 124 ++++++++++++++++------------ 4 files changed, 406 insertions(+), 52 deletions(-) create mode 100644 squad/graph.go create mode 100644 squad/graph_test.go diff --git a/squad/graph.go b/squad/graph.go new file mode 100644 index 0000000..66fac0a --- /dev/null +++ b/squad/graph.go @@ -0,0 +1,192 @@ +package squad + +import ( + "fmt" + "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, " → ")) + } + 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 +} diff --git a/squad/graph_test.go b/squad/graph_test.go new file mode 100644 index 0000000..120a0be --- /dev/null +++ b/squad/graph_test.go @@ -0,0 +1,139 @@ +package squad + +import ( + "strings" + "testing" +) + +func stepsOf(specs ...StepSpec) []StepSpec { + out := make([]StepSpec, len(specs)) + copy(out, specs) + return out +} + +func TestCompileGraphLinearAndAdjacentParallel(t *testing.T) { + steps := stepsOf( + StepSpec{ID: "a", Status: "pending"}, + StepSpec{ID: "b", Status: "pending", Parallel: true}, + StepSpec{ID: "c", Status: "pending", Parallel: true}, + StepSpec{ID: "d", Status: "pending"}, + ) + g, err := compileGraph(steps) + if err != nil { + t.Fatal(err) + } + // 线性链：b/c 同层依赖 a，d 依赖 c。 + if len(g.needs["a"]) != 0 { + t.Fatalf("a needs = %v", g.needs["a"]) + } + if len(g.needs["b"]) != 1 || g.needs["b"][0] != "a" || g.needs["c"][0] != "a" { + t.Fatalf("parallel needs = %v %v", g.needs["b"], g.needs["c"]) + } + if len(g.needs["d"]) != 1 || g.needs["d"][0] != "c" { + t.Fatalf("d needs = %v", g.needs["d"]) + } + // 首轮可执行：只有 a。 + layer, ok := g.nextLayer(steps) + if !ok || len(layer.par) != 1 || layer.par[0] != 0 || len(layer.seq) != 0 { + t.Fatalf("first layer = %+v", layer) + } + // a 完成后：b、c 同层并行。 + steps[0].Status = "completed" + layer, ok = g.nextLayer(steps) + if !ok || len(layer.par) != 2 || layer.par[0] != 1 || layer.par[1] != 2 { + t.Fatalf("second layer = %+v", layer) + } +} + +func TestCompileGraphNeedsPruneAndMixed(t *testing.T) { + steps := stepsOf( + StepSpec{ID: "a", Status: "pending"}, + StepSpec{ID: "b", Status: "pending"}, + StepSpec{ID: "c", Status: "pending", Needs: []string{"a"}}, + ) + g, err := compileGraph(steps) + if err != nil { + t.Fatal(err) + } + // a 完成后：b（线性依赖 a）与 c（needs a）同层。 + steps[0].Status = "completed" + layer, ok := g.nextLayer(steps) + if !ok || len(layer.par) != 2 { + t.Fatalf("layer = %+v", layer) + } + // b 失败（非关键降级）后 c 仍按 needs 调度（与线性扫描的降级继续一致）。 + steps[1].Status = "failed" + layer, ok = g.nextLayer(steps) + if !ok || len(layer.par)+len(layer.seq) == 0 { + t.Fatalf("after b failed: layer = %+v", layer) + } +} + +func TestCompileGraphRejectsUnknownAndCycleAndDup(t *testing.T) { + if _, err := compileGraph(stepsOf( + StepSpec{ID: "a", Status: "pending", Needs: []string{"ghost"}}, + )); err == nil || !strings.Contains(err.Error(), "未知步骤") { + t.Fatalf("unknown ref: %v", err) + } + if _, err := compileGraph(stepsOf( + StepSpec{ID: "a", Status: "pending", Needs: []string{"b"}}, + StepSpec{ID: "b", Status: "pending", Needs: []string{"a"}}, + )); err == nil || !strings.Contains(err.Error(), "成环") { + t.Fatalf("cycle: %v", err) + } + if _, err := compileGraph(stepsOf( + StepSpec{ID: "a", Status: "pending"}, + StepSpec{ID: "a", Status: "pending"}, + )); err == nil || !strings.Contains(err.Error(), "重复") { + t.Fatalf("dup: %v", err) + } + if _, err := compileGraph(stepsOf( + StepSpec{ID: "a", Status: "pending", Needs: []string{"a"}}, + )); err == nil || !strings.Contains(err.Error(), "自身") { + t.Fatalf("self ref: %v", err) + } +} + +func TestWriteSetConflictSerializes(t *testing.T) { + steps := stepsOf( + StepSpec{ID: "z", Status: "completed"}, + StepSpec{ID: "a", Status: "pending", Parallel: true, Writes: []string{"report.md"}}, + StepSpec{ID: "b", Status: "pending", Parallel: true, Writes: []string{"report.md"}}, + StepSpec{ID: "c", Status: "pending", Parallel: true, Writes: []string{"notes.md"}}, + StepSpec{ID: "d", Status: "pending", Parallel: true}, + ) + // 全部首轮可执行（无依赖）：b 与 a 冲突 → b 串行；c、d 无冲突 → 并行。 + g, err := compileGraph(steps) + if err != nil { + t.Fatal(err) + } + layer, ok := g.nextLayer(steps) + if !ok { + t.Fatal("no layer") + } + if len(layer.seq) != 1 || layer.seq[0] != 2 { + t.Fatalf("seq = %v", layer.seq) + } + if len(layer.par) != 3 || layer.par[0] != 1 || layer.par[1] != 3 || layer.par[2] != 4 { + t.Fatalf("par = %v", layer.par) + } +} + +func TestFailedDependencyStillSchedules(t *testing.T) { + // 与线性扫描的「非关键失败降级继续」一致：依赖 failed 仍视为满足。 + steps := stepsOf( + StepSpec{ID: "b", Status: "failed"}, + StepSpec{ID: "a", Status: "pending", Needs: []string{"b"}}, + ) + g, err := compileGraph(steps) + if err != nil { + t.Fatal(err) + } + layer, ok := g.nextLayer(steps) + if !ok || len(layer.par) != 1 || layer.par[0] != 1 { + t.Fatalf("failed dependency should still schedule: %+v", layer) + } + if pendingCount(steps) != 1 { + t.Fatalf("pending = %d", pendingCount(steps)) + } +} diff --git a/squad/models.go b/squad/models.go index 289a83f..dac1795 100644 --- a/squad/models.go +++ b/squad/models.go @@ -88,6 +88,9 @@ type StepSpec struct { Mode string `json:"mode,omitempty"` // 预留 Parallel bool `json:"parallel"` Needs []string `json:"needs,omitempty"` + // Writes 写集：本步骤会写入的工件相对路径（调度依据）。 + // 同一 superstep 内写集相交的步骤强制串行，不相交才可并行（Q1 WP1）。 + Writes []string `json:"writes,omitempty"` Targets []string `json:"targets,omitempty"` // 期望交付物相对路径，缺省 output.md Status string `json:"status"` // pending, running, completed, failed, skipped // Attempts 该步骤进入 running 的次数。中断恢复用它决定 auto 策略是否还能重试。 diff --git a/squad/workflow.go b/squad/workflow.go index d718174..25582ca 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -19,80 +19,100 @@ type flowDriver struct { } // advance 执行到边界；返回 (结果, 挂起上下文)，二者互斥非 nil（infra 错误时都为 nil 且 err 非 nil）。 +// 推进模型（Q1 WP1/WP2）：每轮从权威 Steps 现场编译调度图，按 superstep +// 分层推进——层内无写集冲突的步骤并行（runParallel 复用），有冲突的按声明序 +// 串行（含动态调整）；动态调整改 Steps 后下一轮自然生效。 func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult, gate *hitlGate, err error) { wf := st.Workflow() wf.Status = "in_progress" st.setWorkflow(wf) d.emit(st, Event{Type: EvPhase, Text: "开始执行工作流"}) - i := 0 - for i < len(st.Plan.Workflow.Steps) { - step := st.Plan.Workflow.Steps[i] - switch step.Status { - case "completed", "skipped", "failed": - i++ - continue + superNo := 0 + for { + g, gerr := compileGraph(st.Plan.Workflow.Steps) + if gerr != nil { + wf2 := st.Workflow() + wf2.Status = "failed" + st.setWorkflow(wf2) + return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), + Error: "graph: " + gerr.Error(), WorkDir: st.Meta.WorkDir}, nil, nil } - - if out, g, err := d.gateBeforeRun(ctx, st, i); out != nil || g != nil || err != nil { - if out == nil && g != nil { - out = heldResult(st, g) + layer, ok := g.nextLayer(st.Plan.Workflow.Steps) + if !ok { + if n := pendingCount(st.Plan.Workflow.Steps); n > 0 { + wf2 := st.Workflow() + wf2.Status = "failed" + st.setWorkflow(wf2) + return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), + Error: fmt.Sprintf("graph: 依赖无法满足，仍有 %d 个步骤无法调度", n), WorkDir: st.Meta.WorkDir}, nil, nil } - return out, g, err + break } + superNo++ + d.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("superstep %d 推进（并行 %d 步，串行 %d 步）", superNo, len(layer.par), len(layer.seq))}) - if step.Parallel { - // 收集相邻并行组。组内每一步都先过审批和权限，任一需要挂起则整组不执行。 - var group []int - for i < len(st.Plan.Workflow.Steps) { - s := st.Plan.Workflow.Steps[i] - if s.Parallel && s.Status != "completed" && s.Status != "skipped" && s.Status != "failed" { - if out, g, err := d.gateBeforeRun(ctx, st, i); out != nil || g != nil || err != nil { - if out == nil && g != nil { - out = heldResult(st, g) - } - return out, g, err + // 并行段：组内每一步都先过审批和权限，任一需要挂起则整组不执行。 + if len(layer.par) > 0 { + for _, idx := range layer.par { + if out, g2, err := d.gateBeforeRun(ctx, st, idx); out != nil || g2 != nil || err != nil { + if out == nil && g2 != nil { + out = heldResult(st, g2) } - group = append(group, i) - i++ - continue + return out, g2, err } - break } - if out, g, err := d.runParallel(ctx, st, group); out != nil || g != nil || err != nil { - if out == nil && g != nil { - out = heldResult(st, g) + if out, g2, err := d.runParallel(ctx, st, layer.par); out != nil || g2 != nil || err != nil { + if out == nil && g2 != nil { + out = heldResult(st, g2) } - return out, g, err + return out, g2, err } if g2 := d.checkBlockingGaps(st); g2 != nil { return heldResult(st, g2), g2, nil } - continue + // 单成员并行组实际是顺序执行：与旧线性扫描一致，其后可动态调整 + // （多成员组不做，避免并发竞态）。 + if len(layer.par) == 1 { + idx := layer.par[0] + if st.Plan.Workflow.Steps[idx].AllowDynamicAdjustment && d.opts.Evaluator != nil { + out, g2, err := d.applyDynamicAdjustment(ctx, st, idx) + if out != nil || g2 != nil || err != nil { + if out == nil && g2 != nil { + out = heldResult(st, g2) + } + return out, g2, err + } + } + } } - out, g, err := d.runOneStep(ctx, st, i) - if out != nil || g != nil || err != nil { - if out == nil && g != nil { - out = heldResult(st, g) + // 串行段（写集冲突步骤，按声明序逐个执行；顺序步骤后可动态调整）。 + for _, idx := range layer.seq { + if out, g2, err := d.gateBeforeRun(ctx, st, idx); out != nil || g2 != nil || err != nil { + if out == nil && g2 != nil { + out = heldResult(st, g2) + } + return out, g2, err + } + out, g2, err := d.runOneStep(ctx, st, idx) + if out != nil || g2 != nil || err != nil { + if out == nil && g2 != nil { + out = heldResult(st, g2) + } + return out, g2, err } - return out, g, err - } - i++ - - // 阻塞缺口挂起。 - if g2 := d.checkBlockingGaps(st); g2 != nil { - return heldResult(st, g2), g2, nil - } - - // 动态调整（仅顺序步骤后，避免并发竞态）。 - if st.Plan.Workflow.Steps[i-1].AllowDynamicAdjustment && d.opts.Evaluator != nil { - out, g, err := d.applyDynamicAdjustment(ctx, st, i-1) - if out != nil || g != nil || err != nil { - if out == nil && g != nil { - out = heldResult(st, g) + if g2 := d.checkBlockingGaps(st); g2 != nil { + return heldResult(st, g2), g2, nil + } + if st.Plan.Workflow.Steps[idx].AllowDynamicAdjustment && d.opts.Evaluator != nil { + out, g2, err := d.applyDynamicAdjustment(ctx, st, idx) + if out != nil || g2 != nil || err != nil { + if out == nil && g2 != nil { + out = heldResult(st, g2) + } + return out, g2, err } - return out, g, err } } } -- 2.54.0