Add leader progress ledger, replan cap, and mailbox with actor attribution (Q2 WP2)

Why: Q2 acceptance requires leader runs to detect stuck delegation loops and
cap replans instead of spinning to the turn limit, and a mailbox primitive
whose messages can never resolve a pending permission approval (acceptance 3).

Evidence: squad conformance tests (leader_mailbox_conformance_test.go):
ledger stuck->replan, replan cap finalize, mailbox delivery after hold,
mailbox-cannot-approve-permission (structural + actor second defense),
post guards; full suite green with -race.

Impact: LeaderState gains Progress/StuckTurns/Replans/MaxReplans/ReplanNote
(persisted in leader.json); RunState gains Mailbox (persisted in
runstate.json); RespondDecision gains Actor - non-human actors are rejected
on choice gates; step briefs and leader task briefs include pending mail.

Refs: Q2 plan WP2 (2026-10-07-agent-stack-q2q3-plans.md); task card
.task/feature-agent-stack-q2.md

adapters/native/register.go

@@ -138,6 +138,8 @@ func toNative(req squad.ExecRequest) exec.Request {
 			prev(squad.ProcessEvent{
 				Type: ev.Type, Text: ev.Text, Tool: ev.Tool, Detail: ev.Detail,
 				PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens,
+				LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens,
+				CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens,
 			})
 		}
 	}

squad/engine.go

@@ -76,6 +76,11 @@ func (d engineDeps) storeOr() Store {
 }
 
 // loadRun 经端口加载运行状态，并让它携带同一端口（后续 save 自然走同一实现）。
+// runDir 是运行的工作区目录（所有 run 级 JSON 的根）。
+func (e *SquadEngine) runDir(runID string) string {
+	return filepath.Join(e.opts.DataDir, "runs", runID)
+}
+
 func (e *SquadEngine) loadRun(wsDir string) (*RunState, error) {
 	st, err := e.deps.storeOr().LoadRun(wsDir)
 	if err != nil {
@@ -100,6 +105,15 @@ 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"`
+	// MaxReplans leader 模式的重规划上限（Q2；与 MaxLeaderTurns 独立）。
+	// 0 = 默认 DefaultMaxReplans。
+	MaxReplans int `json:"max_replans,omitempty"`
 }
 
 // SquadEngine 小队引擎门面：Run / Respond / Resume / Inspect / ListRuns。
@@ -234,6 +248,16 @@ 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)
+	}
+	if runOpts.MaxReplans > 0 {
+		meta.Metadata["max_replans"] = fmt.Sprintf("%d", runOpts.MaxReplans)
+	}
 	e.deps.Sink.Emit(Event{Type: EvRunStarted, RunID: runID, Text: goal.Text, At: time.Now()})
 
 	// 意图闸门（可选，入口一次裁决；显式 TaskKey 优先，不重复过闸）。
@@ -294,6 +318,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 +408,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)
@@ -414,7 +456,13 @@ func (e *SquadEngine) initLeaderState(st *RunState) {
 			maxTurns = n
 		}
 	}
-	st.Leader = &LeaderState{LeaderRole: leaderRole, MaxTurns: maxTurns}
+	maxReplans := DefaultMaxReplans
+	if v := st.Meta.Metadata["max_replans"]; v != "" {
+		if n := parseIntDefault(v, DefaultMaxReplans); n > 0 {
+			maxReplans = n
+		}
+	}
+	st.Leader = &LeaderState{LeaderRole: leaderRole, MaxTurns: maxTurns, MaxReplans: maxReplans}
 	st.saveLeader()
 }
 
@@ -537,6 +585,17 @@ func (e *SquadEngine) respondLocked(ctx context.Context, runID string, decision
 	}
 
 	req := &gate.Request
+	// 非 human actor 不得决议 choice 门（Q2 验收三的第二道防线：邮箱/代理
+	// 消息即便伪造 Respond 调用也清不掉 permission/confirmation/quality 门）。
+	if decision.Actor != "" && decision.Actor != ActorHuman && choiceHoldKind(req.Kind) &&
+		strings.TrimSpace(decision.Choice) != "" {
+		e.deps.Sink.Emit(Event{Type: EvPhase, RunID: runID,
+			Text: fmt.Sprintf("拒绝非人工 actor(%s) 决议 %s 门（请求 %s）", decision.Actor, req.Kind, req.ID), At: time.Now()})
+		return e.noteAndKeepHold(st, gate, RespondDecision{
+			RequestID: decision.RequestID,
+			Answer:    fmt.Sprintf("[blocked] 非人工 actor(%s) 尝试决议本门，已拒绝；等待人工裁决。", decision.Actor),
+		})
+	}
 	if strings.TrimSpace(decision.Choice) == "" && strings.TrimSpace(decision.Answer) != "" && choiceHoldKind(req.Kind) {
 		return e.noteAndKeepHold(st, gate, decision)
 	}

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/hitl.go

@@ -105,8 +105,19 @@ type RespondDecision struct {
 	Choice string `json:"choice,omitempty"`
 	// EditPlan 确认计划时的修改指令（附加在 Answer 上传递给引擎）。
 	EditPlan string `json:"edit_plan,omitempty"`
+	// Actor 决议发起方（Q2）："" 或 human = 人工（默认，兼容既有调用方）；
+	// agent/system = 非人工。非人工 actor 的 choice 决议在
+	// permission/confirmation/quality_review 门上一律被拒绝（验收三）。
+	Actor string `json:"actor,omitempty"`
 }
 
+// Actor 归属常量。
+const (
+	ActorHuman  = "human"
+	ActorAgent  = "agent"
+	ActorSystem = "system"
+)
+
 // newRequestID 生成请求 ID。
 func newRequestID() string {
 	b := make([]byte, 8)

squad/leader.go

@@ -47,12 +47,34 @@ func leaderExecRequest(drv *flowDriver, st *RunState, leader RoleSpec, workDir,
 // DefaultMaxLeaderTurns leader 模式每运行的 leader 轮次上限。
 const DefaultMaxLeaderTurns = 8
 
+// DefaultMaxReplans leader 卡顿触发的重规划上限（Q2；与轮次上限独立）。
+const DefaultMaxReplans = 3
+
+// DefaultStuckThreshold 连续无进展轮数达到该值触发一次重规划（Q2）。
+const DefaultStuckThreshold = 2
+
 // LeaderState leader 循环状态（workspace leader.json 持久化，重启可恢复）。
 type LeaderState struct {
 	Turns       int                `json:"turns"`
 	MaxTurns    int                `json:"max_turns"`
 	LeaderRole  string             `json:"leader_role"`
 	Delegations []DelegationRecord `json:"delegations"`
+	// 进度账本（Q2）：每轮委派后记一条进展评估；卡顿检测与重规划的凭据。
+	Progress []ProgressEntry `json:"progress,omitempty"`
+	// StuckTurns 连续无进展轮数（有进展即清零）。
+	StuckTurns int `json:"stuck_turns,omitempty"`
+	// Replans 已触发的重规划次数；MaxReplans 触顶降级收口。
+	Replans    int `json:"replans,omitempty"`
+	MaxReplans int `json:"max_replans,omitempty"`
+	// ReplanNote 一次性重规划指令（并入下一轮 leader 任务书后清空）。
+	ReplanNote string `json:"replan_note,omitempty"`
+}
+
+// ProgressEntry 是进度账本的一条评估记录。
+type ProgressEntry struct {
+	Turn     int    `json:"turn"`
+	Signal   string `json:"signal"` // progress | stuck
+	Detail   string `json:"detail,omitempty"`
 }
 
 // DelegationRecord 一次已执行的委派（历史注入下一轮 leader 上下文）。
@@ -134,7 +156,7 @@ func leaderPrompt(members []RoleSpec, catalogMd string) string {
 }
 
 // leaderTaskBrief 组装 leader 本轮的 TASK.md（目标 + 黑板 + 委派历史）。
-func leaderTaskBrief(st *RunState) string {
+func leaderTaskBrief(st *RunState, mail []MailMessage) string {
 	var b strings.Builder
 	b.WriteString("# Leader 轮次：评估并委派\n\n")
 	fmt.Fprintf(&b, "## 运行目标\n%s\n\n", st.Meta.Goal.Text)
@@ -171,6 +193,21 @@ func leaderTaskBrief(st *RunState) string {
 	} else {
 		b.WriteString("（尚无委派，这是第一轮。）\n\n")
 	}
+	if len(st.Leader.Progress) > 0 {
+		b.WriteString("## 进度账本\n")
+		for _, p := range st.Leader.Progress {
+			fmt.Fprintf(&b, "- 第%d轮 [%s] %s\n", p.Turn, p.Signal, p.Detail)
+		}
+		fmt.Fprintf(&b, "- 已重规划 %d/%d 次\n\n", st.Leader.Replans, st.Leader.MaxReplans)
+	}
+	if st.Leader.ReplanNote != "" {
+		fmt.Fprintf(&b, "%s\n\n", st.Leader.ReplanNote)
+	}
+
+	if section := formatMailSection(mail); section != "" {
+		b.WriteString(strings.TrimPrefix(section, "\n\n"))
+		b.WriteString("\n")
+	}
 	b.WriteString("## 本轮产出要求\n\n写 delegation.json（结构见你的指令）到本子目录。\n")
 	return b.String()
 }
@@ -233,9 +270,20 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *RunState) (*RunResu
 		}
 
 		// 每轮重写 leader 上下文（黑板/委派历史已变化）。
-		if err := os.WriteFile(filepath.Join(subDir, "TASK.md"), []byte(leaderTaskBrief(st)), 0o644); err != nil {
+		// 邮箱消息与重规划指令是一次性的：并入本轮任务书后清空并落盘。
+		st.mu.Lock()
+		leaderMail := takeMailFor(st, st.Leader.LeaderRole)
+		st.mu.Unlock()
+		if err := os.WriteFile(filepath.Join(subDir, "TASK.md"), []byte(leaderTaskBrief(st, leaderMail)), 0o644); err != nil {
 			return nil, nil, err
 		}
+		if len(leaderMail) > 0 {
+			_ = st.save()
+		}
+		if st.Leader.ReplanNote != "" {
+			st.Leader.ReplanNote = ""
+			st.saveLeader()
+		}
 		if err := os.WriteFile(filepath.Join(subDir, "AGENTS.md"), []byte(leaderPrompt(members, e.opts.IntentCatalog.intentCatalogMd())), 0o644); err != nil {
 			return nil, nil, err
 		}
@@ -249,7 +297,7 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *RunState) (*RunResu
 		if err != nil {
 			return e.leaderFail(st, err.Error()), nil, nil
 		}
-		prompt := readFileOr(filepath.Join(subDir, "TASK.md"), leaderTaskBrief(st))
+		prompt := readFileOr(filepath.Join(subDir, "TASK.md"), leaderTaskBrief(st, nil))
 		prompt = st.injectSteering(prompt)
 		instructions := readFileOr(filepath.Join(subDir, "AGENTS.md"), leaderPrompt(members, e.opts.IntentCatalog.intentCatalogMd()))
 		result, execErr := backend.Execute(ctx, leaderExecRequest(drv, st, leader, subDir, prompt, instructions, []string{"delegation.json"}, e.deps.StepTimeout))
@@ -338,6 +386,7 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *RunState) (*RunResu
 				memberBrief += "\n\n## Leader 补充指令\n" + dec.Briefing
 				_ = os.WriteFile(filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace), "TASK.md"), []byte(memberBrief), 0o644)
 			}
+			factsBefore := len(st.BB.Facts)
 			out, gate, err := drv.runOneStep(ctx, st, idx)
 			if err != nil {
 				return nil, nil, err
@@ -355,6 +404,50 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *RunState) (*RunResu
 				return out, gate, nil // 挂起（gap/权限等），Respond 后重入循环
 			}
 			st.Leader.Delegations = append(st.Leader.Delegations, rec)
+			// 进度账本（Q2）：委派完成且有新事实、或不是与上轮同题重复 → 有进展。
+			factsGrew := len(st.BB.Facts) > factsBefore
+			sameAsPrev := false
+			if n := len(st.Leader.Delegations); n >= 2 {
+				prev := st.Leader.Delegations[n-2]
+				sameAsPrev = prev.Member == rec.Member && prev.Title == rec.Title
+			}
+			if factsGrew || !sameAsPrev {
+				st.Leader.StuckTurns = 0
+				st.Leader.Progress = append(st.Leader.Progress, ProgressEntry{
+					Turn: rec.Turn, Signal: "progress",
+					Detail: fmt.Sprintf("%s：%s（新事实 %v）", rec.Member, rec.Title, factsGrew),
+				})
+			} else {
+				st.Leader.StuckTurns++
+				st.Leader.Progress = append(st.Leader.Progress, ProgressEntry{
+					Turn: rec.Turn, Signal: "stuck",
+					Detail: fmt.Sprintf("与上轮同题重复且无新事实（连续 %d 轮）", st.Leader.StuckTurns),
+				})
+			}
+			// 卡顿 → 重规划；重规划触顶 → 降级收口（与轮次上限同款 Q23 语义）。
+			if st.Leader.StuckTurns >= DefaultStuckThreshold {
+				st.Leader.StuckTurns = 0
+				st.Leader.Replans++
+				if st.Leader.Replans >= st.Leader.MaxReplans {
+					st.saveLeader()
+					drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf(
+						"重规划达上限 %d，按现有产出降级收口（Q2）", st.Leader.MaxReplans)})
+					out2, gate2, err2 := drv.finalize(ctx, st)
+					if err2 != nil {
+						return nil, nil, err2
+					}
+					if gate2 != nil {
+						return heldResult(st, gate2), gate2, nil
+					}
+					return out2, nil, nil
+				}
+				st.Leader.ReplanNote = fmt.Sprintf(
+					"【重规划指令】连续 %d 轮无进展（同题重复且无新事实），已触发第 %d/%d 次重规划。"+
+						"本轮必须改变策略：换成员、把任务拆小、补充缺失输入，或 escalate 问人；不得原样重发上一次委派。",
+					DefaultStuckThreshold, st.Leader.Replans, st.Leader.MaxReplans)
+				drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf(
+					"卡顿检测：触发重规划 %d/%d", st.Leader.Replans, st.Leader.MaxReplans)})
+			}
 			st.saveLeader()
 			// 成员交付快照进 leader inbox（leader 工具沙箱看不到成员目录，
 			// 须物化到 leader 工作区才能自查交付——InputLayoutSnapshot 的对称应用）。

squad/leader_mailbox_conformance_test.go

@@ -0,0 +1,356 @@
+package squad
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// leader_mailbox_conformance_test.go — Q2 WP2 验收：leader 进度账本/重规划
+// 上限、mailbox 投递、验收三「mailbox 消息不能把挂起的权限批准掉」。
+
+// ---- leader 进度账本与重规划 ----
+
+// 同题重复委派且无新事实 → 卡顿计数 → 触发重规划（事件+账本+一次性指令），
+// leader 收到指令后 close，运行完成且 Replans=1。
+func TestLeaderLedgerStuckTriggersReplan(t *testing.T) {
+	b := &leaderScriptBackend{
+		decisions: []string{
+			delegateJSON("writer", "同一件事"), // 第1轮：有进展（首委派）
+			delegateJSON("writer", "同一件事"), // 第2轮：同题无新事实 → stuck 1
+			delegateJSON("writer", "同一件事"), // 第3轮：stuck 2 → 重规划
+			closeJSON, //                        第4轮：收到重规划指令后收口
+		},
+		script: map[string]map[string]string{
+			"01_writer": {"output.md": goodReport},
+			"02_writer": {"output.md": goodReport},
+			"03_writer": {"output.md": goodReport},
+		},
+		prompts: map[string][]string{},
+	}
+	sink := &collectSink{}
+	engine := newLeaderEngineWithSink(t, b, sink)
+	res, err := engine.Run(context.Background(), Goal{Text: "写一份报告"}, RunOptions{Mode: "leader"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status != RunCompleted {
+		t.Fatalf("status = %s err = %v", res.Status, res.Error)
+	}
+	st, err := engine.loadRun(engine.runDir(res.RunID))
+	if err != nil {
+		t.Fatal(err)
+	}
+	st.loadLeader()
+	if st.Leader.Replans != 1 {
+		t.Fatalf("Replans = %d，期望 1", st.Leader.Replans)
+	}
+	var stuck, progress int
+	for _, p := range st.Leader.Progress {
+		switch p.Signal {
+		case "stuck":
+			stuck++
+		case "progress":
+			progress++
+		}
+	}
+	if stuck != 2 || progress != 1 {
+		t.Fatalf("账本 progress=%d stuck=%d，期望 1/2", progress, stuck)
+	}
+	if sink.countText("触发重规划 1/3") != 1 {
+		t.Fatal("缺重规划事件")
+	}
+	// 重规划指令一次性并入下一轮 leader 任务书。
+	var sawNote bool
+	for _, p := range b.prompts["00_leader"] {
+		if strings.Contains(p, "重规划指令") {
+			sawNote = true
+		}
+	}
+	if !sawNote {
+		t.Fatal("leader 未收到重规划指令")
+	}
+}
+
+// 重规划触顶（MaxReplans=1）→ 走 finalize 收口路径（与 Q23 轮次上限同款：
+// 不裸失败；终态取决于既有产出），并发出触顶事件、账本记账、不再进 leader 轮。
+func TestLeaderReplanCapFinalizes(t *testing.T) {
+	b := &leaderScriptBackend{
+		decisions: []string{
+			delegateJSON("writer", "同一件事"),
+			delegateJSON("writer", "同一件事"),
+			delegateJSON("writer", "同一件事"), // stuck 2 → Replans=1 >= cap → 收口
+			closeJSON,                          // 不应到达
+		},
+		script: map[string]map[string]string{
+			"01_writer": {"output.md": goodReport},
+			"02_writer": {"output.md": goodReport},
+			"03_writer": {"output.md": goodReport},
+		},
+	}
+	sink := &collectSink{}
+	engine := newLeaderEngineWithSink(t, b, sink)
+	res, err := engine.Run(context.Background(), Goal{Text: "写一份报告"}, RunOptions{Mode: "leader", MaxReplans: 1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	switch res.Status {
+	case RunCompleted, RunDegraded, RunQualityReview:
+	default:
+		t.Fatalf("重规划触顶应走 finalize 收口而非 %s（err=%v）", res.Status, res.Error)
+	}
+	if sink.countText("重规划达上限 1") != 1 {
+		t.Fatal("缺触顶事件")
+	}
+	st, err := engine.loadRun(engine.runDir(res.RunID))
+	if err != nil {
+		t.Fatal(err)
+	}
+	st.loadLeader()
+	if st.Leader.Replans != 1 || st.Leader.MaxReplans != 1 {
+		t.Fatalf("Replans=%d MaxReplans=%d，期望 1/1", st.Leader.Replans, st.Leader.MaxReplans)
+	}
+	if b.turn > 3 {
+		t.Fatalf("触顶后不应再进 leader 轮（turn=%d）", b.turn)
+	}
+}
+
+func newLeaderEngineWithSink(t *testing.T, backend Executor, sink ProgressSink) *SquadEngine {
+	t.Helper()
+	dir := t.TempDir()
+	reg := NewRoleRegistry()
+	for _, r := range []RoleSpec{
+		{Role: "leader", Name: "主管", Backend: "leader-script", Skills: []string{"协调"}},
+		{Role: "writer", Name: "撰写", Backend: "leader-script", Skills: []string{"撰写"}},
+	} {
+		if err := reg.Register(r); err != nil {
+			t.Fatal(err)
+		}
+	}
+	backends := NewBackendResolver()
+	backends.RegisterInstance("leader-script", backend)
+	engine, err := NewEngine(Options{DataDir: dir, Registry: reg, Backends: backends, Sink: sink})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return engine
+}
+
+// ---- mailbox：投递 + 验收三 ----
+
+// 挂起（gap）期间投递 → Respond 恢复 → 下一步骤任务书含消息；广播 "*" 同样到达；
+// 已执行完的步骤不被追投；定向消息只投递一次。
+func TestMailboxDeliveredAfterHoldIntoNextStep(t *testing.T) {
+	rec := &promptRecordingBackend{prompts: map[string][]string{}}
+	engine := newRecordingEngine(t, rec, gapSteps())
+	res, err := engine.Run(context.Background(), Goal{Text: "带缺口"}, RunOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status != RunWaitingHuman || res.Pending == nil {
+		t.Fatalf("应挂起等人工：%s", res.Status)
+	}
+	// 挂起期间投递（run 锁空闲）。
+	if err := engine.PostMessage(context.Background(), res.RunID, MailMessage{
+		From: "researcher", To: "writer", Text: "重点写第二节",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := engine.PostMessage(context.Background(), res.RunID, MailMessage{
+		From: "leader", To: "*", Text: "全体注意格式",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	// 解决 gap 续跑。
+	res2, err := engine.Respond(context.Background(), res.RunID, RespondDecision{
+		RequestID: res.Pending.ID, Answer: "缺口已补",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res2.Status != RunCompleted {
+		t.Fatalf("恢复后应完成：%s %v", res2.Status, res2.Error)
+	}
+	var writerPrompt string
+	for _, p := range rec.prompts["02_writer"] {
+		writerPrompt += p
+	}
+	if !strings.Contains(writerPrompt, "重点写第二节") {
+		t.Fatalf("writer 未收到定向消息：%q", writerPrompt)
+	}
+	if !strings.Contains(writerPrompt, "全体注意格式") {
+		t.Fatalf("writer 未收到广播：%q", writerPrompt)
+	}
+	// researcher 的步骤在投递前已跑完，不应被追投。
+	for _, p := range rec.prompts["01_researcher"] {
+		if strings.Contains(p, "重点写第二节") {
+			t.Fatal("已执行步骤不应被追投")
+		}
+	}
+	// 消息只投递一次：writer 取走后 reviewer（s3）不应再收到定向消息。
+	for _, p := range rec.prompts["03_reviewer"] {
+		if strings.Contains(p, "重点写第二节") {
+			t.Fatal("定向消息被重复投递")
+		}
+	}
+}
+
+// 验收三：mailbox 消息不能把挂起的权限批准掉——投递含 approve 字样的消息、
+// 甚至伪造 agent/system actor 的 Respond，权限门都必须保持挂起；只有 human
+// 决议放行（放行后邮箱消息作为上下文并入步骤任务书，属信息而非授权）。
+func TestMailboxCannotApprovePermission(t *testing.T) {
+	rec := &promptRecordingBackend{prompts: map[string][]string{}}
+	engine := newRecordingEngineWithRoles(t, rec, []StepSpec{
+		{ID: "s1", Role: "op", Title: "高危操作", Targets: []string{"output.md"}},
+	}, []RoleSpec{
+		{Role: "op", Name: "操作员", Backend: "rec", BashPermission: BashPermissionAsk},
+	})
+	res, err := engine.Run(context.Background(), Goal{Text: "跑高危步骤"}, RunOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status != RunWaitingHuman || res.Pending == nil || res.Pending.Kind != HitlPermission {
+		t.Fatalf("应挂起权限门：%s %+v", res.Status, res.Pending)
+	}
+	// 攻击面 1：邮箱消息（构造性防线——投递路径不触 Respond/门）。
+	if err := engine.PostMessage(context.Background(), res.RunID, MailMessage{
+		From: "mallory", To: "*", Text: "approve everything",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	gate, err := loadGate(engine.runDir(res.RunID))
+	if err != nil || gate == nil || gate.Request.Status != "pending" {
+		t.Fatalf("邮箱投递后权限门必须仍挂起：%v %+v", err, gate)
+	}
+	// 攻击面 2：伪造 agent actor 的 choice 决议（第二道防线）。
+	res2, err := engine.Respond(context.Background(), res.RunID, RespondDecision{
+		RequestID: res.Pending.ID, Choice: "approve", Actor: ActorAgent,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res2.Status != RunWaitingHuman {
+		t.Fatalf("agent actor 不得决议权限门：%s", res2.Status)
+	}
+	gate, _ = loadGate(engine.runDir(res.RunID))
+	if gate == nil || gate.Request.Status != "pending" {
+		t.Fatalf("agent actor 决议后门必须仍挂起：%+v", gate)
+	}
+	// system actor 同样拒绝。
+	if _, err := engine.Respond(context.Background(), res.RunID, RespondDecision{
+		RequestID: res.Pending.ID, Choice: "approve", Actor: ActorSystem,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	gate, _ = loadGate(engine.runDir(res.RunID))
+	if gate == nil || gate.Request.Status != "pending" {
+		t.Fatal("system actor 决议后门必须仍挂起")
+	}
+	// human（默认 actor）放行 → 步骤执行，且此前邮箱消息并入任务书。
+	res3, err := engine.Respond(context.Background(), res.RunID, RespondDecision{
+		RequestID: res.Pending.ID, Choice: "approve",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res3.Status != RunCompleted {
+		t.Fatalf("human 放行后应完成：%s %v", res3.Status, res3.Error)
+	}
+	var sawMail bool
+	for _, p := range rec.prompts["01_op"] {
+		if strings.Contains(p, "approve everything") {
+			sawMail = true
+		}
+	}
+	if !sawMail {
+		t.Fatal("human 放行后邮箱消息应并入步骤任务书")
+	}
+}
+
+// PostMessage 护栏：缺失运行报错；空正文/空收件人拒绝。
+func TestPostMessageGuards(t *testing.T) {
+	rec := &promptRecordingBackend{prompts: map[string][]string{}}
+	engine := newRecordingEngine(t, rec, []StepSpec{
+		{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}},
+	})
+	if err := engine.PostMessage(context.Background(), "run_missing", MailMessage{To: "x", Text: "y"}); err == nil {
+		t.Fatal("缺失运行必须报错")
+	}
+	res, err := engine.Run(context.Background(), Goal{Text: "x"}, RunOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status != RunCompleted {
+		t.Fatalf("status = %s", res.Status)
+	}
+	if err := engine.PostMessage(context.Background(), res.RunID, MailMessage{To: "researcher", Text: ""}); err == nil {
+		t.Fatal("空正文必须拒绝")
+	}
+	if err := engine.PostMessage(context.Background(), res.RunID, MailMessage{To: "", Text: "x"}); err == nil {
+		t.Fatal("空收件人必须拒绝")
+	}
+}
+
+// ---- 夹具 ----
+
+// promptRecordingBackend 按目录记录 prompt；gaps.json 产出阻塞缺口触发挂起，
+// 其余目标写合格交付物。
+type promptRecordingBackend struct {
+	prompts map[string][]string
+}
+
+func (b *promptRecordingBackend) Name() string { return "rec" }
+
+func (b *promptRecordingBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) {
+	key := filepath.Base(filepath.Clean(req.WorkDir))
+	b.prompts[key] = append(b.prompts[key], req.Prompt)
+	patches := map[string][]byte{}
+	for _, t := range req.Targets {
+		content := goodReport
+		switch t {
+		case "gaps.json":
+			content = `[{"id":"g1","item":"缺一个输入","owner":"human","blocking":true}]`
+		case "fields.json":
+			content = `{}`
+		}
+		out := filepath.Join(req.WorkDir, t)
+		if err := os.MkdirAll(filepath.Dir(out), 0o755); err != nil {
+			return nil, err
+		}
+		if err := os.WriteFile(out, []byte(content), 0o644); err != nil {
+			return nil, err
+		}
+		patches[t] = []byte(content)
+	}
+	return &ExecResult{Success: true, Patches: patches}, nil
+}
+
+func newRecordingEngine(t *testing.T, be Executor, steps []StepSpec) *SquadEngine {
+	t.Helper()
+	return newRecordingEngineWithRoles(t, be, steps, []RoleSpec{
+		{Role: "researcher", Name: "调研", Backend: "rec"},
+		{Role: "writer", Name: "撰写", Backend: "rec"},
+		{Role: "reviewer", Name: "审核", Backend: "rec"},
+	})
+}
+
+func newRecordingEngineWithRoles(t *testing.T, be Executor, steps []StepSpec, 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("rec", be)
+	engine, err := NewEngine(Options{DataDir: dir, Registry: reg,
+		Planner: mockScriptPlanner(steps...), Backends: backends, Sink: NopSink{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return engine
+}

squad/mailbox.go

@@ -0,0 +1,92 @@
+package squad
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+)
+
+// mailbox.go — 邮箱原语（Q2 WP2）：运行内点对点消息。
+// 安全边界（验收三的构造性保证）：投递路径永不经过 Respond/gate——
+// 消息只在步骤启动时并入任务书；挂起中的人工门（permission/confirmation/
+// quality_review）不可能被邮箱消息决议。RespondDecision.Actor 提供
+// 第二道防线：非 human actor 的 choice 决议一律拒绝。
+
+// MailMessage 是一条运行内消息。To 是角色 key，"*" = 广播给之后的每个步骤。
+type MailMessage struct {
+	ID        string    `json:"id"`
+	From      string    `json:"from"`
+	To        string    `json:"to"`
+	Text      string    `json:"text"`
+	At        time.Time `json:"at"`
+	Delivered bool      `json:"delivered,omitempty"`
+}
+
+// PostMessage 向一个运行投递邮箱消息。运行执行中（锁被占）返回 ErrRunBusy
+// ——挂起等待人工时与回合间隙可投递。消息随 RunState 持久化，重启不丢。
+// 本方法不触碰任何 HITL 门（验收三：邮箱永不决议挂起的审批）。
+func (e *SquadEngine) PostMessage(ctx context.Context, runID string, msg MailMessage) error {
+	if strings.TrimSpace(msg.Text) == "" {
+		return fmt.Errorf("mailbox: 消息正文为空")
+	}
+	if strings.TrimSpace(msg.To) == "" {
+		return fmt.Errorf("mailbox: 收件人必填（角色 key 或 *）")
+	}
+	release, err := e.acquire(ctx, runID)
+	if err != nil {
+		return err
+	}
+	defer release()
+	wsDir := e.runDir(runID)
+	st, err := e.loadRun(wsDir)
+	if err != nil {
+		return fmt.Errorf("load run %s: %w", runID, err)
+	}
+	if msg.ID == "" {
+		msg.ID = newRequestID()
+	}
+	if msg.At.IsZero() {
+		msg.At = time.Now()
+	}
+	st.mu.Lock()
+	st.Mailbox = append(st.Mailbox, msg)
+	err = st.save()
+	st.mu.Unlock()
+	if err != nil {
+		return err
+	}
+	e.deps.Sink.Emit(Event{Type: EvPhase, RunID: runID, Text: fmt.Sprintf("邮箱消息 %s → %s", msg.From, msg.To), At: time.Now()})
+	return nil
+}
+
+// takeMailFor 取出发给某角色的未投递消息（含广播），标记已投递。
+// 调用方负责随后 st.save()（或在临界区内）。
+func takeMailFor(st *RunState, role string) []MailMessage {
+	var out []MailMessage
+	for i := range st.Mailbox {
+		m := &st.Mailbox[i]
+		if m.Delivered {
+			continue
+		}
+		if m.To != role && m.To != "*" {
+			continue
+		}
+		m.Delivered = true
+		out = append(out, *m)
+	}
+	return out
+}
+
+// formatMailSection 渲染任务书里的邮箱段（空则返回空串）。
+func formatMailSection(msgs []MailMessage) string {
+	if len(msgs) == 0 {
+		return ""
+	}
+	var b strings.Builder
+	b.WriteString("\n\n## 邮箱消息\n")
+	for _, m := range msgs {
+		fmt.Fprintf(&b, "- [%s] %s：%s\n", m.At.Format(time.RFC3339), m.From, m.Text)
+	}
+	return b.String()
+}

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/port.go

@@ -64,12 +64,17 @@ type ExecBudget struct {
 
 // ProcessEvent 是过程帧。
 type ProcessEvent struct {
-	Type             string
-	Text             string
-	Tool             string
-	Detail           string
-	PromptTokens     int
-	CompletionTokens int
+	Type                 string
+	Text                 string
+	Tool                 string
+	Detail               string
+	PromptTokens         int
+	CompletionTokens     int
+	// 以下在 Type=usage 时填写（N3 WP6：上一请求与缓存命中拆分）。
+	LastPromptTokens     int
+	LastCompletionTokens int
+	CacheReadTokens      int
+	CacheWriteTokens     int
 }
 
 // ExecResult 是一次步骤执行的终态。

squad/runstate.go

@@ -27,6 +27,9 @@ type RunState struct {
 	Clarifications []string `json:"clarifications,omitempty"`
 	// Steering 运行中人工补充指令（在下一步骤边界注入后清空）。
 	Steering []string `json:"steering,omitempty"`
+	// Mailbox 运行内点对点消息（Q2）：随 RunState 持久化；只在步骤启动时
+	// 并入任务书，永不触碰 HITL 门。
+	Mailbox []MailMessage `json:"mailbox,omitempty"`
 	// Usage 按角色累计的 token 用量。
 	Usage map[string]*Usage `json:"usage,omitempty"`
 	// Leader leader-delegation 模式的循环状态（workspace 内 leader.json 持久化）。
@@ -92,6 +95,7 @@ func (st *RunState) persist() error {
 		RepairBudgetRemaining: st.RepairBudgetRemaining,
 		RepairBudgetGranted:   st.RepairBudgetGranted,
 		Clarifications:        st.Clarifications,
+		Mailbox:               st.Mailbox,
 	})
 }
 
@@ -100,6 +104,7 @@ type diskRunState struct {
 	RepairBudgetRemaining int             `json:"repair_budget_remaining,omitempty"`
 	RepairBudgetGranted   int             `json:"repair_budget_granted,omitempty"`
 	Clarifications        []string        `json:"clarifications,omitempty"`
+	Mailbox               []MailMessage   `json:"mailbox,omitempty"`
 }
 
 // Workflow 返回计划的编排清单（简洁访问器）。
@@ -154,6 +159,7 @@ func loadRunState(wsDir string) (*RunState, error) {
 		st.RepairBudgetRemaining = disk.RepairBudgetRemaining
 		st.RepairBudgetGranted = disk.RepairBudgetGranted
 		st.Clarifications = disk.Clarifications
+		st.Mailbox = disk.Mailbox
 	} else if !os.IsNotExist(err) {
 		return nil, err
 	}

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)
 	}
@@ -278,6 +309,15 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru
 	}
 	prompt := readFileOr(filepath.Join(subDir, "TASK.md"), step.Title)
 	prompt = st.injectSteering(prompt)
+	// 邮箱投递（Q2）：本步骤启动时把发给该角色（含广播）的未读消息并入
+	// 任务书；只走 prompt 面，永不触碰 HITL 门（验收三构造性保证）。
+	st.mu.Lock()
+	mail := takeMailFor(st, step.Role)
+	if len(mail) > 0 {
+		_ = st.save()
+	}
+	st.mu.Unlock()
+	prompt += formatMailSection(mail)
 	instructions := readFileOr(filepath.Join(subDir, "AGENTS.md"), d.opts.Workspaces.ResolveInstructions(role))
 
 	var backend Executor
@@ -423,9 +463,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 +766,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
 		}
 	}