Persist attempt ids and verify products on interrupted recovery
ci / vet-and-test (pull_request) Successful in 24s
ci / docker-checkout-hostgateway (pull_request) Successful in 0s

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
This commit is contained in:
2026-10-07 09:18:08 +00:00
parent 3870489614
commit 96d1e0e49c
4 changed files with 202 additions and 2 deletions
+173
View File
@@ -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)
}
}
+23
View File
@@ -903,6 +903,29 @@ func (e *SquadEngine) recoverInterrupted(ctx context.Context, st *runState) (*Ru
out, _, err := drv.failStep(ctx, st, i) out, _, err := drv.failStep(ctx, st, i)
return out, err 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" st.Plan.Workflow.Steps[i].Status = "pending"
if err := st.save(); err != nil { if err := st.save(); err != nil {
return nil, err return nil, err
+3
View File
@@ -95,6 +95,9 @@ type StepSpec struct {
Status string `json:"status"` // pending, running, completed, failed, skipped Status string `json:"status"` // pending, running, completed, failed, skipped
// Attempts 该步骤进入 running 的次数。中断恢复用它决定 auto 策略是否还能重试。 // 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"` DurationMs int64 `json:"duration_ms,omitempty"`
Error string `json:"error,omitempty"` Error string `json:"error,omitempty"`
// RequiresApproval 为 true 时,该步骤启动前进入 HITL 计划确认。 // RequiresApproval 为 true 时,该步骤启动前进入 HITL 计划确认。
+2 -1
View File
@@ -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}) d.emit(st, Event{Type: EvStepStarted, StepID: step.ID, Role: step.Role, Text: step.Title})
step.Status = "running" step.Status = "running"
step.Attempts++ step.Attempts++
step.AttemptID = fmt.Sprintf("%s/%s/%d", st.Meta.ID, step.ID, step.Attempts)
step.Error = "" step.Error = ""
// 与收口落盘同一把锁:并行步骤一边改 Steps、一边全量 marshal 会数据竞争。 // 与收口落盘同一把锁:并行步骤一边改 Steps、一边全量 marshal 会数据竞争。
st.mu.Lock() st.mu.Lock()
@@ -303,7 +304,7 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *runState, idx int) (*Ru
ToolAllowlist: role.Tools, ToolAllowlist: role.Tools,
Skills: firstNonEmptySlice(step.Skills, role.Skills), Skills: firstNonEmptySlice(step.Skills, role.Skills),
SessionScope: st.Meta.ID + "/" + step.ID, SessionScope: st.Meta.ID + "/" + step.ID,
ExecutionID: st.Meta.ID + "-" + step.ID, ExecutionID: step.AttemptID,
Budget: stepBudget, Budget: stepBudget,
} }
if step.IntentID != "" { if step.IntentID != "" {