From 8e987bf535efbbbf317147542e858217f2bb3504 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 3 Oct 2026 01:48:10 +0000 Subject: [PATCH] Persist step approvals and guard squad HTTP entry points Why: A step confirmation cancelled the whole run, approvals and repair budget vanished on restart, and the HTTP server accepted anonymous and oversized requests. Evidence: GOWORK=off go vet ./... && GOWORK=off go test -count=1 ./... && GOWORK=off go test -race ./squad/... passed. scripts/submodule-consistency.sh ok. Impact: Approve and skip apply to one step; runstate.json and unix flock survive restart; wakeup leaves permission holds pending; serve defaults to loopback and requires a token off-loopback. Refs: Q1 Q2 Q3 Q4 Q5 Q8 Q9 Q14 Q15 Q16 Co-authored-by: Cursor --- PROJECT.md | 2 + cli/cli.go | 21 ++- deps/agent-intent | 2 +- deps/agent-native | 2 +- deps/error-code-system | 2 +- deps/logging-system | 2 +- server/guard_test.go | 74 +++++++++++ server/server.go | 128 +++++++++++++++--- squad/blackboard.go | 6 +- squad/blueprint.go | 10 +- squad/engine.go | 246 +++++++++++++++++++++++++++++----- squad/hitl.go | 6 +- squad/id.go | 19 +++ squad/leader.go | 12 +- squad/lock_other.go | 11 ++ squad/lock_unix.go | 42 ++++++ squad/models.go | 6 +- squad/recovery_test.go | 294 +++++++++++++++++++++++++++++++++++++++++ squad/registry.go | 10 +- squad/runstate.go | 25 +++- squad/workflow.go | 18 ++- squad/workspace.go | 38 +++++- store/store.go | 12 +- 23 files changed, 890 insertions(+), 98 deletions(-) create mode 100644 server/guard_test.go create mode 100644 squad/id.go create mode 100644 squad/lock_other.go create mode 100644 squad/lock_unix.go create mode 100644 squad/recovery_test.go diff --git a/PROJECT.md b/PROJECT.md index 7351978..28311bd 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -28,6 +28,8 @@ `go.mod` 的 `replace` 都指向 `./deps/`。嵌套在 `deps/agent-native/deps/`、`deps/agent-intent/deps/` 里的同名副本不参与本仓编译，提交必须与本仓 gitlink 一致（`scripts/submodule-consistency.sh`）。 +同一 run 的临界区在 unix 上额外持有 `runs//.lock` 的 flock，拿不到锁时返回 `ErrRunBusy`。非 unix 平台没有这把文件锁，只保留进程内互斥。 + 克隆： ```bash diff --git a/cli/cli.go b/cli/cli.go index 2278f81..6a025cc 100644 --- a/cli/cli.go +++ b/cli/cli.go @@ -7,6 +7,7 @@ import ( "encoding/json" "flag" "fmt" + "net" "net/http" "os" "path/filepath" @@ -61,7 +62,7 @@ func usage() { fmt.Print(`agent-squad — 通用智能体小队编排底座 用法: - agent-squad serve [--addr :8090] [--data DIR] [--planner rule|llm] [--model M] + agent-squad serve [--addr 127.0.0.1:8090] [--data DIR] [--token TOK] [--planner rule|llm] [--model M] agent-squad role list|add|rm|recommend|derive ... agent-squad blueprint list|add|rm ... agent-squad run "目标文本" [--task-key K] [--backend B] [--materials DIR] [--hitl-mode interactive|auto|disabled] @@ -206,7 +207,9 @@ func first(vals ...string) string { func cmdServe(args []string) int { fs := flag.NewFlagSet("serve", flag.ContinueOnError) - addr := fs.String("addr", ":8090", "监听地址") + addr := fs.String("addr", "127.0.0.1:8090", "监听地址") + token := fs.String("token", os.Getenv("AGENT_SQUAD_TOKEN"), "Bearer token；也可用 AGENT_SQUAD_TOKEN") + insecure := fs.Bool("insecure", false, "允许在没有 token 时监听非回环地址") data := fs.String("data", dataDir(), "数据目录") planner := fs.String("planner", "rule", "规划器：rule|llm") model := fs.String("model", "", "默认模型（llm 规划器/后端）") @@ -220,7 +223,12 @@ func cmdServe(args []string) int { return 1 } defer a.close() + if *token == "" && !*insecure && !loopbackAddr(*addr) { + fmt.Fprintln(os.Stderr, "拒绝在没有 token 时监听非回环地址；传入 --token 或 --insecure") + return 2 + } srv := server.New(a.engine, a.sink, a.st) + srv.SetToken(*token) srv.SetLogger(a.log) if plat, err := platform.Open(filepath.Join(*data, "orchestration.db")); err != nil { fmt.Fprintln(os.Stderr, "编排库打开失败:", err) @@ -236,6 +244,15 @@ func cmdServe(args []string) int { return httpListen(*addr, srv.Handler()) } +func loopbackAddr(addr string) bool { + host, _, err := net.SplitHostPort(addr) + if err != nil { + host = addr + } + host = strings.Trim(host, "[]") + return host == "127.0.0.1" || host == "::1" || strings.EqualFold(host, "localhost") +} + func httpListen(addr string, h http.Handler) int { if err := http.ListenAndServe(addr, h); err != nil { fmt.Fprintln(os.Stderr, "服务退出:", err) diff --git a/deps/agent-intent b/deps/agent-intent index 3674dc4..3130190 160000 --- a/deps/agent-intent +++ b/deps/agent-intent @@ -1 +1 @@ -Subproject commit 3674dc4bfd0b49191d21f0afab4002381e41a429 +Subproject commit 3130190acb396589d6022c243bbe07ce7c31c881 diff --git a/deps/agent-native b/deps/agent-native index a34689c..694b08c 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit a34689c78fbddbfa34b343cd6f85cca36c4d2f9d +Subproject commit 694b08cdbef5d52496f4425c9a22d1b77df14c23 diff --git a/deps/error-code-system b/deps/error-code-system index dbeed2f..a3ee67b 160000 --- a/deps/error-code-system +++ b/deps/error-code-system @@ -1 +1 @@ -Subproject commit dbeed2f1b67eb417ce089e7033a228578fbbeb03 +Subproject commit a3ee67bba269e1276c14e22c026171ecf2cf6190 diff --git a/deps/logging-system b/deps/logging-system index 2c5716f..49b5d8b 160000 --- a/deps/logging-system +++ b/deps/logging-system @@ -1 +1 @@ -Subproject commit 2c5716fdcffbc4696fc0371afc70f39c44174201 +Subproject commit 49b5d8be604d05e4246dffa0d10fa1c2b7ad45d7 diff --git a/server/guard_test.go b/server/guard_test.go new file mode 100644 index 0000000..a3beff7 --- /dev/null +++ b/server/guard_test.go @@ -0,0 +1,74 @@ +package server + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "agent-squad/squad" + "agent-squad/store" +) + +func TestMissingTokenIsUnauthorized(t *testing.T) { + dir := t.TempDir() + st, err := store.New(dir) + if err != nil { + t.Fatal(err) + } + reg := squad.NewRoleRegistry() + engine, err := squad.NewEngine(squad.Options{ + DataDir: dir, Registry: reg, Planner: &squad.MockPlanner{}, Sink: squad.NopSink{}, + }) + if err != nil { + t.Fatal(err) + } + srv := New(engine, squad.NewFanoutSink(8), st) + srv.SetToken("secret") + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/v1/roles") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("status %d", resp.StatusCode) + } + req, _ := http.NewRequest(http.MethodGet, ts.URL+"/v1/roles", nil) + req.Header.Set("Authorization", "Bearer secret") + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("authed status %d", resp.StatusCode) + } +} + +func TestPathTraversalRejected(t *testing.T) { + ts, _ := newTestServer(t) + resp, err := http.Get(ts.URL + "/v1/runs/a..b") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status %d", resp.StatusCode) + } +} + +func TestOversizedBodyIs413(t *testing.T) { + ts, _ := newTestServer(t) + body := strings.NewReader(`{"role":"x","name":"` + strings.Repeat("a", 2<<20) + `"}`) + resp, err := http.Post(ts.URL+"/v1/roles", "application/json", body) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusRequestEntityTooLarge { + t.Fatalf("status %d", resp.StatusCode) + } +} diff --git a/server/server.go b/server/server.go index 1672620..6a74675 100644 --- a/server/server.go +++ b/server/server.go @@ -4,11 +4,14 @@ package server import ( "context" + "crypto/subtle" "encoding/base64" "encoding/json" + "errors" "fmt" "net/http" "strconv" + "strings" "sync" "time" @@ -34,8 +37,12 @@ type Server struct { latest map[string]*squad.RunResult // 最近一次推进结果（供轮询） log *logsys.Logger plat *platform.Store + token string } +// SetToken 要求除健康检查外的请求带 Bearer token。空串表示不校验。 +func (s *Server) SetToken(token string) { s.token = token } + // SetLogger 设置组合根日志器。nil 表示不写日志。 func (s *Server) SetLogger(log *logsys.Logger) { if s == nil { @@ -87,7 +94,54 @@ func (s *Server) Handler() http.Handler { if s.plat != nil { platform.Mount(mux, s.plat) } - return httpx.Trace(mux) + var h http.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + limit := int64(1 << 20) + if r.Method == http.MethodPost && r.URL.Path == "/v1/runs" { + limit = 32 << 20 + } + r.Body = http.MaxBytesReader(w, r.Body, limit) + mux.ServeHTTP(w, r) + }) + if s.token != "" { + inner := h + h = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/health" { + inner.ServeHTTP(w, r) + return + } + const prefix = "Bearer " + header := r.Header.Get("Authorization") + got := strings.TrimPrefix(header, prefix) + if len(header) != len(prefix)+len(s.token) || subtle.ConstantTimeCompare([]byte(got), []byte(s.token)) != 1 { + w.WriteHeader(http.StatusUnauthorized) + return + } + inner.ServeHTTP(w, r) + }) + } + return httpx.Trace(h) +} + +func (s *Server) decode(w http.ResponseWriter, r *http.Request, dst any) bool { + err := json.NewDecoder(r.Body).Decode(dst) + if err == nil { + return true + } + var maxErr *http.MaxBytesError + if errors.As(err, &maxErr) || strings.Contains(err.Error(), "request body too large") { + w.WriteHeader(http.StatusRequestEntityTooLarge) + return false + } + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + return false +} + +func rejectBadID(w http.ResponseWriter, id string) bool { + if squad.ValidID(id) { + return false + } + w.WriteHeader(http.StatusBadRequest) + return true } func writeJSON(w http.ResponseWriter, code int, v any) { @@ -132,8 +186,7 @@ func (s *Server) handleListRoles(w http.ResponseWriter, r *http.Request) { func (s *Server) handlePutRole(w http.ResponseWriter, r *http.Request) { var spec squad.RoleSpec - if err := json.NewDecoder(r.Body).Decode(&spec); err != nil { - s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + if !s.decode(w, r, &spec) { return } if key := r.PathValue("key"); key != "" && spec.Role == "" { @@ -148,6 +201,9 @@ func (s *Server) handlePutRole(w http.ResponseWriter, r *http.Request) { func (s *Server) handleDeleteRole(w http.ResponseWriter, r *http.Request) { key := r.PathValue("key") + if rejectBadID(w, key) { + return + } if !s.engine.Registry().Remove(key) { s.writeErr(w, r, errcode.New(errcode.SquadRoleNotFound, fmt.Sprintf("role %s not found", key))) return @@ -158,8 +214,7 @@ func (s *Server) handleDeleteRole(w http.ResponseWriter, r *http.Request) { func (s *Server) handleDeriveRole(w http.ResponseWriter, r *http.Request) { var req squad.DeriveRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + if !s.decode(w, r, &req) { return } spec, err := s.engine.Registry().Derive(req) @@ -183,8 +238,7 @@ func (s *Server) handleListBlueprints(w http.ResponseWriter, r *http.Request) { func (s *Server) handlePutBlueprint(w http.ResponseWriter, r *http.Request) { var bp squad.Blueprint - if err := json.NewDecoder(r.Body).Decode(&bp); err != nil { - s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + if !s.decode(w, r, &bp) { return } if err := s.engine.Blueprints().Register(bp); err != nil { @@ -196,6 +250,9 @@ func (s *Server) handlePutBlueprint(w http.ResponseWriter, r *http.Request) { func (s *Server) handleDeleteBlueprint(w http.ResponseWriter, r *http.Request) { key := r.PathValue("key") + if rejectBadID(w, key) { + return + } if !s.engine.Blueprints().Remove(key) { s.writeErr(w, r, errcode.New(errcode.SquadBlueprintNotFound, fmt.Sprintf("blueprint %s not found", key))) return @@ -221,8 +278,7 @@ type createRunRequest struct { func (s *Server) handleCreateRun(w http.ResponseWriter, r *http.Request) { var req createRunRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + if !s.decode(w, r, &req) { return } for name, b64 := range req.MaterialsB64 { @@ -276,7 +332,11 @@ func (s *Server) handleListRuns(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleInspectRun(w http.ResponseWriter, r *http.Request) { - res, err := s.engine.Inspect(r.PathValue("id")) + id := r.PathValue("id") + if rejectBadID(w, id) { + return + } + res, err := s.engine.Inspect(id) if err != nil { s.writeErr(w, r, errcode.Wrap(err, errcode.SquadRunNotFound, "运行不存在")) return @@ -285,7 +345,11 @@ func (s *Server) handleInspectRun(w http.ResponseWriter, r *http.Request) { } func (s *Server) handlePending(w http.ResponseWriter, r *http.Request) { - req, err := s.engine.Pending(r.PathValue("id")) + id := r.PathValue("id") + if rejectBadID(w, id) { + return + } + req, err := s.engine.Pending(id) if err != nil || req == nil { writeJSON(w, http.StatusOK, map[string]any{"pending": nil}) return @@ -295,11 +359,14 @@ func (s *Server) handlePending(w http.ResponseWriter, r *http.Request) { func (s *Server) handleRespond(w http.ResponseWriter, r *http.Request) { var decision squad.RespondDecision - if err := json.NewDecoder(r.Body).Decode(&decision); err != nil { - s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + if !s.decode(w, r, &decision) { return } - res, err := s.engine.Respond(r.Context(), r.PathValue("id"), decision) + id := r.PathValue("id") + if rejectBadID(w, id) { + return + } + res, err := s.engine.Respond(r.Context(), id, decision) if err != nil { s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return @@ -310,7 +377,11 @@ func (s *Server) handleRespond(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleResume(w http.ResponseWriter, r *http.Request) { - res, err := s.engine.Resume(r.Context(), r.PathValue("id")) + id := r.PathValue("id") + if rejectBadID(w, id) { + return + } + res, err := s.engine.Resume(r.Context(), id) if err != nil { s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return @@ -321,7 +392,11 @@ func (s *Server) handleResume(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleBlackboard(w http.ResponseWriter, r *http.Request) { - bb, err := s.engine.Blackboard(r.PathValue("id")) + id := r.PathValue("id") + if rejectBadID(w, id) { + return + } + bb, err := s.engine.Blackboard(id) if err != nil { s.writeErr(w, r, errcode.Wrap(err, errcode.SquadRunNotFound, "运行不存在")) return @@ -333,11 +408,14 @@ func (s *Server) handleSteer(w http.ResponseWriter, r *http.Request) { var req struct { Text string `json:"text"` } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + if !s.decode(w, r, &req) { return } - if err := s.engine.Steer(r.PathValue("id"), req.Text); err != nil { + id := r.PathValue("id") + if rejectBadID(w, id) { + return + } + if err := s.engine.Steer(id, req.Text); err != nil { s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } @@ -349,11 +427,14 @@ func (s *Server) handleSetField(w http.ResponseWriter, r *http.Request) { Field string `json:"field"` Value string `json:"value"` } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) + if !s.decode(w, r, &req) { return } - if err := s.engine.SetBlackboardField(r.PathValue("id"), req.Field, req.Value); err != nil { + id := r.PathValue("id") + if rejectBadID(w, id) { + return + } + if err := s.engine.SetBlackboardField(id, req.Field, req.Value); err != nil { s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } @@ -389,6 +470,9 @@ func (s *Server) StartWakeupSweeper(interval time.Duration) (stop func()) { // handleEvents SSE：cursor 之后重放缓冲再实时推送。 func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) { + if rejectBadID(w, r.PathValue("id")) { + return + } flusher, ok := w.(http.Flusher) if !ok { s.writeErr(w, r, errcode.New(errcode.SysInternal, "streaming unsupported")) diff --git a/squad/blackboard.go b/squad/blackboard.go index 61008de..be8eacd 100644 --- a/squad/blackboard.go +++ b/squad/blackboard.go @@ -34,11 +34,7 @@ func saveBlackboard(wsDir string, bb *BlackboardState) error { if err != nil { return err } - tmp := filepath.Join(wsDir, ".blackboard.json.tmp") - if err := os.WriteFile(tmp, data, 0o644); err != nil { - return err - } - return os.Rename(tmp, filepath.Join(wsDir, "blackboard.json")) + return AtomicWrite(filepath.Join(wsDir, "blackboard.json"), append(data, '\n')) } // mergeStepArtifacts 合并步骤产出：fields.json（字符串键值）与 diff --git a/squad/blueprint.go b/squad/blueprint.go index 7d12be8..9e3867d 100644 --- a/squad/blueprint.go +++ b/squad/blueprint.go @@ -13,7 +13,7 @@ type BlueprintRegistry struct { mu sync.RWMutex bps map[string]Blueprint // OnChange 可选钩子：注册/删除后回调（store 层用于落盘）。 - OnChange func(bp Blueprint) + OnChange func(bp Blueprint) error } // NewBlueprintRegistry 创建空注册表。 @@ -23,8 +23,8 @@ func NewBlueprintRegistry() *BlueprintRegistry { // Register 注册或更新模板。 func (r *BlueprintRegistry) Register(bp Blueprint) error { - if strings.TrimSpace(bp.Key) == "" { - return fmt.Errorf("blueprint key cannot be empty") + if !ValidID(bp.Key) { + return invalidID(bp.Key) } if len(bp.Steps) == 0 { return fmt.Errorf("blueprint %s has no steps", bp.Key) @@ -41,7 +41,9 @@ func (r *BlueprintRegistry) Register(bp Blueprint) error { r.bps[bp.Key] = bp r.mu.Unlock() if r.OnChange != nil { - r.OnChange(bp) + if err := r.OnChange(bp); err != nil { + return err + } } return nil } diff --git a/squad/engine.go b/squad/engine.go index 01d245f..abb3a09 100644 --- a/squad/engine.go +++ b/squad/engine.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -152,15 +153,42 @@ func (e *SquadEngine) runLock(runID string) *sync.Mutex { return l } +// ErrRunBusy 表示该 run 的文件锁正被占用，且在 ctx 截止前没有拿到。 +// 非 unix 平台没有 flock，只保留进程内互斥；跨进程互斥需要 unix。 +var ErrRunBusy = errors.New("run is busy") + +// acquire 进入 run 临界区：先校验 id，再拿进程内锁，然后按 ctx 轮询文件锁。 +func (e *SquadEngine) acquire(ctx context.Context, runID string) (func(), error) { + if !ValidID(runID) { + return nil, invalidID(runID) + } + if ctx == nil { + ctx = context.Background() + } + mu := e.runLock(runID) + mu.Lock() + release, err := lockRunFile(ctx, filepath.Join(e.opts.DataDir, "runs", runID)) + if err != nil { + mu.Unlock() + return nil, err + } + return func() { + release() + mu.Unlock() + }, nil +} + // Run 发起一次运行：规划（含澄清/确认挂起）→ 执行 → 收口。 func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (*RunResult, error) { if strings.TrimSpace(goal.Text) == "" { return nil, fmt.Errorf("goal text is required") } runID := fmt.Sprintf("run_%s", time.Now().UTC().Format("20060102T150405.000000000")) - lock := e.runLock(runID) - lock.Lock() - defer lock.Unlock() + release, err := e.acquire(ctx, runID) + if err != nil { + return nil, err + } + defer release() meta := RunMetadata{ ID: runID, Goal: goal, Status: string(RunPlanning), @@ -309,6 +337,7 @@ func (e *SquadEngine) planAndExecute(ctx context.Context, st *runState, runOpts Context: fmt.Sprintf("确认原因：%s\n编制：\n%s\n步骤：\n%s", planOut.ConfirmReason, strings.Join(roles, "\n"), strings.Join(lines, "\n")), CreatedAt: time.Now(), Status: "pending", + Reason: "plan", } gate := &hitlGate{Request: *req, Plan: planOut.Plan, PrevStatus: RunPlanning} if err := st.saveGate(gate); err != nil { @@ -450,10 +479,15 @@ func (e *SquadEngine) failNew(meta *RunMetadata, err error) (*RunResult, error) // Respond 应用人工决策并续跑。按请求 ID 幂等：重复回应同一请求返回当前状态。 func (e *SquadEngine) Respond(ctx context.Context, runID string, decision RespondDecision) (*RunResult, error) { - lock := e.runLock(runID) - lock.Lock() - defer lock.Unlock() + release, err := e.acquire(ctx, runID) + if err != nil { + return nil, err + } + defer release() + return e.respondLocked(ctx, runID, decision) +} +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) if err != nil { @@ -487,6 +521,12 @@ func (e *SquadEngine) Respond(ctx context.Context, runID string, decision Respon case HitlClarification: return e.respondClarification(ctx, st, gate, decision) case HitlConfirmation: + if req.Reason == "interrupted" { + return e.respondInterrupted(ctx, st, req, decision) + } + if req.StepID != "" && gate.Plan == nil { + return e.respondStepApproval(ctx, st, req, decision) + } switch strings.ToLower(decision.Choice) { case "approve": if gate.Plan != nil { @@ -734,11 +774,113 @@ func (d *flowDriver) failStepResume(ctx context.Context, st *runState, idx int) return d.failStep(ctx, st, idx) } +func (e *SquadEngine) respondStepApproval(ctx context.Context, st *runState, req *HumanRequest, decision RespondDecision) (*RunResult, error) { + switch strings.ToLower(decision.Choice) { + case "approve": + st.Approved["step:"+req.StepID] = true + case "skip": + if !markStep(st, req.StepID, "skipped", "") { + return nil, fmt.Errorf("step %s not found", req.StepID) + } + default: + return e.cancelRun(st, "人工中止（步骤审批）"), nil + } + st.Meta.Status = string(RunExecuting) + if err := st.save(); err != nil { + return nil, err + } + return e.resumeAfterHold(ctx, st) +} + +func (e *SquadEngine) respondInterrupted(ctx context.Context, st *runState, req *HumanRequest, decision RespondDecision) (*RunResult, error) { + switch strings.ToLower(decision.Choice) { + case "retry": + if !markStep(st, req.StepID, "pending", "") { + return nil, fmt.Errorf("step %s not found", req.StepID) + } + case "skip": + if !markStep(st, req.StepID, "skipped", "") { + return nil, fmt.Errorf("step %s not found", req.StepID) + } + default: + return e.cancelRun(st, "人工中止（中断恢复）"), nil + } + st.Meta.Status = string(RunExecuting) + if err := st.save(); err != nil { + return nil, err + } + return e.resumeAfterHold(ctx, st) +} + +func markStep(st *runState, id, status, errText string) bool { + for i := range st.Plan.Workflow.Steps { + if st.Plan.Workflow.Steps[i].ID != id { + continue + } + st.Plan.Workflow.Steps[i].Status = status + st.Plan.Workflow.Steps[i].Error = errText + return true + } + return false +} + +// recoverInterrupted 处理崩溃时停在 running 的步骤。 +// interactive 挂起确认，不自动再执行；auto 允许一次重试；disabled 直接失败。 +func (e *SquadEngine) recoverInterrupted(ctx context.Context, st *runState) (*RunResult, error) { + for i := range st.Plan.Workflow.Steps { + step := st.Plan.Workflow.Steps[i] + if step.Status != "running" { + continue + } + switch st.policy().mode() { + case HitlDisabled: + st.Plan.Workflow.Steps[i].Status = "failed" + st.Plan.Workflow.Steps[i].Error = "interrupted" + if err := st.save(); err != nil { + return nil, err + } + drv := &flowDriver{opts: e.deps} + out, _, err := drv.failStep(ctx, st, i) + return out, err + case HitlAuto: + if step.Attempts >= 2 { + st.Plan.Workflow.Steps[i].Status = "failed" + st.Plan.Workflow.Steps[i].Error = "interrupted twice" + if err := st.save(); err != nil { + return nil, err + } + drv := &flowDriver{opts: e.deps} + out, _, err := drv.failStep(ctx, st, i) + return out, err + } + st.Plan.Workflow.Steps[i].Status = "pending" + if err := st.save(); err != nil { + return nil, err + } + return nil, nil + default: + drv := &flowDriver{opts: e.deps} + out, _, err := drv.hold(st, HitlConfirmation, step.ID, step.Role, + fmt.Sprintf("步骤「%s」在执行中中断。如何继续？", step.Title), + []HumanOption{ + {Value: "retry", Label: "重试"}, + {Value: "skip", Label: "跳过该步骤"}, + {Value: "abort", Label: "中止运行"}, + }, + "interrupted", "interrupted", nil) + return out, err + } + } + return nil, nil +} + // Resume 恢复运行：有挂起则返回挂起状态；未完成则继续推进（重启恢复入口）。 func (e *SquadEngine) Resume(ctx context.Context, runID string) (*RunResult, error) { - lock := e.runLock(runID) - lock.Lock() - defer lock.Unlock() + release, err := e.acquire(ctx, runID) + if err != nil { + return nil, err + } + defer release() wsDir := filepath.Join(e.opts.DataDir, "runs", runID) st, err := loadRunState(wsDir) @@ -752,11 +894,17 @@ func (e *SquadEngine) Resume(ctx context.Context, runID string) (*RunResult, err if RunStatus(st.Meta.Status).Terminal() { return e.Inspect(runID) } + if out, err := e.recoverInterrupted(ctx, st); out != nil || err != nil { + return out, err + } return e.resumeAfterHold(ctx, st) } // Inspect 查看运行当前状态（含挂起请求与黑板）。 func (e *SquadEngine) Inspect(runID string) (*RunResult, error) { + if !ValidID(runID) { + return nil, invalidID(runID) + } wsDir := filepath.Join(e.opts.DataDir, "runs", runID) st, err := loadRunState(wsDir) if err != nil { @@ -783,6 +931,9 @@ func (e *SquadEngine) Inspect(runID string) (*RunResult, error) { // Blackboard 查看运行黑板。 func (e *SquadEngine) Blackboard(runID string) (*BlackboardState, error) { + if !ValidID(runID) { + return nil, invalidID(runID) + } wsDir := filepath.Join(e.opts.DataDir, "runs", runID) return loadBlackboard(wsDir) } @@ -814,6 +965,9 @@ func (e *SquadEngine) ListRuns() ([]RunMetadata, error) { // Pending 返回运行当前挂起请求（无则 nil）。 func (e *SquadEngine) Pending(runID string) (*HumanRequest, error) { + if !ValidID(runID) { + return nil, invalidID(runID) + } gate, err := loadGate(filepath.Join(e.opts.DataDir, "runs", runID)) if err != nil || gate == nil { return nil, err @@ -827,9 +981,11 @@ func (e *SquadEngine) Steer(runID, text string) error { if strings.TrimSpace(text) == "" { return fmt.Errorf("steering text is empty") } - lock := e.runLock(runID) - lock.Lock() - defer lock.Unlock() + release, err := e.acquire(context.Background(), runID) + if err != nil { + return err + } + defer release() st, err := loadRunState(filepath.Join(e.opts.DataDir, "runs", runID)) if err != nil { @@ -849,9 +1005,11 @@ func (e *SquadEngine) SetBlackboardField(runID, key, value string) error { if strings.TrimSpace(key) == "" { return fmt.Errorf("field key is empty") } - lock := e.runLock(runID) - lock.Lock() - defer lock.Unlock() + release, err := e.acquire(context.Background(), runID) + if err != nil { + return err + } + defer release() st, err := loadRunState(filepath.Join(e.opts.DataDir, "runs", runID)) if err != nil { @@ -883,35 +1041,51 @@ func (e *SquadEngine) SweepWakeups(ctx context.Context) int { if err != nil || gate == nil || gate.Request.Status != "pending" || gate.Request.WaitCondition == nil { continue } - cond := gate.Request.WaitCondition - met, fact := e.evaluateWaitCondition(meta.ID, cond) - if !met { + release, err := e.acquire(ctx, meta.ID) + if err != nil { continue } - gate.Request.WakeupFires++ - if gate.Request.WakeupFires >= MaxWakeupFires { - // 护栏耗尽：自动中止，不无限自动续跑。 - _, _ = e.Respond(ctx, meta.ID, RespondDecision{RequestID: gate.Request.ID, Choice: "abort", - Answer: "自动唤醒次数达上限，自动中止"}) + if e.fireWakeup(ctx, meta.ID) { woken++ - continue - } - _ = writeJSON(filepath.Join(e.opts.DataDir, "runs", meta.ID, "pending.json"), gate) - decision := RespondDecision{RequestID: gate.Request.ID, Answer: "[自动唤醒] " + fact} - switch gate.Request.Kind { - case HitlConfirmation, HitlPermission, HitlQualityReview: - decision.Choice = "approve" // 事实型条件满足视为放行依据 - default: - decision.Choice = "resolved" - } - if _, err := e.Respond(ctx, meta.ID, decision); err == nil { - woken++ - e.deps.Sink.Emit(Event{Type: "wakeup_fired", RunID: meta.ID, Text: fact, At: time.Now()}) } + release() } return woken } +// fireWakeup 在已持有 run 锁时处理一条挂起。只自动放行 gap 与 escalation。 +func (e *SquadEngine) fireWakeup(ctx context.Context, runID string) bool { + gate, err := loadGate(filepath.Join(e.opts.DataDir, "runs", runID)) + if err != nil || gate == nil || gate.Request.Status != "pending" || gate.Request.WaitCondition == nil { + return false + } + met, fact := e.evaluateWaitCondition(runID, gate.Request.WaitCondition) + if !met { + return false + } + gate.Request.WakeupFires++ + if err := writeJSON(filepath.Join(e.opts.DataDir, "runs", runID, "pending.json"), gate); err != nil { + return false + } + if gate.Request.WakeupFires >= MaxWakeupFires { + _, err := e.respondLocked(ctx, runID, RespondDecision{RequestID: gate.Request.ID, Choice: "abort", + Answer: "自动唤醒次数达上限，自动中止"}) + return err == nil + } + switch gate.Request.Kind { + case HitlGap, HitlEscalation: + decision := RespondDecision{RequestID: gate.Request.ID, Choice: "resolved", Answer: "[自动唤醒] " + fact} + if _, err := e.respondLocked(ctx, runID, decision); err != nil { + return false + } + e.deps.Sink.Emit(Event{Type: "wakeup_fired", RunID: runID, Text: fact, At: time.Now()}) + return true + default: + e.deps.Sink.Emit(Event{Type: "wakeup_held", RunID: runID, Text: fact, At: time.Now()}) + return false + } +} + // evaluateWaitCondition 评估等待条件；返回是否满足与事实描述。 func (e *SquadEngine) evaluateWaitCondition(runID string, cond *WaitCondition) (bool, string) { switch cond.Type { diff --git a/squad/hitl.go b/squad/hitl.go index ae3f336..a852f1a 100644 --- a/squad/hitl.go +++ b/squad/hitl.go @@ -75,8 +75,10 @@ type HumanRequest struct { Kind HumanRequestKind `json:"kind"` Severity string `json:"severity,omitempty"` // action_required（默认）/ attention Question string `json:"question"` - Options []HumanOption `json:"options,omitempty"` - Context string `json:"context,omitempty"` // 脱敏摘要（计划/缺口/质检报告） + // Reason 区分同一 Kind 的来源：plan、step_approval、interrupted。空值保持旧行为。 + Reason string `json:"reason,omitempty"` + Options []HumanOption `json:"options,omitempty"` + Context string `json:"context,omitempty"` // 脱敏摘要（计划/缺口/质检报告） // WaitCondition 非空时，条件满足可自动唤醒（人工回应仍然优先）。 WaitCondition *WaitCondition `json:"wait_condition,omitempty"` // WakeupFires 已自动唤醒次数（护栏：达到 MaxWakeupFires 后过期）。 diff --git a/squad/id.go b/squad/id.go new file mode 100644 index 0000000..ad061f2 --- /dev/null +++ b/squad/id.go @@ -0,0 +1,19 @@ +package squad + +import ( + "fmt" + "regexp" + "strings" +) + +// idPattern 限制 run、角色、蓝图 key：首字符为字母或数字，其余为字母、数字、点、下划线、连字符，最长 128，且不得含 ".."。 +var idPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`) + +// ValidID 报告 id 是否可安全拼进数据目录路径。 +func ValidID(id string) bool { + return idPattern.MatchString(id) && !strings.Contains(id, "..") +} + +func invalidID(id string) error { + return fmt.Errorf("invalid id %q", id) +} diff --git a/squad/leader.go b/squad/leader.go index af95d37..9230f55 100644 --- a/squad/leader.go +++ b/squad/leader.go @@ -357,7 +357,9 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *runState) (*RunResu st.saveLeader() // 成员交付快照进 leader inbox（leader 工具沙箱看不到成员目录， // 须物化到 leader 工作区才能自查交付——InputLayoutSnapshot 的对称应用）。 - snapshotDeliverablesToLeader(st, step) + if err := snapshotDeliverablesToLeader(st, step); err != nil { + return nil, nil, err + } // 成员产出后检查阻塞缺口（等待条件挂起在标准 gap 通道上）。 if g2 := drv.checkBlockingGaps(st); g2 != nil { return heldResult(st, g2), g2, nil @@ -386,7 +388,7 @@ func (e *SquadEngine) leaderFail(st *runState, reason string) *RunResult { } // snapshotDeliverablesToLeader 把成员交付文件复制进 leader 工作区 inbox/<步骤id>/。 -func snapshotDeliverablesToLeader(st *runState, step StepSpec) { +func snapshotDeliverablesToLeader(st *runState, step StepSpec) error { srcDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace)) dstDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(leaderSubWorkspace), "inbox", step.ID) for _, t := range step.Targets { @@ -395,9 +397,11 @@ func snapshotDeliverablesToLeader(st *runState, step StepSpec) { continue } dst := filepath.Join(dstDir, filepath.Base(filepath.FromSlash(t))) - _ = os.MkdirAll(filepath.Dir(dst), 0o755) - _ = os.WriteFile(dst, data, 0o644) + if err := AtomicWrite(dst, data); err != nil { + return err + } } + return nil } // parseDelegation 解析 leader 产出的 delegation.json。 diff --git a/squad/lock_other.go b/squad/lock_other.go new file mode 100644 index 0000000..21a4228 --- /dev/null +++ b/squad/lock_other.go @@ -0,0 +1,11 @@ +//go:build !unix + +package squad + +import "context" + +// lockRunFile 在非 unix 平台退回进程内锁（由 SquadEngine.runLock 提供）。 +// 跨进程互斥不可用；调用方不要假设两个进程能靠这把锁互相拒绝。 +func lockRunFile(ctx context.Context, dir string) (func(), error) { + return func() {}, nil +} diff --git a/squad/lock_unix.go b/squad/lock_unix.go new file mode 100644 index 0000000..d667d19 --- /dev/null +++ b/squad/lock_unix.go @@ -0,0 +1,42 @@ +//go:build unix + +package squad + +import ( + "context" + "os" + "path/filepath" + "syscall" + "time" +) + +// lockRunFile 对 runs//.lock 做非阻塞独占锁，并按 ctx 轮询。 +// 直到拿到锁或 ctx 结束。ctx 结束时返回 ErrRunBusy。 +func lockRunFile(ctx context.Context, dir string) (func(), error) { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + f, err := os.OpenFile(filepath.Join(dir, ".lock"), os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return nil, err + } + for { + err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + if err == nil { + return func() { + _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) + _ = f.Close() + }, nil + } + if err != syscall.EWOULDBLOCK && err != syscall.EAGAIN { + _ = f.Close() + return nil, err + } + select { + case <-ctx.Done(): + _ = f.Close() + return nil, ErrRunBusy + case <-time.After(20 * time.Millisecond): + } + } +} diff --git a/squad/models.go b/squad/models.go index 7e7d3fe..289a83f 100644 --- a/squad/models.go +++ b/squad/models.go @@ -90,8 +90,10 @@ type StepSpec struct { Needs []string `json:"needs,omitempty"` Targets []string `json:"targets,omitempty"` // 期望交付物相对路径，缺省 output.md Status string `json:"status"` // pending, running, completed, failed, skipped - DurationMs int64 `json:"duration_ms,omitempty"` - Error string `json:"error,omitempty"` + // Attempts 该步骤进入 running 的次数。中断恢复用它决定 auto 策略是否还能重试。 + Attempts int `json:"attempts,omitempty"` + DurationMs int64 `json:"duration_ms,omitempty"` + Error string `json:"error,omitempty"` // RequiresApproval 为 true 时，该步骤启动前进入 HITL 计划确认。 RequiresApproval bool `json:"requires_approval,omitempty"` // IntentID 非空 = 系统功能步：不经 BackendResolver/智能体，由 IntentRunner 直接执行 diff --git a/squad/recovery_test.go b/squad/recovery_test.go new file mode 100644 index 0000000..ce99d40 --- /dev/null +++ b/squad/recovery_test.go @@ -0,0 +1,294 @@ +package squad + +import ( + "context" + "errors" + "path/filepath" + "sync" + "testing" + "time" + + "agent-native/exec" +) + +func approvalScript() map[string]map[string]string { + return map[string]map[string]string{ + "01_researcher": {"notes.md": "n", "fields.json": `{}`}, + "02_writer": {"output.md": goodReport}, + "03_reviewer": {"output.md": goodReport}, + } +} + +func TestStepApprovalApproveExecutes(t *testing.T) { + steps := baseSteps() + steps[0].RequiresApproval = true + eng, _ := newTestEngine(t, mockScriptPlanner(steps...), approvalScript(), nil) + res, err := eng.Run(context.Background(), Goal{Text: "需要审批"}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunAwaitingConfirmation || res.Pending == nil || res.Pending.Reason != "step_approval" { + t.Fatalf("want step approval hold, got %s %+v", res.Status, res.Pending) + } + res, err = eng.Respond(context.Background(), res.RunID, RespondDecision{Choice: "approve"}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("approve should execute, got %s %s", res.Status, res.Error) + } +} + +func TestStepApprovalSkipDoesNotCancel(t *testing.T) { + steps := baseSteps() + steps[0].RequiresApproval = true + eng, _ := newTestEngine(t, mockScriptPlanner(steps...), approvalScript(), nil) + res, err := eng.Run(context.Background(), Goal{Text: "跳过一步"}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + res, err = eng.Respond(context.Background(), res.RunID, RespondDecision{Choice: "skip"}) + if err != nil { + t.Fatal(err) + } + if res.Status == RunCancelled || res.Status == RunFailed { + t.Fatalf("skip must not cancel the run, got %s %s", res.Status, res.Error) + } + if res.Plan == nil || len(res.Plan.Steps) == 0 || res.Plan.Steps[0].Status != "skipped" { + t.Fatalf("first step should be skipped: %+v", res.Plan) + } +} + +func TestApprovalSurvivesRestart(t *testing.T) { + steps := baseSteps() + steps[0].RequiresApproval = true + eng, dir := newTestEngine(t, mockScriptPlanner(steps...), approvalScript(), nil) + res, err := eng.Run(context.Background(), Goal{Text: "授权要留下"}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := eng.Respond(context.Background(), res.RunID, RespondDecision{Choice: "approve"}); err != nil { + t.Fatal(err) + } + st, err := loadRunState(res.WorkDir) + if err != nil { + t.Fatal(err) + } + if !st.Approved["step:s1"] { + t.Fatalf("approval not stored: %+v", st.Approved) + } + st.Meta.Status = string(RunExecuting) + st.Meta.CompletedAt = nil + wf := st.Workflow() + wf.Status = "running" + wf.Steps[0].Status = "pending" + st.setWorkflow(wf) + if err := st.save(); err != nil { + t.Fatal(err) + } + eng2, err := NewEngine(Options{ + DataDir: dir, Registry: eng.Registry(), Planner: mockScriptPlanner(steps...), + Backends: eng.opts.Backends, Sink: NopSink{}, + }) + if err != nil { + t.Fatal(err) + } + res, err = eng2.Resume(context.Background(), res.RunID) + if err != nil { + t.Fatal(err) + } + if res.Status == RunAwaitingConfirmation { + t.Fatal("restart asked for approval again") + } + if res.Status != RunCompleted { + t.Fatalf("resume after restart: %s %s", res.Status, res.Error) + } +} + +type countBackend struct { + mu sync.Mutex + n int +} + +func (b *countBackend) Name() string { return "script" } + +func (b *countBackend) Execute(ctx context.Context, req exec.Request) (*exec.Result, error) { + b.mu.Lock() + b.n++ + b.mu.Unlock() + return (&scriptBackend{script: approvalScript()}).Execute(ctx, req) +} + +func (b *countBackend) calls() int { + b.mu.Lock() + defer b.mu.Unlock() + return b.n +} + +func TestInterruptedRunningStepStaysHeld(t *testing.T) { + backend := &countBackend{} + dir := t.TempDir() + 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", backend) + eng, err := NewEngine(Options{ + DataDir: dir, Registry: reg, Planner: mockScriptPlanner(baseSteps()...), + Backends: backends, Sink: NopSink{}, + }) + if err != nil { + t.Fatal(err) + } + res, err := eng.Run(context.Background(), Goal{Text: "先跑完再假装中断"}, RunOptions{}) + if err != nil || res.Status != RunCompleted { + t.Fatalf("setup run: %v %+v", err, res) + } + before := backend.calls() + 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[0].Status = "running" + wf.Steps[0].Attempts = 1 + st.setWorkflow(wf) + if err := st.save(); err != nil { + t.Fatal(err) + } + eng2, err := NewEngine(Options{DataDir: dir, Registry: reg, Planner: eng.opts.Planner, Backends: backends, Sink: NopSink{}}) + if err != nil { + t.Fatal(err) + } + res, err = eng2.Resume(context.Background(), res.RunID) + if err != nil { + t.Fatal(err) + } + if res.Status != RunAwaitingConfirmation || res.Pending == nil || res.Pending.Reason != "interrupted" { + t.Fatalf("want interrupted hold, got %s %+v", res.Status, res.Pending) + } + if backend.calls() != before { + t.Fatalf("interrupted step executed again: calls %d -> %d", before, backend.calls()) + } +} + +type blockBackend struct { + started chan struct{} + release chan struct{} + once sync.Once +} + +func (b *blockBackend) Name() string { return "script" } + +func (b *blockBackend) Execute(ctx context.Context, req exec.Request) (*exec.Result, error) { + b.once.Do(func() { close(b.started) }) + select { + case <-b.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + return (&scriptBackend{script: approvalScript()}).Execute(ctx, req) +} + +func TestTwoEnginesOneRunIsBusy(t *testing.T) { + backend := &blockBackend{started: make(chan struct{}), release: make(chan struct{})} + dir := t.TempDir() + reg := NewRoleRegistry() + for _, role := range []string{"researcher", "writer", "reviewer"} { + _ = reg.Register(RoleSpec{Role: role, Backend: "script"}) + } + backends := NewBackendResolver() + backends.RegisterInstance("script", backend) + mk := func() *SquadEngine { + t.Helper() + eng, err := NewEngine(Options{ + DataDir: dir, Registry: reg, Planner: mockScriptPlanner(baseSteps()...), + Backends: backends, Sink: NopSink{}, + }) + if err != nil { + t.Fatal(err) + } + return eng + } + engA := mk() + done := make(chan struct{}) + go func() { + defer close(done) + _, _ = engA.Run(context.Background(), Goal{Text: "占住锁"}, RunOptions{}) + }() + select { + case <-backend.started: + case <-time.After(3 * time.Second): + t.Fatal("run did not start") + } + runs, err := engA.ListRuns() + if err != nil || len(runs) != 1 { + t.Fatalf("runs: %v %+v", err, runs) + } + engB := mk() + ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond) + defer cancel() + _, err = engB.Resume(ctx, runs[0].ID) + if !errors.Is(err, ErrRunBusy) { + t.Fatalf("want busy, got %v", err) + } + close(backend.release) + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("holder did not finish") + } +} + +func TestWakeupDoesNotApprovePermission(t *testing.T) { + var kinds []string + sink := ProgressFunc(func(ev Event) { kinds = append(kinds, ev.Type) }) + steps := gapSteps() + script := map[string]map[string]string{ + "01_researcher": {"notes.md": "n", "fields.json": `{}`, "gaps.json": `[{"item":"等字段","blocking":true,"wait_field":"ext"}]`}, + "02_writer": {"output.md": goodReport}, + "03_reviewer": {"output.md": goodReport}, + } + eng, dir := newTestEngine(t, mockScriptPlanner(steps...), script, sink) + res, err := eng.Run(context.Background(), Goal{Text: "权限不能被唤醒放行"}, RunOptions{}) + if err != nil || res.Status != RunWaitingHuman { + t.Fatalf("setup: %v %+v", err, res) + } + gate, err := loadGate(filepath.Join(dir, "runs", res.RunID)) + if err != nil || gate == nil { + t.Fatal(err) + } + gate.Request.Kind = HitlPermission + gate.Request.WaitCondition = &WaitCondition{Type: "field_set", Field: "ext"} + if err := writeJSON(filepath.Join(dir, "runs", res.RunID, "pending.json"), gate); err != nil { + t.Fatal(err) + } + if err := eng.SetBlackboardField(res.RunID, "ext", "ready"); err != nil { + t.Fatal(err) + } + if n := eng.SweepWakeups(context.Background()); n != 0 { + t.Fatalf("permission wakeup must not count as resolved, got %d", n) + } + insp, err := eng.Inspect(res.RunID) + if err != nil { + t.Fatal(err) + } + if insp.Status != RunWaitingHuman || insp.Pending == nil || insp.Pending.Kind != HitlPermission { + t.Fatalf("permission hold changed: %s %+v", insp.Status, insp.Pending) + } + saw := false + for _, k := range kinds { + if k == "wakeup_held" { + saw = true + } + } + if !saw { + t.Fatalf("expected wakeup_held event, got %v", kinds) + } +} diff --git a/squad/registry.go b/squad/registry.go index d1e5cbe..b05dceb 100644 --- a/squad/registry.go +++ b/squad/registry.go @@ -13,7 +13,7 @@ type RoleRegistry struct { mu sync.RWMutex roles map[string]RoleSpec // OnChange 可选钩子：角色注册/派生后回调（store 层用于落盘）。 - OnChange func(spec RoleSpec) + OnChange func(spec RoleSpec) error } // NewRoleRegistry 创建空注册表。 @@ -23,8 +23,8 @@ func NewRoleRegistry() *RoleRegistry { // Register 注册或更新角色；Role 为空或 BashPermission 非法时返回错误。 func (r *RoleRegistry) Register(spec RoleSpec) error { - if strings.TrimSpace(spec.Role) == "" { - return fmt.Errorf("role key cannot be empty") + if !ValidID(spec.Role) { + return invalidID(spec.Role) } switch spec.BashPermission { case "", "deny", "ask", "allow": @@ -44,7 +44,9 @@ func (r *RoleRegistry) Register(spec RoleSpec) error { r.roles[spec.Role] = spec r.mu.Unlock() if r.OnChange != nil { - r.OnChange(spec) + if err := r.OnChange(spec); err != nil { + return err + } } return nil } diff --git a/squad/runstate.go b/squad/runstate.go index 786cb95..056a113 100644 --- a/squad/runstate.go +++ b/squad/runstate.go @@ -68,7 +68,17 @@ func (st *runState) save() error { return err } } - return nil + return writeJSON(filepath.Join(st.Meta.WorkDir, "runstate.json"), diskRunState{ + Approved: st.Approved, + RepairBudgetRemaining: st.RepairBudgetRemaining, + Clarifications: st.Clarifications, + }) +} + +type diskRunState struct { + Approved map[string]bool `json:"approved,omitempty"` + RepairBudgetRemaining int `json:"repair_budget_remaining,omitempty"` + Clarifications []string `json:"clarifications,omitempty"` } // Workflow 返回计划的编排清单（简洁访问器）。 @@ -112,6 +122,19 @@ func loadRunState(wsDir string) (*runState, error) { if st.Approved == nil { st.Approved = map[string]bool{} } + if data, err := os.ReadFile(filepath.Join(wsDir, "runstate.json")); err == nil { + var disk diskRunState + if err := json.Unmarshal(data, &disk); err != nil { + return nil, fmt.Errorf("parse runstate.json: %w", err) + } + if disk.Approved != nil { + st.Approved = disk.Approved + } + st.RepairBudgetRemaining = disk.RepairBudgetRemaining + st.Clarifications = disk.Clarifications + } else if !os.IsNotExist(err) { + return nil, err + } st.loadSteering() if data, err := os.ReadFile(filepath.Join(wsDir, "usage.json")); err == nil { _ = json.Unmarshal(data, &st.Usage) diff --git a/squad/workflow.go b/squad/workflow.go index d8da6ca..ef8aeda 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -52,7 +52,7 @@ func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult, {Value: "reject", Label: "拒绝，步骤失败"}, {Value: "abort", Label: "中止运行"}, }, - fmt.Sprintf("步骤：%s\n角色：%s\n禁忌：%v", step.Title, role.Role, role.ForbiddenActions), nil) + fmt.Sprintf("步骤：%s\n角色：%s\n禁忌：%v", step.Title, role.Role, role.ForbiddenActions), "", nil) } } @@ -71,7 +71,7 @@ func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult, {Value: "skip", Label: "跳过该步骤"}, {Value: "abort", Label: "中止运行"}, }, - fmt.Sprintf("角色：%s\n产出：%v", step.Role, step.Targets), nil) + fmt.Sprintf("角色：%s\n产出：%v", step.Role, step.Targets), "step_approval", nil) } } @@ -180,8 +180,16 @@ 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.Error = "" - st.setStep(idx, step) + // 与收口落盘同一把锁：并行步骤一边改 Steps、一边全量 marshal 会数据竞争。 + st.mu.Lock() + st.Plan.Workflow.Steps[idx] = step + saveErr := st.save() + st.mu.Unlock() + if saveErr != nil { + return nil, nil, saveErr + } started := time.Now() prompt := readFileOr(filepath.Join(subDir, "TASK.md"), step.Title) @@ -655,11 +663,11 @@ func waitConditionFrom(gaps []GapItem) *WaitCondition { } // hold 生成挂起请求并返回结果。 -func (d *flowDriver) hold(st *runState, kind HumanRequestKind, stepID, role, question string, options []HumanOption, context string, extra *hitlGate) (*RunResult, *hitlGate, error) { +func (d *flowDriver) hold(st *runState, kind HumanRequestKind, stepID, role, question string, options []HumanOption, context, reason string, extra *hitlGate) (*RunResult, *hitlGate, error) { req := &HumanRequest{ ID: newRequestID(), RunID: st.Meta.ID, StepID: stepID, Kind: kind, Severity: SeverityActionRequired, - Question: question, Options: options, Context: context, + Question: question, Options: options, Context: context, Reason: reason, CreatedAt: time.Now(), Status: "pending", } status := RunWaitingHuman diff --git a/squad/workspace.go b/squad/workspace.go index a9f6862..686ddf0 100644 --- a/squad/workspace.go +++ b/squad/workspace.go @@ -309,5 +309,41 @@ func writeJSON(path string, v any) error { if err != nil { return err } - return os.WriteFile(path, data, 0o644) + data = append(data, '\n') + return AtomicWrite(path, data) +} + +// AtomicWrite 在同目录创建临时文件，写入后 Sync、Close，再 rename 到目标路径。 +func AtomicWrite(path string, data []byte) error { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o755); err != nil { + return err + } + f, err := os.CreateTemp(dir, ".tmp-*") + if err != nil { + return err + } + tmp := f.Name() + ok := false + defer func() { + if !ok { + _ = os.Remove(tmp) + } + }() + if _, err := f.Write(data); err != nil { + _ = f.Close() + return err + } + if err := f.Sync(); err != nil { + _ = f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + if err := os.Rename(tmp, path); err != nil { + return err + } + ok = true + return nil } diff --git a/store/store.go b/store/store.go index dd5b480..f90ae1c 100644 --- a/store/store.go +++ b/store/store.go @@ -56,12 +56,12 @@ func (s *FileStore) LoadRoles(reg *squad.RoleRegistry) error { } // SaveRole 落盘单个角色。 -func (s *FileStore) SaveRole(spec squad.RoleSpec) { +func (s *FileStore) SaveRole(spec squad.RoleSpec) error { data, err := json.MarshalIndent(spec, "", " ") if err != nil { - return + return err } - _ = os.WriteFile(filepath.Join(s.DataDir, "roles.d", spec.Role+".json"), data, 0o644) + return squad.AtomicWrite(filepath.Join(s.DataDir, "roles.d", spec.Role+".json"), append(data, '\n')) } // DeleteRole 删除角色文件（注册表删除由调用方完成）。 @@ -96,12 +96,12 @@ func (s *FileStore) LoadBlueprints(bps *squad.BlueprintRegistry) error { } // SaveBlueprint 落盘单个蓝图。 -func (s *FileStore) SaveBlueprint(bp squad.Blueprint) { +func (s *FileStore) SaveBlueprint(bp squad.Blueprint) error { data, err := json.MarshalIndent(bp, "", " ") if err != nil { - return + return err } - _ = os.WriteFile(filepath.Join(s.DataDir, "blueprints.d", bp.Key+".json"), data, 0o644) + return squad.AtomicWrite(filepath.Join(s.DataDir, "blueprints.d", bp.Key+".json"), append(data, '\n')) } // DeleteBlueprint 删除蓝图文件。