Add the topology surface with loop, router, and deadline mechanics (Q2 WP1)

Why: the completion doc Q2 requires twelve builtin topologies each with
a scripted-backend conformance; today the engine has only the
standard/leader mode string with no topology abstraction, no loop
re-arm (cycles are fail-closed by design), no in-graph routing, and no
run deadline.
Evidence: GOWORK=off go vet + go test -count=1 green across squad and
all three adapter modules; ten conformance/validation tests:
sequential, parallel, loop-until-field (exact iteration count +
persisted LoopRuns), loop-budget-exhausted (loop_exhausted event, run
still completes), router-selects-branch (loser skipped, run NOT
degraded), router-no-match fail-closed, evaluator-optimizer (draft
re-armed via loop body until verdict=pass), long-running deadline
(pending steps skipped, degraded closeout), unknown/primitive/
unregistered-driver fail-closed, builtin registry completeness (12).
Impact: additive — Topology/Deadline on RunOptions, Topology on
Blueprint, Loop/LoopRuns/Branches/Handoff on StepSpec; empty topology
keeps byte-identical standard behavior; MaxIterations is the total
iteration cap (first run included); router-skipped steps are exempt
from degraded derivation; runParallel now preserves the first specific
RunFailed error instead of re-wrapping via failStep; deadline metadata
uses RFC3339Nano (second-precision RFC3339 truncated sub-second
deadlines into instant expiry).
Refs: Agent-System-Documents#44 (Q2 plan WP1)

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)

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
 }
 

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 视为关键）。

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

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

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