From acbe7441420f42e8af9e57e1b60c4c6bc883443e Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 8 Oct 2026 01:29:03 +0000 Subject: [PATCH] Add task-board driver with claim/lease/complete and 12-topology aggregate (Q2 WP4) Why: Q2 acceptance 1 requires a scripted-backend conformance for every one of the 12 builtin topologies; task_board was the last unlit driver. Evidence: wp4_conformance_test.go: diamond-dependency board respects the dependency gate in claim order; orphan lease (claimed + expired) re-arms on driver re-entry and re-claims with Attempts=2 via Resume; non-critical task failure cascades dependents to skipped and finishes degraded; fail-closed when tasks remain unclosable. TestTopologyConformance_allTwelve runs all 12 names (10 runnable to completion, handoff/mailbox rejected as primitives). Full suite green with -race. Impact: task_board lit in the topology registry (all 12 topologies now covered); RunState gains Board (persisted in runstate.json); RunOptions gains TaskBoardLeaseSeconds; leader topology now maps to mode before plan confirmation so Topology=leader runs no longer hit the confirmation gate. Refs: Q2 plan WP4 (2026-10-07-agent-stack-q2q3-plans.md); task card .task/feature-agent-stack-q2.md --- squad/engine.go | 14 ++ squad/runstate.go | 6 + squad/taskboard.go | 308 +++++++++++++++++++++++++ squad/topology_conformance_test.go | 4 +- squad/wp4_conformance_test.go | 355 +++++++++++++++++++++++++++++ 5 files changed, 685 insertions(+), 2 deletions(-) create mode 100644 squad/taskboard.go create mode 100644 squad/wp4_conformance_test.go diff --git a/squad/engine.go b/squad/engine.go index 64847cb..67bdca1 100644 --- a/squad/engine.go +++ b/squad/engine.go @@ -124,6 +124,9 @@ type RunOptions struct { ReviewVerifyRole string `json:"review_verify_role,omitempty"` // ReviewTarget 被评审材料（默认取 context/goal.md）。 ReviewTarget string `json:"review_target,omitempty"` + // TaskBoardLeaseSeconds task_board 任务租约时限（秒；0 = 默认 600）。 + // 孤儿租约（claimed 且过期）在驱动重入时再武装回 open。 + TaskBoardLeaseSeconds int `json:"task_board_lease_seconds,omitempty"` } // SquadEngine 小队引擎门面：Run / Respond / Resume / Inspect / ListRuns。 @@ -260,6 +263,11 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (* } if runOpts.Topology != "" { meta.Metadata["topology"] = runOpts.Topology + // leader 拓扑映射到委派循环须发生在计划确认之前（确认门按 + // mode=leader 豁免）；后置映射会让 leader 运行误触计划确认。 + if runOpts.Topology == TopologyLeader && meta.Metadata["mode"] == "" { + meta.Metadata["mode"] = "leader" + } } if !runOpts.Deadline.IsZero() { // RFC3339Nano：秒级 RFC3339 会截断亚秒精度，短截止会被立刻判超期。 @@ -283,6 +291,9 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (* if runOpts.ReviewTarget != "" { meta.Metadata["review_target"] = runOpts.ReviewTarget } + if runOpts.TaskBoardLeaseSeconds > 0 { + meta.Metadata["task_board_lease_seconds"] = fmt.Sprintf("%d", runOpts.TaskBoardLeaseSeconds) + } e.deps.Sink.Emit(Event{Type: EvRunStarted, RunID: runID, Text: goal.Text, At: time.Now()}) // 意图闸门（可选，入口一次裁决；显式 TaskKey 优先，不重复过闸）。 @@ -564,6 +575,9 @@ func (e *SquadEngine) dispatchTopologyDriver(ctx context.Context, st *RunState) case TopologyGroupChat: out, err := e.runGroupChatDriver(ctx, st) return out, true, err + case TopologyTaskBoard: + out, err := e.runTaskBoardDriver(ctx, st) + return out, true, err default: return e.driverFail(st, fmt.Sprintf("topology: %q 驱动未实现（fail-closed）", topo)), true, nil } diff --git a/squad/runstate.go b/squad/runstate.go index 99aedaf..54f5966 100644 --- a/squad/runstate.go +++ b/squad/runstate.go @@ -33,6 +33,9 @@ type RunState struct { // Handoffs 所有权转移记录（Q2 handoff 原语）：随 RunState 持久化； // 目标角色的下一步骤一次性接收移交上下文，finalize 取链尾产出。 Handoffs []HandoffRecord `json:"handoffs,omitempty"` + // Board task_board 拓扑的队列状态（Q2）：claim/lease/complete 三原语， + // 随 RunState 持久化；孤儿租约在驱动重入时再武装。 + Board []BoardTask `json:"board,omitempty"` // Usage 按角色累计的 token 用量。 Usage map[string]*Usage `json:"usage,omitempty"` // Leader leader-delegation 模式的循环状态（workspace 内 leader.json 持久化）。 @@ -100,6 +103,7 @@ func (st *RunState) persist() error { Clarifications: st.Clarifications, Mailbox: st.Mailbox, Handoffs: st.Handoffs, + Board: st.Board, }) } @@ -110,6 +114,7 @@ type diskRunState struct { Clarifications []string `json:"clarifications,omitempty"` Mailbox []MailMessage `json:"mailbox,omitempty"` Handoffs []HandoffRecord `json:"handoffs,omitempty"` + Board []BoardTask `json:"board,omitempty"` } // Workflow 返回计划的编排清单（简洁访问器）。 @@ -166,6 +171,7 @@ func loadRunState(wsDir string) (*RunState, error) { st.Clarifications = disk.Clarifications st.Mailbox = disk.Mailbox st.Handoffs = disk.Handoffs + st.Board = disk.Board } else if !os.IsNotExist(err) { return nil, err } diff --git a/squad/taskboard.go b/squad/taskboard.go new file mode 100644 index 0000000..e1cad05 --- /dev/null +++ b/squad/taskboard.go @@ -0,0 +1,308 @@ +package squad + +import ( + "context" + "fmt" + "strings" + "time" +) + +// taskboard.go — task_board 拓扑驱动（Q2 WP4）。 +// 形态：计划步骤 = 板上任务（Needs = 依赖门）；执行语义是队列化的 +// claim/lease/complete 三原语——工作者认领任务（带租约时限）、执行、完成； +// 进程死亡留下的孤儿租约（claimed 且租约过期）在驱动重入时再武装回 open +// （Attempts 保留，超限由步骤级重试护栏兜底）。全部任务收口后走标准 +// finalize（交付物/质检/降级判定与图拓扑同一收口面）。 +// 挂起（HITL）安全：claimed 且步骤未完的任务在重入时续跑，不重复认领。 + +func init() { + _ = RegisterTopologyDriver(TopologyTaskBoard) +} + +// DefaultTaskLeaseSeconds 任务租约缺省时限（秒）。 +const DefaultTaskLeaseSeconds = 600 + +// boardSkipReason 依赖门无法满足（上游 failed/skipped）时的跳过原因。 +const boardSkipReason = "task_board: 依赖未满足（上游失败或跳过）" + +// BoardTask 是任务板上一条任务的队列状态。 +type BoardTask struct { + ID string `json:"id"` + Status string `json:"status"` // open / claimed / done / failed / skipped + ClaimedBy string `json:"claimed_by,omitempty"` + // LeaseExpires 认领租约到期时刻；过期未 complete = 孤儿租约，重入再武装。 + LeaseExpires time.Time `json:"lease_expires,omitempty"` + // Attempts 认领次数（孤儿再武装累加；执行重试由步骤级护栏负责）。 + Attempts int `json:"attempts,omitempty"` +} + +// runTaskBoardDriver task_board 拓扑入口（幂等可恢复）。 +func (e *SquadEngine) runTaskBoardDriver(ctx context.Context, st *RunState) (*RunResult, error) { + drv := &flowDriver{opts: e.deps} + e.initBoard(st) + e.rearmOrphanLeases(st) + leaseTTL := DefaultTaskLeaseSeconds + if v := st.Meta.Metadata["task_board_lease_seconds"]; v != "" { + if n := parseIntDefault(v, DefaultTaskLeaseSeconds); n > 0 { + leaseTTL = n + } + } + + for { + idx, held := e.boardHeldTask(st) + if !held { + var ok bool + idx, ok = e.boardClaimNext(st, leaseTTL) + if !ok { + // 无可认领：把依赖已死（上游 failed/skipped）的 open 任务级联 + // 置 skipped；再无变化则收口。 + if !e.boardSkipDeadlocked(st) { + break + } + continue + } + } + out, gate, err := e.boardRunTask(ctx, drv, st, idx) + if err != nil { + return nil, err + } + if gate != nil { + if out == nil { + out = heldResult(st, gate) + } + return out, nil + } + if out != nil { // 关键任务失败：fail-closed 传播（failStep 已置运行失败） + return out, nil + } + e.boardSettleTask(st, idx) + } + return e.boardFinish(ctx, drv, st) +} + +// initBoard 首次进入时从计划步骤派生任务板（恢复重入直读持久化状态）。 +func (e *SquadEngine) initBoard(st *RunState) { + st.mu.Lock() + defer st.mu.Unlock() + if len(st.Board) > 0 { + return + } + for _, s := range st.Plan.Workflow.Steps { + status := "open" + switch s.Status { + case "completed": + status = "done" + case "failed": + status = "failed" + case "skipped": + status = "skipped" + } + st.Board = append(st.Board, BoardTask{ID: s.ID, Status: status}) + } + _ = st.save() +} + +// rearmOrphanLeases 孤儿租约再武装：claimed 且租约过期 → open（进程死亡 +// 恢复路径；Attempts 保留作观测）。 +func (e *SquadEngine) rearmOrphanLeases(st *RunState) { + now := time.Now() + st.mu.Lock() + var rearmed []string + for i := range st.Board { + t := &st.Board[i] + if t.Status == "claimed" && !t.LeaseExpires.IsZero() && t.LeaseExpires.Before(now) { + t.Status = "open" + t.ClaimedBy = "" + rearmed = append(rearmed, t.ID) + } + } + changed := len(rearmed) > 0 + if changed { + _ = st.save() + } + st.mu.Unlock() + if changed { + (&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, + Text: fmt.Sprintf("task_board：孤儿租约再武装 %s", strings.Join(rearmed, ","))}) + } +} + +// boardHeldTask 返回 claimed 且步骤未完的任务（HITL 挂起/中断重入的续跑点）。 +func (e *SquadEngine) boardHeldTask(st *RunState) (int, bool) { + for ti := range st.Board { + t := &st.Board[ti] + if t.Status != "claimed" { + continue + } + if idx, s := e.boardStep(st, t.ID); idx >= 0 && (s.Status == "pending" || s.Status == "running") { + return idx, true + } + } + return -1, false +} + +// boardClaimNext 认领下一个可执行任务（open 且依赖门全 done），返回步骤下标。 +func (e *SquadEngine) boardClaimNext(st *RunState, leaseTTL int) (int, bool) { + st.mu.Lock() + defer st.mu.Unlock() + for ti := range st.Board { + t := &st.Board[ti] + if t.Status != "open" { + continue + } + if !e.boardDepsDoneLocked(st, t.ID) { + continue + } + idx, s := e.boardStepLocked(st, t.ID) + if idx < 0 { + continue + } + t.Status = "claimed" + t.ClaimedBy = s.Role + t.Attempts++ + t.LeaseExpires = time.Now().Add(time.Duration(leaseTTL) * time.Second) + _ = st.save() + (&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, StepID: t.ID, Role: s.Role, + Text: fmt.Sprintf("task_board：%s 认领 %s（租约 %ds，第 %d 次）", s.Role, t.ID, leaseTTL, t.Attempts)}) + return idx, true + } + return -1, false +} + +// boardSkipDeadlocked 把依赖已死（上游 failed/skipped）的 open 任务级联置 +// skipped；返回是否有变化（有则继续循环级联）。 +func (e *SquadEngine) boardSkipDeadlocked(st *RunState) bool { + st.mu.Lock() + changed := false + for ti := range st.Board { + t := &st.Board[ti] + if t.Status != "open" { + continue + } + if e.boardDepsDeadLocked(st, t.ID) { + t.Status = "skipped" + if idx, s := e.boardStepLocked(st, t.ID); idx >= 0 { + s.Status = "skipped" + s.Error = boardSkipReason + st.Plan.Workflow.Steps[idx] = *s + } + changed = true + } + } + if changed { + _ = st.save() + } + st.mu.Unlock() + return changed +} + +// boardRunTask 执行已认领任务：过门 → runOneStep。 +func (e *SquadEngine) boardRunTask(ctx context.Context, drv *flowDriver, st *RunState, idx int) (*RunResult, *HitlGate, error) { + out, gate, err := drv.gateBeforeRun(ctx, st, idx) + if err != nil || gate != nil || out != nil { + return out, gate, err + } + return drv.runOneStep(ctx, st, idx) +} + +// boardSettleTask 任务结算：步骤 completed → done；failed（非关键降级）→ failed。 +func (e *SquadEngine) boardSettleTask(st *RunState, idx int) { + step := st.Plan.Workflow.Steps[idx] + status := "done" + if step.Status == "failed" { + status = "failed" + } + st.mu.Lock() + for i := range st.Board { + if st.Board[i].ID == step.ID { + st.Board[i].Status = status + st.Board[i].LeaseExpires = time.Time{} + } + } + _ = st.save() + st.mu.Unlock() + (&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, StepID: step.ID, Role: step.Role, + Text: fmt.Sprintf("task_board：%s → %s", step.ID, status)}) +} + +// boardFinish 全部任务收口后走标准 finalize（交付物/质检/降级与图拓扑同面）。 +// 仍有 open/claimed 任务 = 依赖不可满足，fail-closed 不静默收口。 +func (e *SquadEngine) boardFinish(ctx context.Context, drv *flowDriver, st *RunState) (*RunResult, error) { + done, total := 0, len(st.Board) + for _, t := range st.Board { + switch t.Status { + case "done": + done++ + case "open", "claimed": + return e.driverFail(st, fmt.Sprintf("task_board: 任务 %s 无法收口（依赖不可满足）", t.ID)), nil + } + } + drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("task_board：收口（done %d/%d）", done, total)}) + out, gate, err := drv.finalize(ctx, st) + if err != nil { + return nil, err + } + if gate != nil { + return heldResult(st, gate), nil + } + return out, nil +} + +// boardStep 按任务 ID 找步骤（只读副本）。 +func (e *SquadEngine) boardStep(st *RunState, id string) (int, *StepSpec) { + st.mu.Lock() + defer st.mu.Unlock() + return e.boardStepLocked(st, id) +} + +// boardStepLocked 按任务 ID 找步骤（调用方持锁）。返回下标与副本指针。 +func (e *SquadEngine) boardStepLocked(st *RunState, id string) (int, *StepSpec) { + for i := range st.Plan.Workflow.Steps { + if st.Plan.Workflow.Steps[i].ID == id { + s := st.Plan.Workflow.Steps[i] + return i, &s + } + } + return -1, nil +} + +// boardDepsDoneLocked 依赖门：Needs 全部 done（调用方持锁）。 +func (e *SquadEngine) boardDepsDoneLocked(st *RunState, id string) bool { + needs := e.boardNeedsLocked(st, id) + for _, dep := range needs { + found := false + for i := range st.Board { + if st.Board[i].ID == dep { + found = true + if st.Board[i].Status != "done" { + return false + } + } + } + if !found { + return false // 依赖不在板上 = fail-closed 不认领 + } + } + return true +} + +// boardDepsDeadLocked 依赖已死：任一 Need 为 failed/skipped（级联跳过依据）。 +func (e *SquadEngine) boardDepsDeadLocked(st *RunState, id string) bool { + for _, dep := range e.boardNeedsLocked(st, id) { + for i := range st.Board { + if st.Board[i].ID == dep && (st.Board[i].Status == "failed" || st.Board[i].Status == "skipped") { + return true + } + } + } + return false +} + +func (e *SquadEngine) boardNeedsLocked(st *RunState, id string) []string { + for i := range st.Plan.Workflow.Steps { + if st.Plan.Workflow.Steps[i].ID == id { + return st.Plan.Workflow.Steps[i].Needs + } + } + return nil +} diff --git a/squad/topology_conformance_test.go b/squad/topology_conformance_test.go index d936f3b..64a36ab 100644 --- a/squad/topology_conformance_test.go +++ b/squad/topology_conformance_test.go @@ -384,8 +384,8 @@ func TestTopologyValidationFailsClosed(t *testing.T) { {"未注册名", "bogus_topology", "未知拓扑"}, {"primitive 单跑", TopologyMailbox, "跨拓扑原语"}, {"primitive 单跑 handoff", TopologyHandoff, "跨拓扑原语"}, - // review/group_chat 驱动已于 WP3 点亮；task_board 属 WP4，仍未点亮。 - {"driver 未点亮 task_board", TopologyTaskBoard, "驱动未注册"}, + // review/group_chat（WP3）与 task_board（WP4）驱动均已点亮； + // 12/12 受理面由 TestTopologyConformance_allTwelve 汇总钉住。 } for _, tc := range cases { engine := newTopologyEngine(t, mockScriptPlanner( diff --git a/squad/wp4_conformance_test.go b/squad/wp4_conformance_test.go new file mode 100644 index 0000000..171dc2a --- /dev/null +++ b/squad/wp4_conformance_test.go @@ -0,0 +1,355 @@ +package squad + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// wp4_conformance_test.go — Q2 WP4 验收：task_board 拓扑（claim/lease/complete +// 原语 + 依赖门 + 孤儿租约再武装）与全 12 拓扑 conformance 汇总跑。 + +// ---- task_board ---- + +// 菱形依赖 a→(b,c)→d：认领序尊重依赖门，全部 done，标准 finalize 收口。 +func TestTopologyConformance_taskBoard(t *testing.T) { + be := &iterBackend{calls: map[string]int{}} + sink := &collectSink{} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "a", Role: "researcher", Title: "任务a", Targets: []string{"a.md"}}, + StepSpec{ID: "b", Role: "writer", Title: "任务b", Needs: []string{"a"}, Targets: []string{"b.md"}}, + StepSpec{ID: "c", Role: "reviewer", Title: "任务c", Needs: []string{"a"}, Targets: []string{"c.md"}}, + StepSpec{ID: "d", Role: "writer", Title: "任务d", Needs: []string{"b", "c"}, Targets: []string{"output.md"}}, + ), be, sink) + res, err := engine.Run(context.Background(), Goal{Text: "任务板"}, RunOptions{Topology: TopologyTaskBoard}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + // 认领序尊重依赖门：a 先于 b/c，d 最后（板上按步骤序认领，确定性）。 + var order []string + for _, ev := range sink.snapshot() { + if strings.Contains(ev.Text, "认领") { + order = append(order, ev.StepID) + } + } + want := []string{"a", "b", "c", "d"} + if strings.Join(order, ",") != strings.Join(want, ",") { + t.Fatalf("认领序 = %v，期望 %v", order, want) + } + st, _ := engine.loadRun(engine.runDir(res.RunID)) + for _, task := range st.Board { + if task.Status != "done" { + t.Fatalf("任务 %s = %s，期望 done", task.ID, task.Status) + } + } + if sink.countText("task_board：收口（done 4/4）") != 1 { + t.Fatal("缺收口事件") + } +} + +// 孤儿租约再武装：崩溃（ctx 取消）留下 claimed+过期租约 → Resume 重入 +// 再武装回 open、重新认领（Attempts=2）并跑完。 +func TestTaskBoardOrphanLeaseRearms(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + crashed := false + be := &iterBackend{calls: map[string]int{}, content: func(key string, _ int, _ string) []byte { + if key == "02_writer" && !crashed { + crashed = true + cancel() // 模拟进程死亡：runOneStep 在 ctx 检查处退出，步骤回到 pending + } + return nil + }} + sink := &collectSink{} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "t1", Role: "researcher", Title: "一", Targets: []string{"a.md"}}, + StepSpec{ID: "t2", Role: "writer", Title: "二", Needs: []string{"t1"}, Targets: []string{"output.md"}}, + ), be, sink) + res0, err := engine.Run(ctx, Goal{Text: "任务板"}, RunOptions{Topology: TopologyTaskBoard}) + if err != nil { + t.Fatal(err) + } + if res0.Status != RunCancelled { + t.Fatalf("取消后应为 RunCancelled：%s", res0.Status) + } + // 崩溃现场：t2 claimed、租约未过期；步骤在盘上是 running（setStep 只改 + // 内存，取消路径的 pending 未落盘）——进程死亡即此形态。测试把租约回拨 + // 成过期（孤儿）并把步骤置回 pending（模拟重启前的一致化），再走 Resume。 + st, err := engine.loadRun(engine.runDir(res0.RunID)) + if err != nil { + t.Fatal(err) + } + found := false + for i := range st.Board { + if st.Board[i].ID == "t2" { + found = true + if st.Board[i].Status != "claimed" || st.Board[i].Attempts != 1 { + t.Fatalf("崩溃现场 t2 = %+v", st.Board[i]) + } + st.Board[i].LeaseExpires = time.Now().Add(-time.Hour) // 租约过期 = 孤儿 + } + } + if !found { + t.Fatal("板上无 t2") + } + for i := range st.Plan.Workflow.Steps { + if st.Plan.Workflow.Steps[i].ID == "t2" && st.Plan.Workflow.Steps[i].Status == "running" { + st.Plan.Workflow.Steps[i].Status = "pending" + } + } + if err := st.save(); err != nil { + t.Fatal(err) + } + // 重入：孤儿租约再武装 → 重新认领 → 跑完收口。 + res, err := engine.Resume(context.Background(), st.Meta.ID) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("恢复后应完成：%s %v", res.Status, res.Error) + } + if sink.countText("孤儿租约再武装 t2") != 1 { + t.Fatal("缺孤儿租约再武装事件") + } + st2, _ := engine.loadRun(engine.runDir(res.RunID)) + for _, task := range st2.Board { + if task.ID == "t2" && task.Attempts != 2 { + t.Fatalf("再武装后 Attempts = %d，期望 2", task.Attempts) + } + } +} + +// 依赖门级联：非关键任务失败 → 下游 open 任务级联 skipped → 降级收口。 +func TestTaskBoardDependencyGateCascade(t *testing.T) { + be := &iterBackend{calls: map[string]int{}, content: func(key string, _ int, _ string) []byte { + return nil + }} + failBe := &failOnceBackend{inner: be, failKey: "02_writer"} + crit := false + sink := &collectSink{} + engine := newTopologyEngine(t, mockScriptPlanner( + StepSpec{ID: "t0", Role: "researcher", Title: "零", Targets: []string{"output.md"}}, + StepSpec{ID: "t1", Role: "writer", Title: "一", Needs: []string{"t0"}, Targets: []string{"b.md"}, Critical: &crit}, + StepSpec{ID: "t2", Role: "reviewer", Title: "二", Needs: []string{"t1"}, Targets: []string{"c.md"}}, + ), failBe, sink) + res, err := engine.Run(context.Background(), Goal{Text: "任务板"}, RunOptions{Topology: TopologyTaskBoard}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunDegraded { + t.Fatalf("非关键失败+级联跳过应降级收口：%s %v", res.Status, res.Error) + } + st, _ := engine.loadRun(engine.runDir(res.RunID)) + byID := map[string]BoardTask{} + for _, task := range st.Board { + byID[task.ID] = task + } + if byID["t0"].Status != "done" || byID["t1"].Status != "failed" || byID["t2"].Status != "skipped" { + t.Fatalf("板状态 = %+v", byID) + } + for _, s := range st.Plan.Workflow.Steps { + if s.ID == "t2" && (s.Status != "skipped" || s.Error != boardSkipReason) { + t.Fatalf("t2 = %s %q", s.Status, s.Error) + } + } +} + +// failOnceBackend 对指定目录恒定失败（非瞬态），其余委托 inner。 +type failOnceBackend struct { + inner Executor + failKey string +} + +func (b *failOnceBackend) Name() string { return b.inner.Name() } + +func (b *failOnceBackend) Execute(ctx context.Context, req ExecRequest) (*ExecResult, error) { + key := filepath.Base(filepath.Clean(req.WorkDir)) + if key == b.failKey { + return &ExecResult{Success: false, ErrorReason: "fatal: 拒绝执行"}, nil + } + return b.inner.Execute(ctx, req) +} + +// ---- 12/12 汇总 ---- + +// TestTopologyConformance_allTwelve 全 12 内置拓扑汇总跑：每个拓扑名一次 +// 脚本化最小运行（零模型零网络）。graph/mode/driver 十种断言完成收口； +// handoff/mailbox 两个 primitive 断言 fail-closed 拒绝单跑（其原语能力分别 +// 由 TestTopologyConformance_handoff 与 WP2 mailbox 测试钉住）。 +func TestTopologyConformance_allTwelve(t *testing.T) { + specs := BuiltinTopologies() + if len(specs) != 12 { + t.Fatalf("内置拓扑 = %d，期望 12", len(specs)) + } + for _, spec := range specs { + spec := spec + t.Run(spec.Name, func(t *testing.T) { + switch spec.Name { + case TopologySequential: + runAgg(t, spec.Name, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}}, + ), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunCompleted) + case TopologyParallel: + runAgg(t, spec.Name, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "甲", Parallel: true, Targets: []string{"a.md"}}, + StepSpec{ID: "s2", Role: "writer", Title: "乙", Parallel: true, Targets: []string{"output.md"}}, + ), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunCompleted) + case TopologyLoop: + runAgg(t, spec.Name, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "循环体", Targets: []string{"output.md", "fields.json"}, + Loop: &LoopSpec{MaxIterations: 2, UntilField: "done", UntilEquals: "yes"}}, + ), &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 + }}, RunOptions{}, RunCompleted) + case TopologyRouter: + runAgg(t, spec.Name, mockScriptPlanner(routerSteps()...), + &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 + }}, RunOptions{}, RunCompleted) + case TopologyEvaluatorOptimizer: + runAgg(t, spec.Name, 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"}}}, + ), &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{"verdict": "pass"}) + } + return fieldsJSON(map[string]string{"verdict": "fail"}) + } + return nil + }}, RunOptions{}, RunCompleted) + case TopologyLongRunning: + runAgg(t, spec.Name, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}}, + ), &iterBackend{calls: map[string]int{}}, + RunOptions{Deadline: time.Now().Add(5 * time.Second)}, RunCompleted) + case TopologyLeader: + b := &leaderScriptBackend{ + decisions: []string{delegateJSON("writer", "撰写简报"), closeJSON}, + script: map[string]map[string]string{"01_writer": {"output.md": goodReport}}, + } + engine := newLeaderEngine(t, b) + res, err := engine.Run(context.Background(), Goal{Text: "汇总"}, RunOptions{Topology: TopologyLeader}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + case TopologyReview: + planner := mockScriptPlanner( + StepSpec{ID: "f1", Role: "researcher", Title: "评审", Targets: []string{"findings.json"}}, + ) + planner.Agents = []RoleSpec{ + {Role: "researcher", Name: "评审", Backend: "agg-review"}, + {Role: "verifier", Name: "核实", Backend: "agg-review"}, + } + engine := newAggEngine(t, planner, &aggReviewBackend{}, []RoleSpec{ + {Role: "researcher", Name: "评审", Backend: "agg-review"}, + {Role: "verifier", Name: "核实", Backend: "agg-review"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "评审"}, + RunOptions{Topology: TopologyReview, ReviewTarget: "材料"}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + data, _ := os.ReadFile(filepath.Join(res.WorkDir, "output.md")) + if !strings.Contains(string(data), "已丢弃 0 条未验证") || !strings.Contains(string(data), "样例问题") { + t.Fatalf("报告 = %s", data) + } + case TopologyGroupChat: + engine := newAggEngine(t, mockScriptPlanner( + StepSpec{ID: "p1", Role: "researcher", Title: "甲"}, + StepSpec{ID: "p2", Role: "writer", Title: "乙"}, + ), &gcScriptBackend{}, []RoleSpec{ + {Role: "researcher", Name: "甲", Backend: "gc-script"}, + {Role: "writer", Name: "乙", Backend: "gc-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "群聊"}, + RunOptions{Topology: TopologyGroupChat, GroupChatMaxTurns: 2}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + case TopologyTaskBoard: + runAgg(t, spec.Name, mockScriptPlanner( + StepSpec{ID: "t1", Role: "researcher", Title: "一", Targets: []string{"a.md"}}, + StepSpec{ID: "t2", Role: "writer", Title: "二", Needs: []string{"t1"}, Targets: []string{"output.md"}}, + ), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunCompleted) + case TopologyHandoff, TopologyMailbox: + // primitive：不可单独作为运行拓扑（fail-closed）；原语能力由 + // 专项 conformance 钉住（handoff/mailbox 测试）。 + runAgg(t, spec.Name, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}}, + ), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunFailed) + default: + t.Fatalf("未覆盖的拓扑 %q", spec.Name) + } + }) + } +} + +// runAgg 汇总跑的最小引擎：graph 类拓扑共用 iter 三角色装配。 +func runAgg(t *testing.T, topo string, planner Planner, be Executor, opts RunOptions, want RunStatus) { + t.Helper() + opts.Topology = topo + engine := newTopologyEngine(t, planner, be, nil) + res, err := engine.Run(context.Background(), Goal{Text: "汇总跑 " + topo}, opts) + if err != nil { + t.Fatal(err) + } + if res.Status != want { + t.Fatalf("status = %s err = %q，期望 %s", res.Status, res.Error, want) + } +} + +// aggReviewBackend 汇总跑用的最小 review 后端：finder 出一条发现，verify 判 PASS。 +type aggReviewBackend struct{} + +func (b *aggReviewBackend) Name() string { return "agg-review" } + +func (b *aggReviewBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) { + key := filepath.Base(filepath.Clean(req.WorkDir)) + if strings.HasPrefix(key, "verify_") { + return writeOne(req.WorkDir, "verdict.md", "核对成立。\nVERDICT: PASS"), nil + } + return writeOne(req.WorkDir, "findings.json", + `[{"location":"第1段","severity":"high","issue":"样例问题","evidence":"样例证据"}]`), nil +} + +// newAggEngine 自定义角色表的汇总引擎（backend 实例名 = be.Name()）。 +func newAggEngine(t *testing.T, planner Planner, be Executor, roles []RoleSpec) *SquadEngine { + t.Helper() + dir := t.TempDir() + reg := NewRoleRegistry() + for _, r := range roles { + if err := reg.Register(r); err != nil { + t.Fatal(err) + } + } + backends := NewBackendResolver() + backends.RegisterInstance(be.Name(), be) + engine, err := NewEngine(Options{DataDir: dir, Registry: reg, Planner: planner, Backends: backends, Sink: NopSink{}}) + if err != nil { + t.Fatal(err) + } + return engine +} -- 2.54.0