diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index a4a9cfd..8e79f04 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -6,19 +6,48 @@ on: branches: [main] jobs: vet-and-test: - runs-on: nightly + runs-on: ubuntu-latest + env: + GOPROXY: https://goproxy.cn,direct steps: - - uses: actions/checkout@v4 - with: - submodules: true - - run: go vet ./... && go test -count=1 ./... + - name: Checkout + env: + TOKEN: ${{ secrets.GITHUB_TOKEN }} + SUBMODULE_TOKEN: ${{ secrets.SUBMODULE_TOKEN }} + run: | + set -euo pipefail + git init -q + git remote add origin "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git" + git -c http.extraheader="AUTHORIZATION: token ${TOKEN}" fetch --depth=1 origin "${GITHUB_SHA}" + git checkout -q --detach FETCH_HEAD + if [ -z "${SUBMODULE_TOKEN}" ]; then + echo "SUBMODULE_TOKEN is not configured" >&2 + exit 1 + fi + git config --global url."https://weichen:${SUBMODULE_TOKEN}@git-home.u1680832.nyat.app:30319/".insteadOf "ssh://git@git-ssh.u1680832.nyat.app:31411/" + git -c fetch.recurseSubmodules=false submodule update --init --recursive + - name: Install Go 1.27.1 + run: | + set -euo pipefail + curl -fsSL -o "$RUNNER_TEMP/go.tgz" https://mirrors.aliyun.com/golang/go1.27.1.linux-amd64.tar.gz + tar -C /usr/local -xzf "$RUNNER_TEMP/go.tgz" + ln -sfn /usr/local/go/bin/go /usr/local/bin/go + ln -sfn /usr/local/go/bin/gofmt /usr/local/bin/gofmt + - name: Vet, test, and race + run: GOWORK=off go vet ./... && GOWORK=off go test -count=1 ./... && GOWORK=off go test -count=1 -race ./... docker-checkout-hostgateway: runs-on: ubuntu-latest options: --add-host gitea:host-gateway env: GITHUB_SERVER_URL: http://gitea:3000 steps: - - uses: actions/checkout@v4 - with: - submodules: true - - run: test -f go.mod && test -f Agent-Native/go.mod && echo "docker-mode checkout via host-gateway OK" + - name: Checkout without GitHub actions + env: + TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + set -euo pipefail + git init -q + git remote add origin "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git" + git -c http.extraheader="AUTHORIZATION: token ${TOKEN}" fetch --depth=1 origin "${GITHUB_SHA}" + git checkout -q --detach FETCH_HEAD + test -f go.mod && echo "docker-mode checkout via host-gateway OK" 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..6c86be0 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 err := listenGuard(*addr, *token, *insecure); err != nil { + fmt.Fprintln(os.Stderr, err) + 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,22 @@ func cmdServe(args []string) int { return httpListen(*addr, srv.Handler()) } +func listenGuard(addr, token string, insecure bool) error { + if token == "" && !insecure && !loopbackAddr(addr) { + return fmt.Errorf("拒绝在没有 token 时监听非回环地址;传入 --token 或 --insecure") + } + return nil +} + +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/cli/listen_test.go b/cli/listen_test.go new file mode 100644 index 0000000..78e4fb7 --- /dev/null +++ b/cli/listen_test.go @@ -0,0 +1,28 @@ +package cli + +import "testing" + +func TestListenGuard(t *testing.T) { + cases := []struct { + name string + addr string + token string + insecure bool + ok bool + }{ + {name: "loopback without token", addr: "127.0.0.1:8090", ok: true}, + {name: "public without token", addr: "0.0.0.0:8090", ok: false}, + {name: "public with insecure", addr: "0.0.0.0:8090", insecure: true, ok: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := listenGuard(tc.addr, tc.token, tc.insecure) + if tc.ok && err != nil { + t.Fatal(err) + } + if !tc.ok && err == nil { + t.Fatal("expected refusal") + } + }) + } +} diff --git a/deps/agent-intent b/deps/agent-intent index 3674dc4..7f7879b 160000 --- a/deps/agent-intent +++ b/deps/agent-intent @@ -1 +1 @@ -Subproject commit 3674dc4bfd0b49191d21f0afab4002381e41a429 +Subproject commit 7f7879b06b053f0a62746309b79e0e1b9a00f42d diff --git a/deps/agent-native b/deps/agent-native index a34689c..91c1556 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit a34689c78fbddbfa34b343cd6f85cca36c4d2f9d +Subproject commit 91c155618487e21edeb4f1d3cc27e985aae2b0d2 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/atomic_test.go b/squad/atomic_test.go new file mode 100644 index 0000000..88b6e21 --- /dev/null +++ b/squad/atomic_test.go @@ -0,0 +1,56 @@ +package squad + +import ( + "os" + "path/filepath" + "testing" +) + +func TestAtomicWriteLeavesNoTemp(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "runstate.json") + if err := AtomicWrite(path, []byte(`{"ok":true}`)); err != nil { + t.Fatal(err) + } + got, err := os.ReadFile(path) + if err != nil || string(got) != `{"ok":true}` { + t.Fatalf("content: %v %s", err, got) + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + for _, e := range entries { + if e.Name() != "runstate.json" { + t.Fatalf("temp file left behind: %s", e.Name()) + } + } +} + +func TestAtomicWriteRenameFailureKeepsOriginal(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "occupied") + if err := os.Mkdir(path, 0o755); err != nil { + t.Fatal(err) + } + kept := filepath.Join(path, "keep.txt") + if err := os.WriteFile(kept, []byte("stay"), 0o644); err != nil { + t.Fatal(err) + } + if err := AtomicWrite(path, []byte("new")); err == nil { + t.Fatal("rename onto a non-empty directory should fail") + } + got, err := os.ReadFile(kept) + if err != nil || string(got) != "stay" { + t.Fatalf("original changed: %v %s", err, got) + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + for _, e := range entries { + if e.Name() != "occupied" { + t.Fatalf("temp file left behind: %s", e.Name()) + } + } +} 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..076370f 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 { @@ -475,18 +509,74 @@ func (e *SquadEngine) Respond(ctx context.Context, runID string, decision Respon } req := &gate.Request + if strings.TrimSpace(decision.Choice) == "" && strings.TrimSpace(decision.Answer) != "" && choiceHoldKind(req.Kind) { + return e.noteAndKeepHold(st, gate, decision) + } req.Status = "answered" now := time.Now() req.AnsweredAt = &now + if err := st.saveGate(gate); err != nil { + return nil, err + } e.deps.Sink.Emit(Event{Type: EvHumanResponded, RunID: runID, Text: firstNonEmpty(decision.Choice, decision.Answer), Data: marshalJSON(decision), At: time.Now()}) - _ = st.saveGate(nil) + res, err := e.applyDecision(ctx, st, gate, decision) + if err != nil { + req.Status = "pending" + req.AnsweredAt = nil + _ = st.saveGate(gate) + return nil, err + } + if err := clearSameGate(st, req.ID); err != nil { + return nil, err + } + return res, nil +} + +func choiceHoldKind(kind HumanRequestKind) bool { + switch kind { + case HitlConfirmation, HitlPermission, HitlQualityReview: + return true + default: + return false + } +} + +func (e *SquadEngine) noteAndKeepHold(st *runState, gate *hitlGate, decision RespondDecision) (*RunResult, error) { + if err := st.addSteering(decision.Answer); err != nil { + return nil, err + } + gate.Request.Status = "pending" + gate.Request.AnsweredAt = nil + if err := st.saveGate(gate); err != nil { + return nil, err + } + return heldResult(st, gate), nil +} + +func clearSameGate(st *runState, requestID string) error { + current, err := loadGate(st.Meta.WorkDir) + if err != nil || current == nil || current.Request.ID != requestID { + return err + } + return st.saveGate(nil) +} + +func (e *SquadEngine) applyDecision(ctx context.Context, st *runState, gate *hitlGate, decision RespondDecision) (*RunResult, error) { + req := &gate.Request + now := time.Now() drv := &flowDriver{opts: e.deps} switch req.Kind { 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 { @@ -622,8 +712,8 @@ func (e *SquadEngine) Respond(ctx context.Context, runID string, decision Respon wf.Status = "failed" st.setWorkflow(wf) _ = st.save() - e.deps.Sink.Emit(Event{Type: EvRunFailed, RunID: runID, Text: "人工裁决质检失败", At: time.Now()}) - return &RunResult{RunID: runID, Status: RunFailed, Plan: planPtr(st), + e.deps.Sink.Emit(Event{Type: EvRunFailed, RunID: st.Meta.ID, Text: "人工裁决质检失败", At: time.Now()}) + return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), Error: "quality review rejected by human", WorkDir: st.Meta.WorkDir}, nil } default: @@ -734,11 +824,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 +944,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 +981,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 +1015,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 +1031,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 +1055,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 +1091,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..fcab0a1 --- /dev/null +++ b/squad/recovery_test.go @@ -0,0 +1,409 @@ +package squad + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "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) + } +} + +func TestParallelGroupSecondStepApprovalHolds(t *testing.T) { + steps := baseSteps() + steps[0].Parallel = true + steps[1].Parallel = true + steps[1].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.StepID != "s2" { + t.Fatalf("group should hold on the second step, got %s %+v", res.Status, res.Pending) + } + if _, err := os.Stat(filepath.Join(res.WorkDir, "tasks", "01_researcher", "notes.md")); !os.IsNotExist(err) { + t.Fatalf("parallel group ran before approval: %v", err) + } +} + +func TestParallelGroupSecondStepPermissionHolds(t *testing.T) { + dir := t.TempDir() + reg := NewRoleRegistry() + for _, spec := range []RoleSpec{ + {Role: "researcher", Name: "调研", Backend: "script"}, + {Role: "writer", Name: "撰写", Backend: "script", BashPermission: BashPermissionAsk}, + {Role: "reviewer", Name: "审核", Backend: "script"}, + } { + if err := reg.Register(spec); err != nil { + t.Fatal(err) + } + } + backends := NewBackendResolver() + backends.RegisterInstance("script", &scriptBackend{script: approvalScript()}) + steps := baseSteps() + steps[0].Parallel = true + steps[1].Parallel = true + eng, err := NewEngine(Options{ + DataDir: dir, Registry: reg, Planner: mockScriptPlanner(steps...), Backends: backends, Sink: NopSink{}, + }) + if err != nil { + t.Fatal(err) + } + var res *RunResult + res, err = eng.Run(context.Background(), Goal{Text: "并行组第二步要授权"}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunWaitingHuman || res.Pending == nil || res.Pending.Kind != HitlPermission || res.Pending.StepID != "s2" { + t.Fatalf("group should hold on writer permission, got %s %+v", res.Status, res.Pending) + } + if _, err := os.Stat(filepath.Join(res.WorkDir, "tasks", "01_researcher", "notes.md")); !os.IsNotExist(err) { + t.Fatalf("parallel group ran before permission: %v", err) + } +} + +func TestFreeTextKeepsChoiceHold(t *testing.T) { + for _, kind := range []HumanRequestKind{HitlConfirmation, HitlPermission, HitlQualityReview} { + t.Run(string(kind), func(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 || res.Pending == nil { + t.Fatalf("setup: %v %+v", err, res) + } + work := filepath.Join(dir, "runs", res.RunID) + gate, err := loadGate(work) + if err != nil || gate == nil { + t.Fatal(err) + } + id := gate.Request.ID + gate.Request.Kind = kind + if err := writeJSON(filepath.Join(work, "pending.json"), gate); err != nil { + t.Fatal(err) + } + res, err = eng.Respond(context.Background(), res.RunID, RespondDecision{Answer: "补充一句"}) + if err != nil { + t.Fatal(err) + } + if res.Pending == nil || res.Pending.ID != id || res.Pending.Status != "pending" { + t.Fatalf("hold was cleared: %+v", res.Pending) + } + data, err := os.ReadFile(filepath.Join(work, "steering.json")) + if err != nil || !strings.Contains(string(data), "补充一句") { + t.Fatalf("steering not recorded: %v %s", err, data) + } + }) + } +} + +func TestFailedDecisionRestoresPending(t *testing.T) { + 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, _ := newTestEngine(t, mockScriptPlanner(steps...), script, nil) + res, err := eng.Run(context.Background(), Goal{Text: "错误选择后仍可回应"}, RunOptions{}) + if err != nil || res.Pending == nil { + t.Fatalf("setup: %v %+v", err, res) + } + if _, err := eng.Respond(context.Background(), res.RunID, RespondDecision{Choice: "nope"}); err == nil { + t.Fatal("unknown choice should fail") + } + res, err = eng.Respond(context.Background(), res.RunID, RespondDecision{Choice: "resolved"}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("restored hold should still be answerable, got %s %s", res.Status, res.Error) + } +} 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..3b5e564 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) @@ -170,6 +193,18 @@ func (st *runState) injectSteering(prompt string) string { return b.String() } +// addSteering 追加一条尚未注入的人工补充,并立刻落盘。 +func (st *runState) addSteering(text string) error { + text = strings.TrimSpace(text) + if text == "" { + return nil + } + st.mu.Lock() + defer st.mu.Unlock() + st.Steering = append(st.Steering, text) + return writeJSON(filepath.Join(st.Meta.WorkDir, "steering.json"), st.Steering) +} + // loadSteering 恢复未消费的补充指令。 func (st *runState) loadSteering() { data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "steering.json")) diff --git a/squad/workflow.go b/squad/workflow.go index d8da6ca..87af9b0 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -36,51 +36,25 @@ func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult, continue } - // HITL:角色权限升级(bash=ask 且本运行未授权过该角色)。 - role, _ := findAgent(st.Plan.Squad.Agents, step.Role) - if role.BashPermission == BashPermissionAsk && !st.Approved["role:"+role.Role] { - switch d.policyRole(st, role).mode() { - case HitlDisabled, HitlAuto: - step.Error = "permission denied: role " + role.Role + " requires bash approval (hitl mode=" + string(d.policyRole(st, role).mode()) + ")" - st.Plan.Workflow.Steps[i] = step - return d.failStep(ctx, st, i) - default: - return d.hold(st, HitlPermission, step.ID, step.Role, - fmt.Sprintf("角色 %s(%s)声明 bash_permission=ask,是否授权其在本次运行中使用更高权限?", role.Name, role.Role), - []HumanOption{ - {Value: "approve", Label: "授权本次运行"}, - {Value: "reject", Label: "拒绝,步骤失败"}, - {Value: "abort", Label: "中止运行"}, - }, - fmt.Sprintf("步骤:%s\n角色:%s\n禁忌:%v", step.Title, role.Role, role.ForbiddenActions), nil) - } - } - - // HITL:步骤显式要求审批(角色级策略参与判定)。 - if step.RequiresApproval && !st.Approved["step:"+step.ID] { - switch d.policyRole(st, role).mode() { - case HitlDisabled, HitlAuto: - step.Error = "approval required but hitl mode is " + string(d.policy(st).mode()) - st.Plan.Workflow.Steps[i] = step - return d.failStep(ctx, st, i) - default: - return d.hold(st, HitlConfirmation, step.ID, step.Role, - fmt.Sprintf("步骤「%s」标记为需人工审批后才能执行,是否批准?", step.Title), - []HumanOption{ - {Value: "approve", Label: "批准执行"}, - {Value: "skip", Label: "跳过该步骤"}, - {Value: "abort", Label: "中止运行"}, - }, - fmt.Sprintf("角色:%s\n产出:%v", step.Role, step.Targets), nil) + if out, g, err := d.gateBeforeRun(ctx, st, i); out != nil || g != nil || err != nil { + if out == nil && g != nil { + out = heldResult(st, g) } + return out, g, err } if step.Parallel { - // 收集相邻并行组,组内并发、组间顺序。 + // 收集相邻并行组。组内每一步都先过审批和权限,任一需要挂起则整组不执行。 var group []int for i < len(st.Plan.Workflow.Steps) { s := st.Plan.Workflow.Steps[i] if s.Parallel && s.Status != "completed" && s.Status != "skipped" && s.Status != "failed" { + if out, g, err := d.gateBeforeRun(ctx, st, i); out != nil || g != nil || err != nil { + if out == nil && g != nil { + out = heldResult(st, g) + } + return out, g, err + } group = append(group, i) i++ continue @@ -128,6 +102,48 @@ func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult, return d.finalize(ctx, st) } +// gateBeforeRun 在步骤真正执行前检查权限升级和显式审批。 +// 返回非空结果或挂起时,调用方不得执行该步骤,也不得执行它所在的并行组。 +func (d *flowDriver) gateBeforeRun(ctx context.Context, st *runState, i int) (*RunResult, *hitlGate, error) { + step := st.Plan.Workflow.Steps[i] + role, _ := findAgent(st.Plan.Squad.Agents, step.Role) + if role.BashPermission == BashPermissionAsk && !st.Approved["role:"+role.Role] { + switch d.policyRole(st, role).mode() { + case HitlDisabled, HitlAuto: + step.Error = "permission denied: role " + role.Role + " requires bash approval (hitl mode=" + string(d.policyRole(st, role).mode()) + ")" + st.Plan.Workflow.Steps[i] = step + return d.failStep(ctx, st, i) + default: + return d.hold(st, HitlPermission, step.ID, step.Role, + fmt.Sprintf("角色 %s(%s)声明 bash_permission=ask,是否授权其在本次运行中使用更高权限?", role.Name, role.Role), + []HumanOption{ + {Value: "approve", Label: "授权本次运行"}, + {Value: "reject", Label: "拒绝,步骤失败"}, + {Value: "abort", Label: "中止运行"}, + }, + fmt.Sprintf("步骤:%s\n角色:%s\n禁忌:%v", step.Title, role.Role, role.ForbiddenActions), "", nil) + } + } + if step.RequiresApproval && !st.Approved["step:"+step.ID] { + switch d.policyRole(st, role).mode() { + case HitlDisabled, HitlAuto: + step.Error = "approval required but hitl mode is " + string(d.policy(st).mode()) + st.Plan.Workflow.Steps[i] = step + return d.failStep(ctx, st, i) + default: + return d.hold(st, HitlConfirmation, step.ID, step.Role, + fmt.Sprintf("步骤「%s」标记为需人工审批后才能执行,是否批准?", step.Title), + []HumanOption{ + {Value: "approve", Label: "批准执行"}, + {Value: "skip", Label: "跳过该步骤"}, + {Value: "abort", Label: "中止运行"}, + }, + fmt.Sprintf("角色:%s\n产出:%v", step.Role, step.Targets), "step_approval", nil) + } + } + return nil, nil, nil +} + // runParallel 并发执行一个并行组;关键步骤失败即终止整个运行。 func (d *flowDriver) runParallel(ctx context.Context, st *runState, idxs []int) (*RunResult, *hitlGate, error) { var wg sync.WaitGroup @@ -180,8 +196,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 +679,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 删除蓝图文件。