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
+}

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))
+	}
+}

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 策略是否还能重试。

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
 			}
 		}
 	}