From 96d1e0e49caf7797685366eecf9430533ed5cba1 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 7 Oct 2026 09:18:08 +0000 Subject: [PATCH] Persist attempt ids and verify products on interrupted recovery Why: Q1 WP3 - each step execution carries a stable attempt id (runID/stepID/N) for audit and idempotency, and auto-mode interrupted recovery must not re-execute a step whose products all landed before the crash (side effects exactly once). Evidence: go vet clean; full squad suite ok, including the new kill-at-superstep-boundary test (completed steps never re-execute, running-with-products judged complete by attempt) and missing-product retry tests. Impact: StepSpec gains AttemptID set when entering running and used as the execution id; recoverInterrupted (HitlAuto) checks the step's targets first - all present means completed with artifact merge, otherwise the legacy pending retry. Transient retries share the same AttemptID. Refs: Documents proposal 2026-10-07-agent-stack-n1q1i1-plans.md Q1 WP3 --- squad/attempt_test.go | 173 ++++++++++++++++++++++++++++++++++++++++++ squad/engine.go | 23 ++++++ squad/models.go | 5 +- squad/workflow.go | 3 +- 4 files changed, 202 insertions(+), 2 deletions(-) create mode 100644 squad/attempt_test.go diff --git a/squad/attempt_test.go b/squad/attempt_test.go new file mode 100644 index 0000000..c08697e --- /dev/null +++ b/squad/attempt_test.go @@ -0,0 +1,173 @@ +package squad + +import ( + "context" + "os" + "path/filepath" + "sync" + "testing" +) + +// perStepRecordingBackend 按子工作区归因计数（steps 的 role→tasks/NN_role 布局）。 +type perStepRecordingBackend struct { + mu sync.Mutex + calls map[string]int +} + +func (b *perStepRecordingBackend) Name() string { return "recording" } + +func (b *perStepRecordingBackend) Execute(ctx context.Context, req ExecRequest) (*ExecResult, error) { + b.mu.Lock() + if b.calls == nil { + b.calls = map[string]int{} + } + b.calls[filepath.Base(req.WorkDir)]++ + b.mu.Unlock() + out := map[string][]byte{} + for _, t := range req.Targets { + out[t] = []byte(goodReport) + } + if _, err := MaterializeTargets(req.WorkDir, req.Targets, out); err != nil { + return &ExecResult{Success: false, ErrorReason: err.Error()}, nil + } + return &ExecResult{Success: true, Logs: "ok"}, nil +} + +func (b *perStepRecordingBackend) countOf(sub string) int { + b.mu.Lock() + defer b.mu.Unlock() + return b.calls[sub] +} + +func newAttemptTestEngine(t *testing.T, backend *perStepRecordingBackend, dir string) *SquadEngine { + t.Helper() + reg := NewRoleRegistry() + for _, role := range []string{"researcher", "writer", "reviewer"} { + if err := reg.Register(RoleSpec{Role: role, Backend: "recording"}); err != nil { + t.Fatal(err) + } + } + backends := NewBackendResolver() + backends.RegisterInstance("recording", backend) + eng, err := NewEngine(Options{ + DataDir: dir, Registry: reg, Planner: mockScriptPlanner(baseSteps()...), + Backends: backends, Sink: NopSink{}, + }) + if err != nil { + t.Fatal(err) + } + return eng +} + +func flipToInterrupted(t *testing.T, workDir string, stepIdx int) { + t.Helper() + st, err := loadRunState(workDir) + if err != nil { + t.Fatal(err) + } + st.Meta.Status = string(RunExecuting) + st.Meta.CompletedAt = nil + wf := st.Workflow() + wf.Status = "running" + wf.Steps[stepIdx].Status = "running" + wf.Steps[stepIdx].Attempts = 1 + st.setWorkflow(wf) + if err := st.save(); err != nil { + t.Fatal(err) + } +} + +// TestAttemptIDFormat：attempt id 稳定落盘且形如 runID/stepID/N（Q1 WP3）。 +func TestAttemptIDFormat(t *testing.T) { + backend := &perStepRecordingBackend{} + dir := t.TempDir() + eng := newAttemptTestEngine(t, backend, dir) + res, err := eng.Run(context.Background(), Goal{Text: "格式"}, RunOptions{}) + if err != nil || res.Status != RunCompleted { + t.Fatalf("run: %v %+v", err, res) + } + if res.Plan == nil { + t.Fatal("no plan") + } + for _, s := range res.Plan.Steps { + want := res.RunID + "/" + s.ID + "/1" + if s.AttemptID != want { + t.Fatalf("step %s attempt id = %q, want %q", s.ID, s.AttemptID, want) + } + } +} + +// TestKilledAtSuperstepBoundarySideEffectsOnce：杀进程于 superstep 边界后恢复， +// 已完成步骤不重新执行（执行次数=1）；running 且产物齐全的步骤按 attempt 判定 +// 完成、不 rerun——Q1 验收第一条（WP3）。 +func TestKilledAtSuperstepBoundarySideEffectsOnce(t *testing.T) { + backend := &perStepRecordingBackend{} + dir := t.TempDir() + eng := newAttemptTestEngine(t, backend, dir) + 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) + } + before := map[string]int{} + st, err := loadRunState(res.WorkDir) + if err != nil { + t.Fatal(err) + } + for _, s := range st.Plan.Workflow.Steps { + before[s.ID] = backend.countOf(filepath.Base(filepath.FromSlash(s.SubWorkspace))) + } + + // 第二步停在 running（产物其实在盘上——首次运行真实写出的）。 + flipToInterrupted(t, res.WorkDir, 1) + + eng2 := newAttemptTestEngine(t, backend, dir) + res, err = eng2.Resume(context.Background(), res.RunID) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("resume: %+v", res) + } + for _, s := range res.Plan.Steps { + if s.Status != "completed" && s.Status != "skipped" { + t.Fatalf("step %s not finished: %s", s.ID, s.Status) + } + got := backend.countOf(filepath.Base(filepath.FromSlash(s.SubWorkspace))) + if got != before[s.ID] { + t.Fatalf("step %s executed %d times after kill (before=%d)——已完成/running 步骤被重复执行", s.ID, got, before[s.ID]) + } + } +} + +// TestInterruptedMissingProductsRetries：running 崩溃且产物缺失时按 auto 一次重试（WP3）。 +func TestInterruptedMissingProductsRetries(t *testing.T) { + backend := &perStepRecordingBackend{} + dir := t.TempDir() + eng := newAttemptTestEngine(t, backend, dir) + 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 并删掉其交付物（副作用未发生的形态）。 + flipToInterrupted(t, res.WorkDir, 1) + st, err := loadRunState(res.WorkDir) + if err != nil { + t.Fatal(err) + } + sub := filepath.FromSlash(st.Plan.Workflow.Steps[1].SubWorkspace) + if err := os.Remove(filepath.Join(res.WorkDir, sub, "output.md")); err != nil { + t.Fatal(err) + } + eng2 := newAttemptTestEngine(t, backend, dir) + res, err = eng2.Resume(context.Background(), res.RunID) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("resume: %+v", res) + } + got := backend.countOf(filepath.Base(sub)) + if got != 2 { + t.Fatalf("missing-product step executed %d times, want 2（auto 一次重试）", got) + } +} diff --git a/squad/engine.go b/squad/engine.go index ad689c1..18bf2ab 100644 --- a/squad/engine.go +++ b/squad/engine.go @@ -903,6 +903,29 @@ func (e *SquadEngine) recoverInterrupted(ctx context.Context, st *runState) (*Ru out, _, err := drv.failStep(ctx, st, i) return out, err } + // Q1 WP3 产物校验：步骤在 running 崩溃，但其交付物已全部落盘—— + // 副作用实际发生过，直接判定完成而不是重新执行（副作用次数=1）。 + subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(orDefault(step.SubWorkspace, fmt.Sprintf("tasks/%02d_%s", i+1, step.Role)))) + targets := step.Targets + if len(targets) == 0 { + targets = []string{"output.md"} + } + if missingTargets(subDir, targets) == nil { + step.Status = "completed" + step.Error = "" + st.Plan.Workflow.Steps[i] = step + st.mu.Lock() + _ = mergeStepArtifacts(&st.BB, subDir) + err := st.save() + st.mu.Unlock() + if err != nil { + return nil, err + } + drv := &flowDriver{opts: e.deps} + drv.emit(st, Event{Type: EvStepCompleted, StepID: step.ID, Role: step.Role, + Text: fmt.Sprintf("中断恢复：产物齐全，按 attempt %s 判定完成", step.AttemptID)}) + continue + } st.Plan.Workflow.Steps[i].Status = "pending" if err := st.save(); err != nil { return nil, err diff --git a/squad/models.go b/squad/models.go index dac1795..884b1a0 100644 --- a/squad/models.go +++ b/squad/models.go @@ -94,7 +94,10 @@ type StepSpec struct { Targets []string `json:"targets,omitempty"` // 期望交付物相对路径，缺省 output.md Status string `json:"status"` // pending, running, completed, failed, skipped // Attempts 该步骤进入 running 的次数。中断恢复用它决定 auto 策略是否还能重试。 - Attempts int `json:"attempts,omitempty"` + Attempts int `json:"attempts,omitempty"` + // AttemptID 当前这次尝试的稳定标识：runID/stepID/N（N=Attempts）。 + // 执行事件、产物与副作用记录都挂它；瞬态重试共享同一 AttemptID（Q1 WP3）。 + AttemptID string `json:"attempt_id,omitempty"` DurationMs int64 `json:"duration_ms,omitempty"` Error string `json:"error,omitempty"` // RequiresApproval 为 true 时，该步骤启动前进入 HITL 计划确认。 diff --git a/squad/workflow.go b/squad/workflow.go index 25582ca..3f01b29 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -231,6 +231,7 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *runState, idx int) (*Ru d.emit(st, Event{Type: EvStepStarted, StepID: step.ID, Role: step.Role, Text: step.Title}) step.Status = "running" step.Attempts++ + step.AttemptID = fmt.Sprintf("%s/%s/%d", st.Meta.ID, step.ID, step.Attempts) step.Error = "" // 与收口落盘同一把锁：并行步骤一边改 Steps、一边全量 marshal 会数据竞争。 st.mu.Lock() @@ -303,7 +304,7 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *runState, idx int) (*Ru ToolAllowlist: role.Tools, Skills: firstNonEmptySlice(step.Skills, role.Skills), SessionScope: st.Meta.ID + "/" + step.ID, - ExecutionID: st.Meta.ID + "-" + step.ID, + ExecutionID: step.AttemptID, Budget: stepBudget, } if step.IntentID != "" { -- 2.54.0