diff --git a/squad/engine.go b/squad/engine.go index 9e886e8..c8d3414 100644 --- a/squad/engine.go +++ b/squad/engine.go @@ -100,6 +100,12 @@ type RunOptions struct { MaxLeaderTurns int `json:"max_leader_turns,omitempty"` // CheckpointMode superstep 边界快照档位(Q1 WP4):sync(默认)| async | on_exit。 CheckpointMode string `json:"checkpoint_mode,omitempty"` + // Topology 显式拓扑名(Q2;优先于 Blueprint.Topology;空 = standard 图)。 + // 未注册名 / primitive 单跑 / driver 未点亮均 fail-closed。 + Topology string `json:"topology,omitempty"` + // Deadline 运行级截止时刻(Q2 long_running):superstep 边界检查, + // 超期把未完步骤置 skipped 并降级收口。零值 = 不设限。 + Deadline time.Time `json:"deadline,omitempty"` } // SquadEngine 小队引擎门面:Run / Respond / Resume / Inspect / ListRuns。 @@ -234,6 +240,13 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (* if runOpts.CheckpointMode != "" { meta.Metadata["checkpoint_mode"] = runOpts.CheckpointMode } + if runOpts.Topology != "" { + meta.Metadata["topology"] = runOpts.Topology + } + if !runOpts.Deadline.IsZero() { + // RFC3339Nano:秒级 RFC3339 会截断亚秒精度,短截止会被立刻判超期。 + meta.Metadata["deadline"] = runOpts.Deadline.UTC().Format(time.RFC3339Nano) + } e.deps.Sink.Emit(Event{Type: EvRunStarted, RunID: runID, Text: goal.Text, At: time.Now()}) // 意图闸门(可选,入口一次裁决;显式 TaskKey 优先,不重复过闸)。 @@ -294,6 +307,10 @@ func (e *SquadEngine) planAndExecute(ctx context.Context, st *RunState, runOpts planOut = &PlanOutput{Plan: plan} } else { bpMatch, _ := e.opts.Blueprints.Match(meta.Goal.Text, runOpts.TaskKey) + // 拓扑解析(Q2):RunOptions 优先,Blueprint 兜底。 + if meta.Metadata["topology"] == "" && bpMatch.Topology != "" { + meta.Metadata["topology"] = bpMatch.Topology + } planOut, err = e.opts.Planner.Plan(ctx, PlanInput{ Goal: meta.Goal, Blueprint: bpPtr(bpMatch), Registry: e.opts.Registry, RunID: meta.ID, IntentCatalog: e.opts.IntentCatalog, @@ -380,6 +397,20 @@ func (e *SquadEngine) planAndExecute(ctx context.Context, st *RunState, runOpts if err := e.materializeWorkspace(meta, st, planOut.Plan); err != nil { return nil, err } + // 拓扑校验与分发(Q2):未注册名 / primitive 单跑 / driver 未点亮 fail-closed; + // leader 拓扑映射到既有 Mode 委派循环(不新造执行面)。 + if topo := meta.Metadata["topology"]; topo != "" { + kind, terr := checkTopology(topo) + if terr != nil { + return e.failNew(meta, terr) + } + if kind == TopologyMode && topo == TopologyLeader { + if meta.Metadata["mode"] == "" { + meta.Metadata["mode"] = "leader" + } + } + e.deps.Sink.Emit(Event{Type: EvPhase, RunID: meta.ID, Text: "拓扑 " + topo, At: time.Now()}) + } // leader-delegation 模式:忽略规划步骤,进 leader 委派循环。 if meta.Metadata["mode"] == "leader" { e.initLeaderState(st) diff --git a/squad/graph.go b/squad/graph.go index 461f7ee..aa58695 100644 --- a/squad/graph.go +++ b/squad/graph.go @@ -71,6 +71,45 @@ func compileGraph(steps []StepSpec) (*graph, error) { 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 } diff --git a/squad/models.go b/squad/models.go index 884b1a0..187588d 100644 --- a/squad/models.go +++ b/squad/models.go @@ -66,9 +66,12 @@ type Blueprint struct { Title string `json:"title"` Match []string `json:"match,omitempty"` // 关键词命中即视为命中模板 Description string `json:"description,omitempty"` - Agents []RoleSpec `json:"agents,omitempty"` // 默认编制(引用已注册角色或内联角色) - Steps []StepSpec `json:"steps"` - Template string `json:"template,omitempty"` // 交付物骨架模板(Markdown) + // Topology 声明模板的拓扑(Q2;空 = standard 图行为)。 + // RunOptions.Topology 优先于本字段。 + Topology string `json:"topology,omitempty"` + Agents []RoleSpec `json:"agents,omitempty"` // 默认编制(引用已注册角色或内联角色) + Steps []StepSpec `json:"steps"` + Template string `json:"template,omitempty"` // 交付物骨架模板(Markdown) } // PlanningMode 规划模式。 @@ -115,6 +118,43 @@ type StepSpec struct { Skills []string `json:"skills,omitempty"` // MinScore 质检总分下限。0 = 不额外设门槛,只看 Checker.Passed。 MinScore int `json:"min_score,omitempty"` + // Loop 非空 = 循环步(Q2 loop 拓扑):完成后条件不满足且预算未尽时 + // 重置为 pending 再武装(图不允许环,循环由引擎再武装实现)。 + Loop *LoopSpec `json:"loop,omitempty"` + // LoopRuns 已再武装次数(引擎维护,随计划持久化)。 + LoopRuns int `json:"loop_runs,omitempty"` + // Branches 非空 = 路由步(Q2 router 拓扑):完成后按黑板字段判定, + // 未选中分支的目标步骤置 skipped(skipped=依赖已满足,join 步照常推进)。 + Branches []Branch `json:"branches,omitempty"` + // Handoff 非空 = 完成即把最终答案所有权移交给该角色(Q2 handoff 原语)。 + Handoff string `json:"handoff,omitempty"` +} + +// LoopSpec 是循环步的预算与停机条件(Q2)。 +type LoopSpec struct { + // MaxIterations 再武装上限(必须 >0;触顶即停,带 loop_exhausted 事件)。 + MaxIterations int `json:"max_iterations"` + // UntilField 停机判据的黑板字段名;空 = 只按 MaxIterations 迭代。 + UntilField string `json:"until_field,omitempty"` + // UntilEquals 停机值:Fields[UntilField] == UntilEquals 时停。 + UntilEquals string `json:"until_equals,omitempty"` + // Body 与本步一起再武装的其他步骤 ID(循环体;如 evaluator-optimizer + // 的评估步)。空 = 只再武装本步。 + Body []string `json:"body,omitempty"` +} + +// Branch 是路由步的一条分支(Q2)。When 按声明序取首个命中。 +type Branch struct { + When BranchCond `json:"when"` + // Then 是本分支保留的步骤 ID(其余分支目标置 skipped)。 + Then []string `json:"then"` +} + +// BranchCond 是分支判据:黑板字段等值/不等值。 +type BranchCond struct { + Field string `json:"field"` + Equals string `json:"equals,omitempty"` + NotEquals string `json:"not_equals,omitempty"` } // IsCritical 报告步骤是否关键(nil 视为关键)。 diff --git a/squad/topology.go b/squad/topology.go new file mode 100644 index 0000000..42f11b0 --- /dev/null +++ b/squad/topology.go @@ -0,0 +1,130 @@ +package squad + +import "fmt" + +// topology.go — 拓扑一等化(Q2):内置拓扑名注册表。 +// 拓扑不新造执行面:graph 类编译为 Steps 图(Q1 机制),leader 走既有 +// Mode 委派循环,driver 类由引擎按名分发(review/group_chat/task_board), +// primitive 类是跨拓扑原语(handoff/mailbox),不可单独作为运行拓扑。 +// 未知名与不可运行Kind一律 fail-closed。 + +// 内置拓扑名(completion doc Q2 清单的 12 种)。 +const ( + TopologySequential = "sequential" + TopologyParallel = "parallel" + TopologyLoop = "loop" + TopologyRouter = "router" + TopologyLeader = "leader" + TopologyHandoff = "handoff" + TopologyGroupChat = "group_chat" + TopologyReview = "review" + TopologyEvaluatorOptimizer = "evaluator_optimizer" + TopologyLongRunning = "long_running" + TopologyTaskBoard = "task_board" + TopologyMailbox = "mailbox" +) + +// TopologyKind 是拓扑的执行形态。 +type TopologyKind string + +const ( + // TopologyGraph 编译为 Steps 图,由 flowDriver.advance 执行。 + TopologyGraph TopologyKind = "graph" + // TopologyMode 走既有 Mode 分支(leader 委派循环)。 + TopologyMode TopologyKind = "mode" + // TopologyDriver 由引擎按名分发到专用驱动。 + TopologyDriver TopologyKind = "driver" + // TopologyPrimitive 是跨拓扑原语,不可单独运行。 + TopologyPrimitive TopologyKind = "primitive" +) + +// TopologySpec 是一个内置拓扑的注册项。 +type TopologySpec struct { + Name string + Kind TopologyKind +} + +// builtinTopologies 是 12 种内置拓扑的注册表(唯一权威清单)。 +var builtinTopologies = []TopologySpec{ + {Name: TopologySequential, Kind: TopologyGraph}, + {Name: TopologyParallel, Kind: TopologyGraph}, + {Name: TopologyLoop, Kind: TopologyGraph}, + {Name: TopologyRouter, Kind: TopologyGraph}, + {Name: TopologyLeader, Kind: TopologyMode}, + {Name: TopologyHandoff, Kind: TopologyPrimitive}, + {Name: TopologyGroupChat, Kind: TopologyDriver}, + {Name: TopologyReview, Kind: TopologyDriver}, + {Name: TopologyEvaluatorOptimizer, Kind: TopologyGraph}, + {Name: TopologyLongRunning, Kind: TopologyGraph}, + {Name: TopologyTaskBoard, Kind: TopologyDriver}, + {Name: TopologyMailbox, Kind: TopologyPrimitive}, +} + +// BuiltinTopologies 返回内置拓扑清单(只读副本)。 +func BuiltinTopologies() []TopologySpec { + out := make([]TopologySpec, len(builtinTopologies)) + copy(out, builtinTopologies) + return out +} + +// LookupTopology 按名查注册项。 +func LookupTopology(name string) (TopologySpec, bool) { + for _, t := range builtinTopologies { + if t.Name == name { + return t, true + } + } + return TopologySpec{}, false +} + +// registeredDrivers 是已接线的 driver 拓扑(WP 增量逐个点亮;未点亮 = fail-closed)。 +var registeredDrivers = map[string]bool{} + +// RegisterTopologyDriver 由驱动实现方在 init/装配时点亮一个 driver 拓扑。 +func RegisterTopologyDriver(name string) error { + spec, ok := LookupTopology(name) + if !ok { + return fmt.Errorf("topology: 未注册的拓扑名 %q", name) + } + if spec.Kind != TopologyDriver { + return fmt.Errorf("topology: %q 不是 driver 类(kind=%s)", name, spec.Kind) + } + registeredDrivers[name] = true + return nil +} + +// checkTopology 校验一次运行的有效拓扑名,返回其 Kind。 +// 空名 = 未声明(standard 图行为,兼容既有调用方)。 +// 未注册名 / primitive 单跑 / driver 未点亮都 fail-closed。 +func checkTopology(name string) (TopologyKind, error) { + if name == "" { + return TopologyGraph, nil + } + spec, ok := LookupTopology(name) + if !ok { + return "", fmt.Errorf("topology: 未知拓扑 %q(内置清单:%s)", name, topologyNames()) + } + switch spec.Kind { + case TopologyGraph, TopologyMode: + return spec.Kind, nil + case TopologyPrimitive: + return "", fmt.Errorf("topology: %q 是跨拓扑原语,不可单独作为运行拓扑", name) + case TopologyDriver: + if !registeredDrivers[name] { + return "", fmt.Errorf("topology: %q 驱动未注册(fail-closed)", name) + } + return spec.Kind, nil + } + return "", fmt.Errorf("topology: %q kind 非法", name) +} + +func topologyNames() string { + s := "" + for i, t := range builtinTopologies { + if i > 0 { + s += ", " + } + s += t.Name + } + return s +} diff --git a/squad/topology_conformance_test.go b/squad/topology_conformance_test.go new file mode 100644 index 0000000..5f263f7 --- /dev/null +++ b/squad/topology_conformance_test.go @@ -0,0 +1,436 @@ +package squad + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// topology_conformance_test.go — Q2 验收一的机器证据:每种拓扑一个 +// 脚本化后端的 conformance(零模型、零网络)。 + +// ---- 脚本化夹具 ---- + +// collectSink 收集事件流供断言。 +type collectSink struct { + mu sync.Mutex + events []Event +} + +func (c *collectSink) Emit(ev Event) { + c.mu.Lock() + defer c.mu.Unlock() + c.events = append(c.events, ev) +} + +func (c *collectSink) snapshot() []Event { + c.mu.Lock() + defer c.mu.Unlock() + return append([]Event(nil), c.events...) +} + +func (c *collectSink) countText(sub string) int { + n := 0 + for _, ev := range c.snapshot() { + if strings.Contains(ev.Text, sub) { + n++ + } + } + return n +} + +// iterBackend 按「步骤子目录 × 第几次执行」回放内容(循环/评估拓扑需要 +// 逐轮不同的 fields.json)。 +type iterBackend struct { + mu sync.Mutex + calls map[string]int + content func(stepKey string, iter int, target string) []byte + delay time.Duration +} + +func (b *iterBackend) Name() string { return "iter" } + +func (b *iterBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) { + key := filepath.Base(filepath.Clean(req.WorkDir)) + if b.delay > 0 { + time.Sleep(b.delay) + } + b.mu.Lock() + b.calls[key]++ + iter := b.calls[key] + b.mu.Unlock() + patches := map[string][]byte{} + for _, t := range req.Targets { + var data []byte + if b.content != nil { + data = b.content(key, iter, t) + } + if data == nil { + data = []byte(goodReport) + } + out := filepath.Join(req.WorkDir, t) + _ = os.MkdirAll(filepath.Dir(out), 0o755) + if err := os.WriteFile(out, data, 0o644); err != nil { + return nil, err + } + patches[t] = data + } + return &ExecResult{Success: true, Patches: patches}, nil +} + +func (b *iterBackend) totalCalls() int { + b.mu.Lock() + defer b.mu.Unlock() + n := 0 + for _, v := range b.calls { + n += v + } + return n +} + +func newTopologyEngine(t *testing.T, planner Planner, be Executor, sink ProgressSink) *SquadEngine { + t.Helper() + dir := t.TempDir() + reg := NewRoleRegistry() + for _, spec := range []RoleSpec{ + {Role: "researcher", Name: "调研", Backend: "iter"}, + {Role: "writer", Name: "撰写", Backend: "iter"}, + {Role: "reviewer", Name: "审核", Backend: "iter"}, + } { + if err := reg.Register(spec); err != nil { + t.Fatal(err) + } + } + backends := NewBackendResolver() + backends.RegisterInstance("iter", be) + if sink == nil { + sink = NopSink{} + } + engine, err := NewEngine(Options{ + DataDir: dir, Registry: reg, Planner: planner, Backends: backends, Sink: sink, + }) + if err != nil { + t.Fatal(err) + } + return engine +} + +func fieldsJSON(kv map[string]string) []byte { + b, _ := json.Marshal(kv) + return b +} + +// ---- sequential / parallel:既有图语义 + 拓扑名受理 ---- + +func TestTopologyConformance_sequential(t *testing.T) { + be := &iterBackend{calls: map[string]int{}} + sink := &collectSink{} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"notes.md"}}, + StepSpec{ID: "s2", Role: "writer", Title: "二", Targets: []string{"output.md"}}, + ), be, sink) + res, err := engine.Run(context.Background(), Goal{Text: "顺序任务"}, RunOptions{Topology: TopologySequential}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + if be.totalCalls() != 2 { + t.Fatalf("执行次数 = %d", be.totalCalls()) + } + if sink.countText("拓扑 sequential") != 1 { + t.Fatal("缺拓扑受理事件") + } +} + +func TestTopologyConformance_parallel(t *testing.T) { + be := &iterBackend{calls: map[string]int{}} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "并行甲", Parallel: true, Targets: []string{"a.md"}}, + StepSpec{ID: "s2", Role: "writer", Title: "并行乙", Parallel: true, Targets: []string{"b.md"}}, + StepSpec{ID: "s3", Role: "reviewer", Title: "汇合", Targets: []string{"output.md"}}, + ), be, nil) + res, err := engine.Run(context.Background(), Goal{Text: "并行任务"}, RunOptions{Topology: TopologyParallel}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted || be.totalCalls() != 3 { + t.Fatalf("status = %s calls = %d err = %v", res.Status, be.totalCalls(), res.Error) + } +} + +// ---- loop:条件停机 + 预算触顶 ---- + +func TestTopologyConformance_loopUntilField(t *testing.T) { + // 第 1-2 轮 done=no,第 3 轮 done=yes → 恰执行 3 次。 + be := &iterBackend{calls: map[string]int{}, content: func(_ string, iter int, target string) []byte { + if target == "fields.json" { + if iter >= 3 { + return fieldsJSON(map[string]string{"done": "yes"}) + } + return fieldsJSON(map[string]string{"done": "no"}) + } + return nil + }} + sink := &collectSink{} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "循环体", Targets: []string{"output.md", "fields.json"}, + Loop: &LoopSpec{MaxIterations: 5, UntilField: "done", UntilEquals: "yes"}}, + ), be, sink) + res, err := engine.Run(context.Background(), Goal{Text: "循环任务"}, RunOptions{Topology: TopologyLoop}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + if be.totalCalls() != 3 { + t.Fatalf("应在条件满足时停机(3 次),实际 %d", be.totalCalls()) + } + if sink.countText("再武装") != 2 { + t.Fatalf("再武装事件应 2 次:%d", sink.countText("再武装")) + } + // 计划里的 LoopRuns 持久化为 2。 + if len(res.Plan.Steps) != 1 || res.Plan.Steps[0].LoopRuns != 2 { + t.Fatalf("LoopRuns = %+v", res.Plan.Steps) + } +} + +func TestTopologyConformance_loopBudgetExhausted(t *testing.T) { + // 条件永不满足:恰执行 MaxIterations 次后触顶(非失败)。 + be := &iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte { + if target == "fields.json" { + return fieldsJSON(map[string]string{"done": "no"}) + } + return nil + }} + sink := &collectSink{} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "循环体", Targets: []string{"output.md", "fields.json"}, + Loop: &LoopSpec{MaxIterations: 3, UntilField: "done", UntilEquals: "yes"}}, + ), be, sink) + res, err := engine.Run(context.Background(), Goal{Text: "循环任务"}, RunOptions{Topology: TopologyLoop}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("触顶应是完成而非失败:%s %v", res.Status, res.Error) + } + if be.totalCalls() != 3 { + t.Fatalf("恰 3 次,实际 %d", be.totalCalls()) + } + if sink.countText("loop_exhausted") != 1 { + t.Fatal("缺 loop_exhausted 事件") + } +} + +// ---- router:分支选中/落选跳过/无命中 fail-closed ---- + +func routerSteps() []StepSpec { + return []StepSpec{ + {ID: "r1", Role: "researcher", Title: "路由", Targets: []string{"output.md", "fields.json"}, + Branches: []Branch{ + {When: BranchCond{Field: "route", Equals: "fast"}, Then: []string{"fast"}}, + {When: BranchCond{Field: "route", NotEquals: "fast"}, Then: []string{"slow"}}, + }}, + {ID: "fast", Role: "writer", Title: "快线", Needs: []string{"r1"}, Targets: []string{"output.md"}}, + {ID: "slow", Role: "writer", Title: "慢线", Needs: []string{"r1"}, Targets: []string{"output.md"}}, + } +} + +func TestTopologyConformance_routerSelectsBranch(t *testing.T) { + be := &iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte { + if target == "fields.json" { + return fieldsJSON(map[string]string{"route": "fast"}) + } + return nil + }} + engine := newTopologyEngine(t, mockScriptPlanner(routerSteps()...), be, nil) + res, err := engine.Run(context.Background(), Goal{Text: "路由任务"}, RunOptions{Topology: TopologyRouter}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("分支落选是预期路径,不得降级/失败:%s %v", res.Status, res.Error) + } + byID := map[string]StepSpec{} + for _, s := range res.Plan.Steps { + byID[s.ID] = s + } + if byID["fast"].Status != "completed" { + t.Fatalf("fast = %s", byID["fast"].Status) + } + if byID["slow"].Status != "skipped" || byID["slow"].Error != routerSkipReason { + t.Fatalf("slow = %s %q", byID["slow"].Status, byID["slow"].Error) + } +} + +func TestTopologyConformance_routerNoMatchFailsClosed(t *testing.T) { + be := &iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte { + if target == "fields.json" { + return fieldsJSON(map[string]string{"route": ""}) // 两条分支都不命中 + } + return nil + }} + steps := routerSteps() + // 去掉 NotEquals 兜底分支,制造无命中。 + steps[0].Branches = []Branch{{When: BranchCond{Field: "route", Equals: "fast"}, Then: []string{"fast"}}} + engine := newTopologyEngine(t, mockScriptPlanner(steps...), be, nil) + res, err := engine.Run(context.Background(), Goal{Text: "路由任务"}, RunOptions{Topology: TopologyRouter}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunFailed || !strings.Contains(res.Error, "无分支命中") { + t.Fatalf("无命中必须 fail-closed:%s %v", res.Status, res.Error) + } +} + +// ---- evaluator-optimizer:draft→eval 循环复合,收敛即停 ---- + +func TestTopologyConformance_evaluatorOptimizer(t *testing.T) { + // eval 第 1-2 轮 verdict=fail,第 3 轮 pass;draft 随 body 再武装。 + be := &iterBackend{calls: map[string]int{}, content: func(key string, iter int, target string) []byte { + if target == "fields.json" { + if iter >= 3 { + return fieldsJSON(map[string]string{"verdict": "pass"}) + } + return fieldsJSON(map[string]string{"verdict": "fail"}) + } + if target == "verdict.md" { + return []byte(fmt.Sprintf("评审第 %d 轮\n\n%s", iter, goodReport)) + } + return nil + }} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "draft", Role: "writer", Title: "起草", Targets: []string{"output.md"}}, + StepSpec{ID: "eval", Role: "reviewer", Title: "评审", Targets: []string{"verdict.md", "fields.json"}, + Loop: &LoopSpec{MaxIterations: 4, UntilField: "verdict", UntilEquals: "pass", Body: []string{"draft"}}}, + ), be, nil) + res, err := engine.Run(context.Background(), Goal{Text: "起草并评审"}, RunOptions{Topology: TopologyEvaluatorOptimizer}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + byID := map[string]StepSpec{} + for _, s := range res.Plan.Steps { + byID[s.ID] = s + } + if byID["eval"].LoopRuns != 2 { + t.Fatalf("eval 应再武装 2 轮:%+v", byID["eval"]) + } + // draft 3 次 + eval 3 次。 + if be.totalCalls() != 6 { + t.Fatalf("总执行次数 = %d,期望 6", be.totalCalls()) + } +} + +// ---- long-running:截止时刻 → 未完步骤 skipped、降级收口 ---- + +func TestTopologyConformance_longRunningDeadline(t *testing.T) { + // 时序鲁棒设计:每步 300ms、截止 +500ms——装配耗时不影响断言核心: + // s1 必完成(首个 superstep 在截止前开跑),s3 必被跳过(第三个 + // superstep 边界必超期),收口必为 degraded。s2 完成与否取决于装配 + // 耗时,不做断言。 + be := &iterBackend{calls: map[string]int{}, delay: 300 * time.Millisecond} + sink := &collectSink{} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"notes.md"}}, + StepSpec{ID: "s2", Role: "writer", Title: "二", Targets: []string{"output.md"}}, + StepSpec{ID: "s3", Role: "reviewer", Title: "三", Targets: []string{"output.md"}}, + ), be, sink) + res, err := engine.Run(context.Background(), Goal{Text: "长任务"}, RunOptions{ + Topology: TopologyLongRunning, + Deadline: time.Now().Add(500 * time.Millisecond), + }) + if err != nil { + t.Fatal(err) + } + if res.Status != RunDegraded { + t.Fatalf("截止后应降级收口:%s %v", res.Status, res.Error) + } + byID := map[string]StepSpec{} + for _, s := range res.Plan.Steps { + byID[s.ID] = s + } + if byID["s1"].Status != "completed" { + t.Fatalf("s1 = %s", byID["s1"].Status) + } + if byID["s3"].Status != "skipped" || byID["s3"].Error != "deadline exceeded" { + t.Fatalf("s3 = %s %q", byID["s3"].Status, byID["s3"].Error) + } + if sink.countText("截止时刻") != 1 { + t.Fatal("缺截止事件") + } +} + +// ---- 拓扑校验 fail-closed ---- + +func TestTopologyValidationFailsClosed(t *testing.T) { + be := &iterBackend{calls: map[string]int{}} + cases := []struct { + name string + topo string + wantErr string + }{ + {"未注册名", "bogus_topology", "未知拓扑"}, + {"primitive 单跑", TopologyMailbox, "跨拓扑原语"}, + {"primitive 单跑 handoff", TopologyHandoff, "跨拓扑原语"}, + {"driver 未点亮", TopologyReview, "驱动未注册"}, + {"driver 未点亮 group_chat", TopologyGroupChat, "驱动未注册"}, + {"driver 未点亮 task_board", TopologyTaskBoard, "驱动未注册"}, + } + for _, tc := range cases { + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}}, + ), be, nil) + res, err := engine.Run(context.Background(), Goal{Text: "x"}, RunOptions{Topology: tc.topo}) + if err != nil { + t.Fatalf("%s:err = %v", tc.name, err) + } + if res.Status != RunFailed || !strings.Contains(res.Error, tc.wantErr) { + t.Fatalf("%s:%s %q(期望含 %q)", tc.name, res.Status, res.Error, tc.wantErr) + } + } +} + +// leader 拓扑名映射到既有 Mode 委派循环(本测只钉「受理且映射」, +// leader 行为 conformance 在 leader_test.go / WP2 增强)。 +func TestTopologyLeaderMapsToMode(t *testing.T) { + spec, ok := LookupTopology(TopologyLeader) + if !ok || spec.Kind != TopologyMode { + t.Fatalf("leader 注册形态:%+v %v", spec, ok) + } + if _, err := checkTopology(TopologyLeader); err != nil { + t.Fatalf("leader 拓扑应可受理:%v", err) + } +} + +// 12 种内置拓扑全部注册且 Kind 合法(清单完整性)。 +func TestBuiltinTopologiesComplete(t *testing.T) { + want := map[string]TopologyKind{ + TopologySequential: TopologyGraph, TopologyParallel: TopologyGraph, + TopologyLoop: TopologyGraph, TopologyRouter: TopologyGraph, + TopologyLeader: TopologyMode, TopologyHandoff: TopologyPrimitive, + TopologyGroupChat: TopologyDriver, TopologyReview: TopologyDriver, + TopologyEvaluatorOptimizer: TopologyGraph, TopologyLongRunning: TopologyGraph, + TopologyTaskBoard: TopologyDriver, TopologyMailbox: TopologyPrimitive, + } + got := BuiltinTopologies() + if len(got) != len(want) { + t.Fatalf("注册数 = %d,期望 %d", len(got), len(want)) + } + for _, spec := range got { + if k, ok := want[spec.Name]; !ok || k != spec.Kind { + t.Fatalf("拓扑 %s kind = %s(期望 %s)", spec.Name, spec.Kind, k) + } + } +} diff --git a/squad/workflow.go b/squad/workflow.go index ebc53fc..1c18e0e 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -29,6 +29,15 @@ func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult, d.emit(st, Event{Type: EvPhase, Text: "开始执行工作流"}) superNo := 0 + // 运行级截止(Q2 long_running):superstep 边界检查,超期未完步骤置 + // skipped 并降级收口;零值/不可解析 = 不设限。 + var runDeadline time.Time + if v := st.Meta.Metadata["deadline"]; v != "" { + // RFC3339Nano 同时接受秒级与纳秒级两种写法。 + if t, terr := time.Parse(time.RFC3339Nano, v); terr == nil { + runDeadline = t + } + } cpw := newCheckpointWriter(CheckpointMode(st.Meta.Metadata["checkpoint_mode"]), st.storeOr()) // 推进退出前 flush:async 落完、on_exit 写最终快照(Q1 WP4 三档的出口语义)。 var lastGraph *graph @@ -43,6 +52,19 @@ func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult, } }() for { + if !runDeadline.IsZero() && time.Now().After(runDeadline) { + d.emit(st, Event{Type: EvPhase, Text: "运行截止时刻已到,未完步骤按 skipped 收口"}) + st.mu.Lock() + for i := range st.Plan.Workflow.Steps { + if st.Plan.Workflow.Steps[i].Status == "pending" || st.Plan.Workflow.Steps[i].Status == "running" { + st.Plan.Workflow.Steps[i].Status = "skipped" + st.Plan.Workflow.Steps[i].Error = "deadline exceeded" + } + } + _ = st.save() + st.mu.Unlock() + return d.finalize(ctx, st) + } g, gerr := compileGraph(st.Plan.Workflow.Steps) if gerr != nil { wf2 := st.Workflow() @@ -197,6 +219,7 @@ func (d *flowDriver) runParallel(ctx context.Context, st *RunState, idxs []int) var wg sync.WaitGroup var mu sync.Mutex failedCritical := -1 + var firstFail *RunResult var gate *HitlGate for _, idx := range idxs { wg.Add(1) @@ -211,6 +234,11 @@ func (d *flowDriver) runParallel(ctx context.Context, st *RunState, idxs []int) } if (out != nil && out.Status == RunFailed) || err != nil { cancel() + // 保留首个带具体错误的失败结果(如 router 无命中); + // failStep 的通用格式只在没有更具体结果时兜底。 + if out != nil && out.Status == RunFailed && firstFail == nil { + firstFail = out + } if st.Plan.Workflow.Steps[idx].IsCritical() && failedCritical < 0 { failedCritical = idx } @@ -221,6 +249,9 @@ func (d *flowDriver) runParallel(ctx context.Context, st *RunState, idxs []int) if gate != nil { return heldResult(st, gate), gate, nil } + if firstFail != nil { + return firstFail, nil, nil + } if failedCritical >= 0 { return d.failStep(ctx, st, failedCritical) } @@ -423,9 +454,148 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru } d.emit(st, Event{Type: EvStepCompleted, StepID: step.ID, Role: step.Role, Text: step.Title, Data: marshalJSON(map[string]any{"duration_ms": step.DurationMs, "targets": targets})}) + + // 循环步再武装(Q2 loop):停机条件不满足且预算未尽 → 本步与 loop.body + // 一起重置为 pending;触顶发 loop_exhausted 事件后按完成继续。 + if step.Loop != nil { + if d.loopShouldRearm(st, step) { + if out, g2, err2 := d.rearmLoop(ctx, st, idx, step); out != nil || g2 != nil || err2 != nil { + return out, g2, err2 + } + return nil, nil, nil + } + if step.Loop.UntilField != "" && step.LoopRuns+1 >= step.Loop.MaxIterations { + d.emit(st, Event{Type: EvPhase, StepID: step.ID, Text: fmt.Sprintf("loop_exhausted:%s 达 %d 轮上限,停机条件未满足", step.ID, step.Loop.MaxIterations)}) + } + } + // 路由步分支裁决(Q2 router):首个命中的分支保留,其余分支目标置 + // skipped;无分支命中 fail-closed 成失败运行。 + if len(step.Branches) > 0 { + return d.applyBranches(ctx, st, idx, step) + } return nil, nil, nil } +// loopShouldRearm 判定循环步是否再武装:停机条件(黑板字段等值)满足或 +// 预算用尽则停。字段判据读的是本轮 mergeStepArtifacts 之后的黑板。 +func (d *flowDriver) loopShouldRearm(st *RunState, step StepSpec) bool { + // MaxIterations 是总迭代上限(含首跑):已跑 LoopRuns+1 次。 + if step.LoopRuns+1 >= step.Loop.MaxIterations { + return false + } + if step.Loop.UntilField != "" { + st.mu.Lock() + v := st.BB.Fields[step.Loop.UntilField] + st.mu.Unlock() + if v == step.Loop.UntilEquals { + return false + } + } + return true +} + +// rearmLoop 把循环步与 loop.body 一起重置为 pending(同一临界区内落盘)。 +func (d *flowDriver) rearmLoop(_ context.Context, st *RunState, idx int, step StepSpec) (*RunResult, *HitlGate, error) { + st.mu.Lock() + step.LoopRuns++ + step.Status = "pending" + step.Error = "" + st.Plan.Workflow.Steps[idx] = step + for _, bodyID := range step.Loop.Body { + for j := range st.Plan.Workflow.Steps { + if st.Plan.Workflow.Steps[j].ID == bodyID && st.Plan.Workflow.Steps[j].Status == "completed" { + st.Plan.Workflow.Steps[j].Status = "pending" + st.Plan.Workflow.Steps[j].Error = "" + } + } + } + err := st.save() + round := step.LoopRuns + st.mu.Unlock() + if err != nil { + return nil, nil, err + } + d.emit(st, Event{Type: EvPhase, StepID: step.ID, Text: fmt.Sprintf("loop 第 %d 轮再武装:%s", round+1, step.ID)}) + return nil, nil, nil +} + +// applyBranches 裁决路由步分支:When 按声明序取首个命中;未选中分支的 +// pending 目标置 skipped(skipped=依赖已满足,join 步照常推进)。 +func (d *flowDriver) applyBranches(_ context.Context, st *RunState, _ int, step StepSpec) (*RunResult, *HitlGate, error) { + st.mu.Lock() + fields := map[string]string{} + for k, v := range st.BB.Fields { + fields[k] = v + } + st.mu.Unlock() + var selected *Branch + for i := range step.Branches { + if branchMatches(fields, step.Branches[i].When) { + selected = &step.Branches[i] + break + } + } + if selected == nil { + wf := st.Workflow() + wf.Status = "failed" + st.setWorkflow(wf) + _ = st.save() + return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), + Error: fmt.Sprintf("router: 步骤 %s 无分支命中(fail-closed)", step.ID), WorkDir: st.Meta.WorkDir}, nil, nil + } + keep := map[string]bool{} + for _, t := range selected.Then { + keep[t] = true + } + skipped := 0 + st.mu.Lock() + for j := range st.Plan.Workflow.Steps { + sj := &st.Plan.Workflow.Steps[j] + if sj.ID == step.ID || keep[sj.ID] || sj.Status != "pending" { + continue + } + for _, br := range step.Branches { + inBranch := false + for _, t := range br.Then { + if t == sj.ID { + inBranch = true + break + } + } + if inBranch { + sj.Status = "skipped" + sj.Error = routerSkipReason + skipped++ + break + } + } + } + err := st.save() + st.mu.Unlock() + if err != nil { + return nil, nil, err + } + d.emit(st, Event{Type: EvPhase, StepID: step.ID, + Text: fmt.Sprintf("router %s 选中分支 %v(跳过 %d 步)", step.ID, selected.Then, skipped)}) + return nil, nil, nil +} + +// routerSkipReason 是分支落选的 skipped 标记(finalize 的降级豁免凭据)。 +const routerSkipReason = "router: 分支未选中" + +// branchMatches 判定分支条件:Equals 非空须相等,NotEquals 非空须不等; +// 字段未设时按零值 "" 参与比较(compileGraph 已保证两者至少其一非空)。 +func branchMatches(fields map[string]string, c BranchCond) bool { + v := fields[c.Field] + if c.Equals != "" && v != c.Equals { + return false + } + if c.NotEquals != "" && v == c.NotEquals { + return false + } + return true +} + // failStep 按关键性处理失败步骤:关键 → 运行失败;非关键 → 标记后继续(返回 nil 让循环推进)。 func (d *flowDriver) failStep(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) { st.mu.Lock() @@ -587,7 +757,11 @@ func (d *flowDriver) finalize(ctx context.Context, st *RunState) (*RunResult, *H wf := st.Workflow() degraded := false for _, s := range wf.Steps { - if s.Status == "failed" || s.Status == "skipped" { + if s.Status == "failed" { + degraded = true + } + // router 分支落选是预期路径,不算降级(Q2)。 + if s.Status == "skipped" && s.Error != routerSkipReason { degraded = true } }