Merge pull request 'Add the Store port with tiered checkpoints and centralize termination (Q1 WP4/WP6/WP7)' (#25) from feature/agent-stack-q1 into main

cmd/agent-squad/server/server.go

@@ -462,6 +462,8 @@ func (s *Server) StartWakeupSweeper(interval time.Duration) (stop func()) {
 				return
 			case <-t.C:
 				s.engine.SweepWakeups(context.Background())
+				// Q6：与事件唤醒并列，过期挂起按 ExpireAction 合成决策（此前是死代码）。
+				s.engine.SweepExpired(context.Background(), time.Now())
 			}
 		}
 	}()

squad/checkpoint_test.go

@@ -0,0 +1,113 @@
+package squad
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// TestCheckpointFilesAtBoundaries：sync 默认档——每个 superstep 边界有快照，
+// latest 指向最后一层且图指纹与重编译一致（Q1 WP4）。
+func TestCheckpointFilesAtBoundaries(t *testing.T) {
+	eng, _ := newTestEngine(t, mockScriptPlanner(baseSteps()...), map[string]map[string]string{
+		"01_researcher": {"notes.md": goodReport, "fields.json": "{}"},
+		"02_writer":     {"output.md": goodReport},
+		"03_reviewer":   {"output.md": goodReport},
+	}, NopSink{})
+	res, err := eng.Run(context.Background(), Goal{Text: "快照"}, RunOptions{})
+	if err != nil || res.Status != RunCompleted {
+		t.Fatalf("run: %v %+v", err, res)
+	}
+	latest, err := FileStore{}.LoadLatestCheckpoint(res.WorkDir)
+	if err != nil || latest == nil {
+		t.Fatalf("latest checkpoint: %v", err)
+	}
+	if latest.Superstep < 3 {
+		t.Fatalf("superstep = %d, want >= 3（三步各占一层）", latest.Superstep)
+	}
+	g, err := compileGraph(res.Plan.Steps)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if latest.GraphHash != g.hash() {
+		t.Fatalf("hash = %s, want %s", latest.GraphHash, g.hash())
+	}
+	if len(latest.StepStates) != 3 {
+		t.Fatalf("step states = %d", len(latest.StepStates))
+	}
+	if _, err := os.Stat(filepath.Join(res.WorkDir, "checkpoints", "001.json")); err != nil {
+		t.Fatalf("first boundary snapshot missing: %v", err)
+	}
+}
+
+// TestCheckpointAsyncAndOnExit：async 在退出前落完；on_exit 只有最终快照。
+func TestCheckpointAsyncAndOnExit(t *testing.T) {
+	for _, mode := range []string{"async", "on_exit"} {
+		eng, _ := newTestEngine(t, mockScriptPlanner(baseSteps()...), map[string]map[string]string{
+			"01_researcher": {"notes.md": goodReport, "fields.json": "{}"},
+			"02_writer":     {"output.md": goodReport},
+			"03_reviewer":   {"output.md": goodReport},
+		}, NopSink{})
+		res, err := eng.Run(context.Background(), Goal{Text: "快照"}, RunOptions{CheckpointMode: mode})
+		if err != nil || res.Status != RunCompleted {
+			t.Fatalf("%s run: %v %+v", mode, err, res)
+		}
+		latest, err := FileStore{}.LoadLatestCheckpoint(res.WorkDir)
+		if err != nil || latest == nil {
+			t.Fatalf("%s latest: %v", mode, err)
+		}
+		if latest.Superstep < 3 {
+			t.Fatalf("%s superstep = %d", mode, latest.Superstep)
+		}
+	}
+}
+
+// TestGraphContinuityRejectsStructuralChange：崩溃期间图形态被改则拒绝恢复（WP1 裁定）。
+func TestGraphContinuityRejectsStructuralChange(t *testing.T) {
+	eng, _ := newTestEngine(t, mockScriptPlanner(baseSteps()...), map[string]map[string]string{
+		"01_researcher": {"notes.md": goodReport, "fields.json": "{}"},
+		"02_writer":     {"output.md": goodReport},
+		"03_reviewer":   {"output.md": goodReport},
+	}, NopSink{})
+	dir := eng.opts.DataDir
+	res, err := eng.Run(context.Background(), Goal{Text: "先跑完再中断", HitlMode: HitlAuto}, RunOptions{})
+	if err != nil || res.Status != RunCompleted {
+		t.Fatalf("setup run: %v %+v", err, res)
+	}
+	// 翻回 running（仅状态变化，图结构不变——应先过连续性校验）。
+	st, err := loadRunState(res.WorkDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	st.Meta.Status = string(RunExecuting)
+	st.Meta.CompletedAt = nil
+	wf := st.Workflow()
+	wf.Status = "running"
+	wf.Steps[1].Status = "running"
+	wf.Steps[1].Attempts = 1
+	// 再改结构：把 s3 的 needs 改成另一个合法 id——编译能过但指纹必变。
+	wf.Steps[2].Needs = []string{"s1"}
+	st.setWorkflow(wf)
+	if err := st.save(); err != nil {
+		t.Fatal(err)
+	}
+	// 用一个新的引擎在同一目录上恢复（图编译本身合法：needs 引用存在的 id）。
+	reg := NewRoleRegistry()
+	for _, role := range []string{"researcher", "writer", "reviewer"} {
+		if err := reg.Register(RoleSpec{Role: role, Backend: "script"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	backends := NewBackendResolver()
+	backends.RegisterInstance("script", &scriptBackend{script: map[string]map[string]string{}})
+	eng2, err := NewEngine(Options{DataDir: dir, Registry: reg, Planner: eng.opts.Planner, Backends: backends, Sink: NopSink{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = eng2.Resume(context.Background(), res.RunID)
+	if err == nil || !strings.Contains(err.Error(), "图形态不一致") && !strings.Contains(err.Error(), "graph continuity") {
+		t.Fatalf("expected continuity rejection, got %v", err)
+	}
+}

squad/engine.go

@@ -60,6 +60,26 @@ type engineDeps struct {
 	StepTimeout  time.Duration
 	HitlPolicy   HitlPolicy
 	IntentRunner IntentRunner
+	// Store 运行状态持久化端口（Q1 WP4）；nil 回落 FileStore（装配期注入替换实现）。
+	Store Store
+}
+
+// storeOr 返回注入的 Store，缺省 FileStore。
+func (d engineDeps) storeOr() Store {
+	if d.Store != nil {
+		return d.Store
+	}
+	return FileStore{}
+}
+
+// loadRun 经端口加载运行状态，并让它携带同一端口（后续 save 自然走同一实现）。
+func (e *SquadEngine) loadRun(wsDir string) (*runState, error) {
+	st, err := e.deps.storeOr().LoadRun(wsDir)
+	if err != nil {
+		return nil, err
+	}
+	st.store = e.deps.storeOr()
+	return st, nil
 }
 
 // RunOptions 单次运行的可选项。
@@ -75,6 +95,8 @@ type RunOptions struct {
 	LeaderRole string `json:"leader_role,omitempty"`
 	// MaxLeaderTurns leader 模式每运行的 leader 轮次上限（默认 8，防失控）。
 	MaxLeaderTurns int `json:"max_leader_turns,omitempty"`
+	// CheckpointMode superstep 边界快照档位（Q1 WP4）：sync（默认）| async | on_exit。
+	CheckpointMode string `json:"checkpoint_mode,omitempty"`
 }
 
 // SquadEngine 小队引擎门面：Run / Respond / Resume / Inspect / ListRuns。
@@ -206,6 +228,9 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (*
 			meta.Metadata["max_leader_turns"] = fmt.Sprintf("%d", runOpts.MaxLeaderTurns)
 		}
 	}
+	if runOpts.CheckpointMode != "" {
+		meta.Metadata["checkpoint_mode"] = runOpts.CheckpointMode
+	}
 	e.deps.Sink.Emit(Event{Type: EvRunStarted, RunID: runID, Text: goal.Text, At: time.Now()})
 
 	// 意图闸门（可选，入口一次裁决；显式 TaskKey 优先，不重复过闸）。
@@ -239,7 +264,7 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (*
 		}
 	}
 
-	st := &runState{Meta: meta, Approved: map[string]bool{}, RepairBudgetRemaining: e.opts.RepairBudget}
+	st := &runState{Meta: meta, Approved: map[string]bool{}, RepairBudgetRemaining: e.opts.RepairBudget, store: e.deps.storeOr()}
 	return e.planAndExecute(ctx, st, runOpts)
 }
 
@@ -489,7 +514,7 @@ func (e *SquadEngine) Respond(ctx context.Context, runID string, decision Respon
 
 func (e *SquadEngine) respondLocked(ctx context.Context, runID string, decision RespondDecision) (*RunResult, error) {
 	wsDir := filepath.Join(e.opts.DataDir, "runs", runID)
-	st, err := loadRunState(wsDir)
+	st, err := e.loadRun(wsDir)
 	if err != nil {
 		return nil, fmt.Errorf("load run %s: %w", runID, err)
 	}
@@ -694,8 +719,17 @@ func (e *SquadEngine) applyDecision(ctx context.Context, st *runState, gate *hit
 		case "approve":
 			return e.manualPass(st, decision)
 		case "retry":
-			// 追加一次修复预算并重跑交付步骤。
+			// 追加一次修复预算并重跑交付步骤。累计追加有上限（Q26），
+			// 上限内才加——无上限的人工 retry 会把运行变成永动机。
+			term := e.terminationOf(st)
+			if st.RepairBudgetGranted >= term.MaxRepairGranted {
+				drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("人工修复追加已达上限 %d，不再追加（Q26）", term.MaxRepairGranted)})
+				return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+					Error: fmt.Sprintf("repair budget grants exhausted (%d/%d)", st.RepairBudgetGranted, term.MaxRepairGranted),
+					WorkDir: st.Meta.WorkDir}, nil
+			}
 			st.RepairBudgetRemaining++
+			st.RepairBudgetGranted++
 			st.Meta.Status = string(RunExecuting)
 			_ = st.save()
 			deliverStepID := lastCompletedStepID(st)
@@ -956,7 +990,7 @@ func (e *SquadEngine) Resume(ctx context.Context, runID string) (*RunResult, err
 	defer release()
 
 	wsDir := filepath.Join(e.opts.DataDir, "runs", runID)
-	st, err := loadRunState(wsDir)
+	st, err := e.loadRun(wsDir)
 	if err != nil {
 		return nil, fmt.Errorf("load run %s: %w", runID, err)
 	}
@@ -967,6 +1001,11 @@ func (e *SquadEngine) Resume(ctx context.Context, runID string) (*RunResult, err
 	if RunStatus(st.Meta.Status).Terminal() {
 		return e.Inspect(runID)
 	}
+	// 图连续性校验（Q1 WP4/WP1 裁定）：崩溃期间的检查点图指纹与当前 Steps
+	// 重编译不一致 = 图形态被改过（动态调整或外力），拒绝恢复并提示重新规划。
+	if err := validateGraphContinuity(st); err != nil {
+		return nil, err
+	}
 	if out, err := e.recoverInterrupted(ctx, st); out != nil || err != nil {
 		return out, err
 	}
@@ -979,7 +1018,7 @@ func (e *SquadEngine) Inspect(runID string) (*RunResult, error) {
 		return nil, invalidID(runID)
 	}
 	wsDir := filepath.Join(e.opts.DataDir, "runs", runID)
-	st, err := loadRunState(wsDir)
+	st, err := e.loadRun(wsDir)
 	if err != nil {
 		return nil, fmt.Errorf("load run %s: %w", runID, err)
 	}
@@ -1026,7 +1065,7 @@ func (e *SquadEngine) ListRuns() ([]RunMetadata, error) {
 		if !ent.IsDir() {
 			continue
 		}
-		st, err := loadRunState(filepath.Join(runsDir, ent.Name()))
+		st, err := e.loadRun(filepath.Join(runsDir, ent.Name()))
 		if err != nil {
 			continue
 		}
@@ -1060,7 +1099,7 @@ func (e *SquadEngine) Steer(runID, text string) error {
 	}
 	defer release()
 
-	st, err := loadRunState(filepath.Join(e.opts.DataDir, "runs", runID))
+	st, err := e.loadRun(filepath.Join(e.opts.DataDir, "runs", runID))
 	if err != nil {
 		return fmt.Errorf("load run %s: %w", runID, err)
 	}
@@ -1084,7 +1123,7 @@ func (e *SquadEngine) SetBlackboardField(runID, key, value string) error {
 	}
 	defer release()
 
-	st, err := loadRunState(filepath.Join(e.opts.DataDir, "runs", runID))
+	st, err := e.loadRun(filepath.Join(e.opts.DataDir, "runs", runID))
 	if err != nil {
 		return fmt.Errorf("load run %s: %w", runID, err)
 	}
@@ -1166,7 +1205,7 @@ func (e *SquadEngine) fireWakeup(ctx context.Context, runID string) bool {
 	if err := writeJSON(filepath.Join(e.opts.DataDir, "runs", runID, "pending.json"), gate); err != nil {
 		return false
 	}
-	if gate.Request.WakeupFires >= MaxWakeupFires {
+	if gate.Request.WakeupFires >= e.terminationOf(nil).MaxWakeupFires {
 		_, err := e.respondLocked(ctx, runID, RespondDecision{RequestID: gate.Request.ID, Choice: "abort",
 			Answer: "自动唤醒次数达上限，自动中止"})
 		return err == nil

squad/graph.go

@@ -2,6 +2,7 @@ package squad
 
 import (
 	"fmt"
+	"hash/fnv"
 	"strings"
 )
 
@@ -190,3 +191,24 @@ func pendingCount(steps []StepSpec) int {
 	}
 	return n
 }
+
+// hash 是图结构指纹：声明序、依赖与写集的稳定散列。
+// 恢复连续性校验用它比对崩溃前后的图形态（Q1 WP4；WP1 的图版本裁定落点）。
+func (g *graph) hash() string {
+	h := fnv.New64a()
+	for _, id := range g.order {
+		h.Write([]byte(id))
+		h.Write([]byte{0})
+		for _, n := range g.needs[id] {
+			h.Write([]byte(">"))
+			h.Write([]byte(n))
+		}
+		h.Write([]byte{1})
+		for _, w := range g.writes[id] {
+			h.Write([]byte("w"))
+			h.Write([]byte(w))
+		}
+		h.Write([]byte{2})
+	}
+	return fmt.Sprintf("%016x", h.Sum64())
+}

squad/leader.go

@@ -204,7 +204,8 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *runState) (*RunResu
 	}
 	_ = snapshotContextInto(st.Meta.WorkDir, subDir)
 
-	for st.Leader.Turns < st.Leader.MaxTurns {
+	term := e.terminationOf(st)
+	for st.Leader.Turns < term.MaxLeaderTurns {
 		// HITL：leader 角色权限升级（与标准模式同语义）。
 		if leader.BashPermission == "ask" && !st.Approved["role:"+leader.Role] {
 			if drv.policy(st).mode() != HitlInteractive {
@@ -368,7 +369,17 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *runState) (*RunResu
 			return e.leaderFail(st, fmt.Sprintf("leader 输出了未知 action %q", dec.Action)), nil, nil
 		}
 	}
-	return e.leaderFail(st, fmt.Sprintf("leader 轮次达上限 %d（防失控护栏），运行停止；可提高 MaxLeaderTurns 或人工介入", st.Leader.MaxTurns)), nil, nil
+	// Q23：达上限不再裸失败——按现有产出走 finalize 降级收口（质检/汇总路径与正常收口一致）。
+	drv2 := &flowDriver{opts: e.deps}
+	drv2.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("leader 轮次达上限 %d，按现有产出降级收口（Q23）", term.MaxLeaderTurns)})
+	out, gate2, err := drv2.finalize(ctx, st)
+	if err != nil {
+		return nil, nil, err
+	}
+	if gate2 != nil {
+		return heldResult(st, gate2), gate2, nil
+	}
+	return out, nil, nil
 }
 
 // leaderFail 统一的 leader 模式失败收口。

squad/leader_test.go

@@ -200,8 +200,13 @@ func TestLeaderModeTurnCapGuard(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if res.Status != RunFailed || !strings.Contains(res.Error, "上限") {
-		t.Fatalf("want turn-cap failure, got %s err=%s", res.Status, res.Error)
+	// Q23：达上限走 finalize 降级收口（completed/degraded/质检挂起），
+	// 不再是「运行停止」的裸失败。
+	if res.Status == RunFailed && strings.Contains(res.Error, "运行停止") {
+		t.Fatalf("Q23 closeout should not be a bare failure, got %s err=%s", res.Status, res.Error)
+	}
+	if res.Status != RunCompleted && res.Status != RunDegraded && res.Status != RunQualityReview {
+		t.Fatalf("Q23 closeout status = %s err=%s", res.Status, res.Error)
 	}
 	if b.turn < 3 {
 		t.Fatalf("leader should have run %d turns, ran %d", 3, b.turn)

squad/runstate.go

@@ -21,6 +21,8 @@ type runState struct {
 	Approved map[string]bool `json:"approved,omitempty"`
 	// RepairBudgetRemaining 质检失败后的自动重跑剩余次数。
 	RepairBudgetRemaining int `json:"repair_budget_remaining,omitempty"`
+	// RepairBudgetGranted 人工 retry 累计追加次数（Q26 上限判定用）。
+	RepairBudgetGranted int `json:"repair_budget_granted,omitempty"`
 	// Clarifications 已澄清问答（追加进 Goal 上下文）。
 	Clarifications []string `json:"clarifications,omitempty"`
 	// Steering 运行中人工补充指令（在下一步骤边界注入后清空）。
@@ -29,10 +31,21 @@ type runState struct {
 	Usage map[string]*Usage `json:"usage,omitempty"`
 	// Leader leader-delegation 模式的循环状态（workspace 内 leader.json 持久化）。
 	Leader *LeaderState `json:"-"`
+	// store 运行状态持久化端口（Q1 WP4）；nil 回落 FileStore（现状文件态）。
+	// 替换实现（如 SQLiteStore）在引擎装配期注入。
+	store Store `json:"-"`
 	// mu 保护并行步骤对共享状态（黑板/save/steering/usage）的变更。
 	mu sync.Mutex `json:"-"`
 }
 
+// storeOr 返回注入的 Store，缺省 FileStore。
+func (st *runState) storeOr() Store {
+	if st.store != nil {
+		return st.store
+	}
+	return FileStore{}
+}
+
 func (st *runState) policy() HitlPolicy {
 	return mergeHitlPolicy(HitlPolicy{}, st.Meta.HitlPolicyOf())
 }
@@ -51,6 +64,12 @@ func (m RunMetadata) HitlPolicyOf() *HitlPolicy {
 
 func (st *runState) save() error {
 	st.Meta.UpdatedAt = time.Now()
+	return st.storeOr().SaveRun(st)
+}
+
+// persist 是 FileStore 语义的低层落盘（全部写文件）。Store.SaveRun 调它；
+// runState.save 走 Store 端口，替换实现（SQLiteStore）在装配期注入后自然接管。
+func (st *runState) persist() error {
 	if err := writeJSON(filepath.Join(st.Meta.WorkDir, "workspace.json"), &st.Meta); err != nil {
 		return err
 	}
@@ -71,6 +90,7 @@ func (st *runState) save() error {
 	return writeJSON(filepath.Join(st.Meta.WorkDir, "runstate.json"), diskRunState{
 		Approved:              st.Approved,
 		RepairBudgetRemaining: st.RepairBudgetRemaining,
+		RepairBudgetGranted:   st.RepairBudgetGranted,
 		Clarifications:        st.Clarifications,
 	})
 }
@@ -78,6 +98,7 @@ func (st *runState) save() error {
 type diskRunState struct {
 	Approved              map[string]bool `json:"approved,omitempty"`
 	RepairBudgetRemaining int             `json:"repair_budget_remaining,omitempty"`
+	RepairBudgetGranted   int             `json:"repair_budget_granted,omitempty"`
 	Clarifications        []string        `json:"clarifications,omitempty"`
 }
 
@@ -131,6 +152,7 @@ func loadRunState(wsDir string) (*runState, error) {
 			st.Approved = disk.Approved
 		}
 		st.RepairBudgetRemaining = disk.RepairBudgetRemaining
+		st.RepairBudgetGranted = disk.RepairBudgetGranted
 		st.Clarifications = disk.Clarifications
 	} else if !os.IsNotExist(err) {
 		return nil, err

squad/runstore.go

@@ -0,0 +1,189 @@
+package squad
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// Store 是运行状态持久化端口（Q1 WP4）：run/gate/checkpoint 的读写以
+// 运行工作区为键。现状文件态收编为 FileStore（行为与此前逐文件直写完全一致）；
+// SQLite 实现进 adapters/sqlite（WP5）。事件流仍归 store/eventsink（子包），
+// 不在本端口内重复。
+type Store interface {
+	SaveRun(st *runState) error
+	LoadRun(wsDir string) (*runState, error)
+	SaveGate(wsDir string, gate *hitlGate) error
+	LoadGate(wsDir string) (*hitlGate, error)
+	SaveCheckpoint(wsDir string, cp Checkpoint) error
+	LoadLatestCheckpoint(wsDir string) (*Checkpoint, error)
+}
+
+// FileStore 把 runs/<id>/ 目录布局呈现为 Store（现状文件态）。
+type FileStore struct{}
+
+func (FileStore) SaveRun(st *runState) error {
+	return st.persist()
+}
+
+func (FileStore) LoadRun(wsDir string) (*runState, error) {
+	return loadRunState(wsDir)
+}
+
+func (FileStore) SaveGate(wsDir string, gate *hitlGate) error {
+	st := &runState{}
+	st.Meta.WorkDir = wsDir
+	return st.saveGate(gate)
+}
+
+func (FileStore) LoadGate(wsDir string) (*hitlGate, error) {
+	return loadGate(wsDir)
+}
+
+func (FileStore) SaveCheckpoint(wsDir string, cp Checkpoint) error {
+	dir := filepath.Join(wsDir, "checkpoints")
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	path := filepath.Join(dir, fmt.Sprintf("%03d.json", cp.Superstep))
+	if err := writeJSON(path, cp); err != nil {
+		return err
+	}
+	return writeJSON(filepath.Join(dir, "latest.json"), cp)
+}
+
+func (FileStore) LoadLatestCheckpoint(wsDir string) (*Checkpoint, error) {
+	data, err := os.ReadFile(filepath.Join(wsDir, "checkpoints", "latest.json"))
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	var cp Checkpoint
+	if err := json.Unmarshal(data, &cp); err != nil {
+		return nil, fmt.Errorf("parse latest checkpoint: %w", err)
+	}
+	return &cp, nil
+}
+
+// Checkpoint 是一个 superstep 边界的状态快照（Q1 WP4）。
+// GraphHash 是图结构指纹：恢复时与当前 Steps 重编译比对，不一致说明崩溃
+// 期间图形态被改过（动态调整或外力），按 WP1 裁定拒绝恢复并提示重新规划。
+type Checkpoint struct {
+	Superstep  int            `json:"superstep"`
+	At         time.Time      `json:"at"`
+	GraphHash  string         `json:"graph_hash"`
+	StepStates []StepSnapshot `json:"steps"`
+	wsDir      string         `json:"-"`
+}
+
+// write 立即同步落盘（FileStore 语义）。
+func (cp Checkpoint) write() error {
+	return FileStore{}.SaveCheckpoint(cp.wsDir, cp)
+}
+
+// StepSnapshot 是检查点里的单步快照。
+type StepSnapshot struct {
+	ID        string `json:"id"`
+	Status    string `json:"status"`
+	AttemptID string `json:"attempt_id,omitempty"`
+}
+
+// CheckpointMode 快照写盘档位（Q1 WP4 三档）。
+type CheckpointMode string
+
+const (
+	// CheckpointSync 每个 superstep 边界同步写（默认，与现状「逐步全量落盘」一致）。
+	CheckpointSync CheckpointMode = "sync"
+	// CheckpointAsync 边界快照进队列由单 goroutine 串行写（边界不阻塞；恢复粒度同 sync）。
+	CheckpointAsync CheckpointMode = "async"
+	// CheckpointOnExit 只在推进退出（终态/挂起/错误）时写最终快照。
+	CheckpointOnExit CheckpointMode = "on_exit"
+)
+
+// checkpointWriter 按档位执行边界快照；async 档单 goroutine 串行落盘。
+type checkpointWriter struct {
+	mode CheckpointMode
+	ch   chan Checkpoint
+	done chan error
+}
+
+func newCheckpointWriter(mode CheckpointMode) *checkpointWriter {
+	if mode == "" {
+		mode = CheckpointSync
+	}
+	w := &checkpointWriter{mode: mode}
+	if mode == CheckpointAsync {
+		w.ch = make(chan Checkpoint, 16)
+		w.done = make(chan error, 1)
+		go func() {
+			var first error
+			for cp := range w.ch {
+				if first != nil {
+					continue
+				}
+				if err := cp.write(); err != nil {
+					first = err
+				}
+			}
+			w.done <- first
+		}()
+	}
+	return w
+}
+
+// boundary 由调用方（flowDriver）在 superstep 边界触发：
+// sync 立即写；async 入队（满即错，不静默丢）；on_exit 跳过。
+func (w *checkpointWriter) boundary(wsDir string, cp Checkpoint) error {
+	cp.wsDir = wsDir
+	switch w.mode {
+	case CheckpointAsync:
+		select {
+		case w.ch <- cp:
+		default:
+			return fmt.Errorf("checkpoint queue full at superstep %d", cp.Superstep)
+		}
+		return nil
+	case CheckpointOnExit:
+		return nil
+	default:
+		return cp.write()
+	}
+}
+
+// flush 在推进退出前调用：async 落完并报告首个错误；on_exit 写最终快照。
+func (w *checkpointWriter) flush(wsDir string, cp Checkpoint) error {
+	cp.wsDir = wsDir
+	switch w.mode {
+	case CheckpointAsync:
+		close(w.ch)
+		return <-w.done
+	case CheckpointOnExit:
+		return cp.write()
+	default:
+		return nil
+	}
+}
+
+// validateGraphContinuity 恢复连续性校验（Q1 WP4；WP1 的图版本裁定）。
+// 最新检查点存在且其图指纹与当前 Steps 重编译不一致时拒绝恢复——
+// 崩溃期间图形态被改过，继续推进会把已完成的结果接到错误的图上。
+// 检查点不存在（旧运行）或指纹一致则放行。
+func validateGraphContinuity(st *runState) error {
+	latest, err := st.storeOr().LoadLatestCheckpoint(st.Meta.WorkDir)
+	if err != nil || latest == nil {
+		return err
+	}
+	g, gerr := compileGraph(st.Plan.Workflow.Steps)
+	if gerr != nil {
+		return fmt.Errorf("graph continuity: 当前 Steps 无法编译：%w（请人工介入或重新规划）", gerr)
+	}
+	if latest.GraphHash != g.hash() {
+		return fmt.Errorf("graph continuity: 崩溃前后图形态不一致（checkpoint superstep=%d hash=%s，当前=%s）——可能在动态调整或外部修改后崩溃，请重新规划或人工介入",
+			latest.Superstep, latest.GraphHash, g.hash())
+	}
+	return nil
+}

squad/termination.go

@@ -0,0 +1,120 @@
+package squad
+
+import (
+	"fmt"
+	"time"
+)
+
+// Termination 运行级停止条件的单一注册点（Q1 WP6）：leader 轮次、修复预算、
+// 唤醒次数与步骤超时集中声明；复合判定用 And/Or 原语（AutoGen 终止条件语义）。
+// 默认值与现网一致；引擎选项在 terminationOf 覆盖对应字段。
+type Termination struct {
+	MaxLeaderTurns   int
+	RepairBudget     int
+	MaxRepairGranted int // 人工 retry 累计加预算上限（Q26）：默认 RepairBudget*2
+	MaxWakeupFires   int
+	StepTimeout      time.Duration
+}
+
+// DefaultTermination 与现网四处的硬编码取值一致。
+func DefaultTermination() Termination {
+	return Termination{
+		MaxLeaderTurns:   DefaultMaxLeaderTurns,
+		RepairBudget:     1,
+		MaxRepairGranted: 2,
+		MaxWakeupFires:   MaxWakeupFires,
+		StepTimeout:      10 * time.Minute,
+	}
+}
+
+// terminationOf 取本运行的停止条件：引擎选项与运行状态覆盖默认值。
+func (e *SquadEngine) terminationOf(st *runState) Termination {
+	t := DefaultTermination()
+	if e.opts.RepairBudget > 0 {
+		t.RepairBudget = e.opts.RepairBudget
+		t.MaxRepairGranted = e.opts.RepairBudget * 2
+	}
+	if e.opts.StepTimeout > 0 {
+		t.StepTimeout = e.opts.StepTimeout
+	}
+	if st != nil && st.Leader != nil && st.Leader.MaxTurns > 0 {
+		t.MaxLeaderTurns = st.Leader.MaxTurns
+	}
+	return t
+}
+
+// Counters 运行期计数，由各判定处喂入（值语义快照）。
+type Counters struct {
+	LeaderTurns   int
+	RepairSpent   int
+	RepairGranted int
+	WakeupFires   int
+	Deadline      time.Time
+}
+
+// Cond 是一条停止条件：返回（原因，是否命中）。
+type Cond func(c Counters, now time.Time) (string, bool)
+
+// MaxTurnsC 轮次上限。
+func MaxTurnsC(n int) Cond {
+	return func(c Counters, _ time.Time) (string, bool) {
+		if c.LeaderTurns >= n {
+			return fmt.Sprintf("leader 轮次达上限 %d", n), true
+		}
+		return "", false
+	}
+}
+
+// RepairBudgetC 修复预算耗尽。
+func RepairBudgetC(n int) Cond {
+	return func(c Counters, _ time.Time) (string, bool) {
+		if c.RepairSpent > n {
+			return fmt.Sprintf("修复预算耗尽（%d/%d）", c.RepairSpent, n), true
+		}
+		return "", false
+	}
+}
+
+// WakeupsC 自动唤醒次数超限。
+func WakeupsC(n int) Cond {
+	return func(c Counters, _ time.Time) (string, bool) {
+		if c.WakeupFires > n {
+			return fmt.Sprintf("自动唤醒次数超限（%d/%d）", c.WakeupFires, n), true
+		}
+		return "", false
+	}
+}
+
+// DeadlineC 超过截止时刻。
+func DeadlineC() Cond {
+	return func(c Counters, now time.Time) (string, bool) {
+		if !c.Deadline.IsZero() && now.After(c.Deadline) {
+			return "超过截止时刻", true
+		}
+		return "", false
+	}
+}
+
+// And 全部条件命中才成立。
+func And(conds ...Cond) Cond {
+	return func(c Counters, now time.Time) (string, bool) {
+		for _, k := range conds {
+			if _, hit := k(c, now); !hit {
+				return "", false
+			}
+		}
+		return "all conditions met", true
+	}
+}
+
+// Or 任一条件命中即成立（默认的停止语义）。
+func Or(conds ...Cond) Cond {
+	return func(c Counters, now time.Time) (string, bool) {
+		for _, k := range conds {
+			if r, hit := k(c, now); hit {
+				return r, true
+			}
+		}
+		return "", false
+	}
+}

squad/termination_test.go

@@ -0,0 +1,82 @@
+package squad
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+)
+
+// TestTerminationAndOr：And 全中才成立、Or 任一中即成立（WP6 原语）。
+func TestTerminationAndOr(t *testing.T) {
+	c := Counters{LeaderTurns: 8, RepairSpent: 2, WakeupFires: 0}
+	or := Or(MaxTurnsC(8), WakeupsC(1))
+	if _, hit := or(c, time.Now()); !hit {
+		t.Fatal("or should hit on turns")
+	}
+	and := And(MaxTurnsC(8), WakeupsC(1))
+	if _, hit := and(c, time.Now()); hit {
+		t.Fatal("and should not hit with zero wakeups")
+	}
+	and2 := And(MaxTurnsC(8), RepairBudgetC(1))
+	if _, hit := and2(c, time.Now()); !hit {
+		t.Fatal("and should hit when all met")
+	}
+	dl := DeadlineC()
+	past := Counters{Deadline: time.Now().Add(-time.Minute)}
+	if _, hit := dl(past, time.Now()); !hit {
+		t.Fatal("deadline should hit when past")
+	}
+}
+
+// TestRepairGrantCap：人工 retry 累计追加有上限（Q26）。
+func TestRepairGrantCap(t *testing.T) {
+	engine, _ := newTestEngine(t, mockScriptPlanner(baseSteps()...), map[string]map[string]string{
+		"01_researcher": {"notes.md": goodReport, "fields.json": "{}"},
+		"02_writer":     {"output.md": goodReport},
+		"03_reviewer":   {"output.md": goodReport},
+	}, NopSink{})
+	res, err := engine.Run(context.Background(), Goal{Text: "先跑完再验", HitlMode: HitlAuto}, RunOptions{})
+	if err != nil || res.Status != RunCompleted {
+		t.Fatalf("setup run: %v %+v", err, res)
+	}
+	// 预置：修复授予已达上限 + 一个真实的质检挂起门（与 finalize 的形态一致）。
+	st, err := loadRunState(res.WorkDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	st.RepairBudgetGranted = engine.terminationOf(st).MaxRepairGranted
+	st.Meta.CompletedAt = nil
+	st.Meta.Status = string(RunQualityReview)
+	req := &HumanRequest{
+		ID: newRequestID(), RunID: st.Meta.ID, Kind: HitlQualityReview, Severity: SeverityActionRequired,
+		Question: "交付物质检未通过且自动修复预算已用尽，请人工裁决：",
+		Options: []HumanOption{
+			{Value: "approve", Label: "人工放行（记录人工通过）"},
+			{Value: "retry", Label: "再修复一次（重跑交付步骤）"},
+			{Value: "abort", Label: "判定失败"},
+		},
+		CreatedAt: time.Now(), Status: "pending",
+	}
+	if err := st.saveGate(&hitlGate{Request: *req, PrevStatus: RunExecuting}); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.save(); err != nil {
+		t.Fatal(err)
+	}
+	// 人工 retry 应被拒绝而不是再追加。
+	out, err := engine.Respond(context.Background(), res.RunID, RespondDecision{Choice: "retry"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if out == nil || !strings.Contains(out.Error, "grants exhausted") {
+		t.Fatalf("expected grant-cap failure, got %+v err=%v", out, err)
+	}
+	st2, err := loadRunState(res.WorkDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if st2.RepairBudgetGranted != engine.terminationOf(st2).MaxRepairGranted {
+		t.Fatalf("granted changed after cap: %d", st2.RepairBudgetGranted)
+	}
+}

squad/workflow.go

@@ -29,6 +29,19 @@ func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult,
 	d.emit(st, Event{Type: EvPhase, Text: "开始执行工作流"})
 
 	superNo := 0
+	cpw := newCheckpointWriter(CheckpointMode(st.Meta.Metadata["checkpoint_mode"]))
+	// 推进退出前 flush：async 落完、on_exit 写最终快照（Q1 WP4 三档的出口语义）。
+	var lastGraph *graph
+	defer func() {
+		if lastGraph == nil {
+			if g, gerr := compileGraph(st.Plan.Workflow.Steps); gerr == nil {
+				lastGraph = g
+			}
+		}
+		if lastGraph != nil {
+			_ = cpw.flush(st.Meta.WorkDir, snapshotOf(superNo, lastGraph, st.Plan.Workflow.Steps))
+		}
+	}()
 	for {
 		g, gerr := compileGraph(st.Plan.Workflow.Steps)
 		if gerr != nil {
@@ -115,11 +128,26 @@ func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult,
 				}
 			}
 		}
+
+		// superstep 边界快照（sync 立即 / async 入队；on_exit 由 defer 的 flush 写）。
+		lastGraph = g
+		if err := cpw.boundary(st.Meta.WorkDir, snapshotOf(superNo, g, st.Plan.Workflow.Steps)); err != nil {
+			return nil, nil, err
+		}
 	}
 
 	return d.finalize(ctx, st)
 }
 
+// snapshotOf 组装一个 superstep 边界快照：序号、图结构指纹与全量步骤状态。
+func snapshotOf(superNo int, g *graph, steps []StepSpec) Checkpoint {
+	cp := Checkpoint{Superstep: superNo, At: time.Now(), GraphHash: g.hash()}
+	for _, s := range steps {
+		cp.StepStates = append(cp.StepStates, StepSnapshot{ID: s.ID, Status: s.Status, AttemptID: s.AttemptID})
+	}
+	return cp
+}
+
 // gateBeforeRun 在步骤真正执行前检查权限升级和显式审批。
 // 返回非空结果或挂起时，调用方不得执行该步骤，也不得执行它所在的并行组。
 func (d *flowDriver) gateBeforeRun(ctx context.Context, st *runState, i int) (*RunResult, *hitlGate, error) {