diff --git a/cmd/agent-squad/server/server.go b/cmd/agent-squad/server/server.go index 5ac91d4..c3aa920 100644 --- a/cmd/agent-squad/server/server.go +++ b/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()) } } }() diff --git a/squad/checkpoint_test.go b/squad/checkpoint_test.go new file mode 100644 index 0000000..de2b8e8 --- /dev/null +++ b/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) + } +} diff --git a/squad/engine.go b/squad/engine.go index 18bf2ab..7fca98d 100644 --- a/squad/engine.go +++ b/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 diff --git a/squad/graph.go b/squad/graph.go index 66fac0a..461f7ee 100644 --- a/squad/graph.go +++ b/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()) +} diff --git a/squad/leader.go b/squad/leader.go index 4ce0e61..f2372a8 100644 --- a/squad/leader.go +++ b/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 模式失败收口。 diff --git a/squad/leader_test.go b/squad/leader_test.go index cad7080..41ce06b 100644 --- a/squad/leader_test.go +++ b/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) diff --git a/squad/runstate.go b/squad/runstate.go index 3b5e564..20fd1b0 100644 --- a/squad/runstate.go +++ b/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 diff --git a/squad/runstore.go b/squad/runstore.go new file mode 100644 index 0000000..0b95522 --- /dev/null +++ b/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// 目录布局呈现为 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 +} diff --git a/squad/termination.go b/squad/termination.go new file mode 100644 index 0000000..1074004 --- /dev/null +++ b/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 + } +} diff --git a/squad/termination_test.go b/squad/termination_test.go new file mode 100644 index 0000000..0c4221a --- /dev/null +++ b/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) + } +} diff --git a/squad/workflow.go b/squad/workflow.go index 3f01b29..3d37bc2 100644 --- a/squad/workflow.go +++ b/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) {