diff --git a/.gitmodules b/.gitmodules index c53df3e..624b30f 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,6 +1,12 @@ [submodule "Agent-Native"] - path = Agent-Native + path = deps/agent-native url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Native.git [submodule "deps/development-guidelines"] path = deps/development-guidelines url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Guidelines.git +[submodule "deps/error-code-system"] + path = deps/error-code-system + url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Error-Code-System.git +[submodule "deps/logging-system"] + path = deps/logging-system + url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Logging-System.git diff --git a/Agent-Native b/Agent-Native deleted file mode 160000 index 49e6dd1..0000000 --- a/Agent-Native +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 49e6dd1e0fc65de7f259f21d2f63c56e5b165a48 diff --git a/PROJECT.md b/PROJECT.md index 38d9de6..7cc108c 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -18,14 +18,14 @@ ## 依赖组合 -| 逻辑 ID | 当前路径 | 接入后的路径 | 角色 | -|---|---|---|---| -| agent-native | `Agent-Native` | `deps/agent-native` | build-required | -| development-guidelines | `deps/development-guidelines` | 同左 | development-only | -| error-code-system | 尚未挂载 | `deps/error-code-system` | build-required | -| logging-system | 尚未挂载 | `deps/logging-system` | build-required | +| 逻辑 ID | 路径 | 角色 | +|---|---|---| +| agent-native | `deps/agent-native` | build-required | +| development-guidelines | `deps/development-guidelines` | development-only | +| error-code-system | `deps/error-code-system` | build-required | +| logging-system | `deps/logging-system` | build-required | -`go.mod` 目前为 `replace agent-native => ./Agent-Native`。迁移到 `deps/` 后改为相对 `deps/` 的 replace，并在同一次提交更新本表。 +`go.mod` 的 `replace` 都指向 `./deps/`。嵌套在 `deps/agent-native/deps/` 里的同名副本不参与本仓编译，提交必须与本仓 gitlink 一致（`scripts/submodule-consistency.sh`）。 克隆： diff --git a/README.md b/README.md index 7361d0d..ef2cfa3 100644 --- a/README.md +++ b/README.md @@ -7,20 +7,31 @@ 设计参考 Multica 的 agent orchestration 理念与 Agent-System `internal/squad` 的成熟机制（黑板/沙箱工作区/顺序并行混合调度/动态调整），按"空表起步"通用化重写。 -本仓库是该模块的源码位置。Go 模块路径为 `agent-squad`。智能体执行依赖子模块 [Agent-Native](./Agent-Native)（`replace agent-native => ./Agent-Native`），钉在固定提交，不随 Agent-Native 分支浮动。 +本仓库是该模块的源码位置。Go 模块路径为 `agent-squad`。依赖都在 `deps/`，由本仓 gitlink 固定，不随上游分支浮动： -克隆时带上子模块，否则无法编译： +| 路径 | 作用 | +|---|---| +| `deps/agent-native` | 智能体执行。`replace agent-native => ./deps/agent-native` | +| `deps/error-code-system` | HTTP 错误信封与规范码 | +| `deps/logging-system` | 组合根日志。`trace_id` 来自 `X-Trace-Id`，`run_id` 标识一次编排 | +| `deps/development-guidelines` | 开发规范，不参与编译 | + +Agent-Native 自己的 `deps/` 里还有一份错误码库、日志库和规范。本仓编译只使用本仓 `go.mod` 的 `replace`，嵌套目录里的 `replace` 不生效。`scripts/submodule-consistency.sh` 要求两侧 gitlink 相同，避免两份副本锁到不同提交。 + +克隆时带上子模块，否则无法编译。已有克隆若仍留着根目录 `Agent-Native`，先删掉该残留目录再初始化： ```bash git clone --recurse-submodules ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Squad.git -# 已克隆时：git submodule update --init +# 已克隆时： +git submodule sync --recursive +git submodule update --init --recursive ``` ## 与周边仓库的关系 | 仓库/模块 | 关系 | |---|---| -| Agent-Native（本仓 submodule `./Agent-Native`） | **智能体执行的唯一来源**：角色执行经 agent-native `adapters/native`（自研内核 runtime 循环 + 工作区工具）或产品注册的其他 agent-native 后端（如 pi-agent）；LLM 规划器复用 `runtime/llm` 客户端做编排决策。本库不实现任何 Agent——只做编排与任务拆解。依赖方向 agent-squad → agent-native，反向无依赖 | +| Agent-Native（本仓 submodule `deps/agent-native`） | **智能体执行的唯一来源**：角色执行经 agent-native `adapters/native`（自研内核 runtime 循环 + 工作区工具）或产品注册的其他 agent-native 后端（如 pi-agent）；LLM 规划器复用 `runtime/llm` 客户端做编排决策。本库不实现任何 Agent——只做编排与任务拆解。依赖方向 agent-squad → agent-native，反向无依赖 | | `agent-system-multi/modules/squad`（U4 产品小队暂存仓） | **不在本仓**，互不依赖。产品小队含医疗器械预设角色与任务键，属产品层；后续产品仓接入本底座时，预设角色/任务模板改为向本底座注册（独立一期） | | 主仓 `internal/squad` | 机制语义（工作流驱动/黑板/子工作区/防篡改清单）的权威来源；本底座为通用化重写，不做逐行同步 | @@ -168,7 +179,7 @@ LLM 配置走环境变量：`AGENT_SQUAD_LLM_BASE_URL` / `AGENT_SQUAD_LLM_API_KE ## 快速开始 -在仓库根目录（需已初始化 `Agent-Native` 子模块）： +在仓库根目录（需已初始化 `deps/` 下的子模块）： ```bash go build -o agent-squad . @@ -185,7 +196,8 @@ go build -o agent-squad . ## 验证 ```bash -go vet ./... && go test -count=1 ./... +GOWORK=off go vet ./... && GOWORK=off go test -count=1 ./... +./scripts/submodule-consistency.sh ``` 当前全绿（conformance 全链路 + HITL 各挂起点 + server + store + textextract）。 diff --git a/cli/cli.go b/cli/cli.go index da3749b..020b5b3 100644 --- a/cli/cli.go +++ b/cli/cli.go @@ -18,6 +18,9 @@ import ( "agent-squad/server" "agent-squad/squad" "agent-squad/store" + + "logging-system/config" + "logging-system/logsys" ) // Main CLI 入口；返回进程退出码。 @@ -74,6 +77,7 @@ type app struct { sink *squad.FanoutSink files *store.FileEventSink st *store.FileStore + log *logsys.Logger } func dataDir() string { @@ -126,7 +130,24 @@ func assemble(dir, plannerName, modelOverride string) (*app, error) { if err != nil { return nil, err } - return &app{engine: engine, sink: fanout, files: fileSink, st: st}, nil + logCfg, err := config.Load("") + if err != nil { + return nil, err + } + if logCfg.Service == "" { + logCfg.Service = "agent-squad" + } + logger, err := config.Build(logCfg) + if err != nil { + return nil, err + } + return &app{engine: engine, sink: fanout, files: fileSink, st: st, log: logger}, nil +} + +func (a *app) close() { + if a != nil && a.log != nil { + _ = a.log.Close() + } } func first(vals ...string) string { @@ -154,7 +175,9 @@ func cmdServe(args []string) int { fmt.Fprintln(os.Stderr, "组装失败:", err) return 1 } + defer a.close() srv := server.New(a.engine, a.sink, a.st) + srv.SetLogger(a.log) stopSweep := srv.StartWakeupSweeper(15 * time.Second) defer stopSweep() fmt.Printf("agent-squad 服务启动 http://localhost%s (data=%s, roles=%d, blueprints=%d, wakeup-sweep=15s)\n", @@ -380,6 +403,7 @@ func cmdRun(args []string) int { fmt.Fprintln(os.Stderr, "组装失败:", err) return 1 } + defer a.close() goal := squad.Goal{Text: goalText, HitlMode: squad.HitlMode(*hitlMode)} if *materialsDir != "" { ents, err := os.ReadDir(*materialsDir) @@ -417,6 +441,9 @@ func cmdRun(args []string) int { fmt.Fprintln(os.Stderr, "运行失败:", err) return 1 } + if a.log != nil && res != nil { + a.log.Module("cli").With("run_id", res.RunID).Info("run finished", "status", string(res.Status)) + } printResult(res) if res.Status == squad.RunFailed { return 1 diff --git a/deps/agent-native b/deps/agent-native new file mode 160000 index 0000000..e65c6ba --- /dev/null +++ b/deps/agent-native @@ -0,0 +1 @@ +Subproject commit e65c6ba604c9079e1176d9449e295e0d09bedfd3 diff --git a/deps/error-code-system b/deps/error-code-system new file mode 160000 index 0000000..b67696d --- /dev/null +++ b/deps/error-code-system @@ -0,0 +1 @@ +Subproject commit b67696d393e7abfbdd16fdde1064c07f5ecd2f72 diff --git a/deps/logging-system b/deps/logging-system new file mode 160000 index 0000000..7f59720 --- /dev/null +++ b/deps/logging-system @@ -0,0 +1 @@ +Subproject commit 7f5972007bdcb6047ce7f95ee891f53e0af9d136 diff --git a/go.mod b/go.mod index a83f714..db19e60 100644 --- a/go.mod +++ b/go.mod @@ -2,6 +2,14 @@ module agent-squad go 1.27 -require agent-native v0.0.0 +require ( + agent-native v0.0.0 + error-code-system v0.0.0 + logging-system v0.0.0 +) -replace agent-native => ./Agent-Native +replace agent-native => ./deps/agent-native + +replace error-code-system => ./deps/error-code-system + +replace logging-system => ./deps/logging-system diff --git a/policy/guidelines-adoption.md b/policy/guidelines-adoption.md index 114d657..2b442f8 100644 --- a/policy/guidelines-adoption.md +++ b/policy/guidelines-adoption.md @@ -7,7 +7,7 @@ - 过程文档位置：`docs/records/`（项目内固定目录豁免。个人维护的编排库，任务主记录仍在 Gitea Issue/PR。过程记录不进规范库，也不写成与 Issue 平行的第二状态。） - 版本权威：父仓 gitlink；可读标签：无（内容版本 `0.1.0-draft.6`，提交 `26b562d2e34fbc560711296a309c91c74ffc55b1`） - 批准/生效依据：本仓接入提交 -- 替代范围和迁移说明：本仓此前没有根 `AGENTS.md`。Agent-Native 子模块将从仓库根迁到 `deps/agent-native`，迁移说明见 `docs/records/2026-09-29-guidelines-and-observability.md`。 +- 替代范围和迁移说明：本仓此前没有根 `AGENTS.md`。Agent-Native 子模块已从仓库根迁到 `deps/agent-native`，迁移说明见 `docs/records/2026-09-29-guidelines-and-observability.md`。 | 规范/工作流 | 适用条件与范围 | 采纳或不适用理由 | 执行方式/证据 | |---|---|---|---| diff --git a/scripts/submodule-consistency.sh b/scripts/submodule-consistency.sh new file mode 100755 index 0000000..b363f41 --- /dev/null +++ b/scripts/submodule-consistency.sh @@ -0,0 +1,35 @@ +#!/bin/sh +# 比较本仓 deps 与嵌套在 deps/agent-native 里的同名子模块 gitlink。 +# 编译只用本仓 go.mod 的 replace；这里防止两份副本锁到不同提交。 +set -eu +cd "$(dirname "$0")/.." +fail=0 + +compare() { + label=$1 + outer=$2 + inner=$3 + recorded=$(git rev-parse ":$outer") + nested=$(git -C deps/agent-native rev-parse "HEAD:$inner") + if [ "$recorded" != "$nested" ]; then + echo "mismatch $label outer=$recorded nested=$nested" >&2 + fail=1 + else + echo "ok $label $recorded" + fi +} + +compare development-guidelines deps/development-guidelines deps/development-guidelines +compare error-code-system deps/error-code-system deps/error-code-system +compare logging-system deps/logging-system deps/logging-system + +recorded=$(git rev-parse :deps/agent-native) +actual=$(git -C deps/agent-native rev-parse HEAD) +if [ "$recorded" != "$actual" ]; then + echo "mismatch agent-native recorded=$recorded HEAD=$actual" >&2 + fail=1 +else + echo "ok agent-native $recorded" +fi + +exit "$fail" diff --git a/server/server.go b/server/server.go index c0caa70..31006f8 100644 --- a/server/server.go +++ b/server/server.go @@ -14,6 +14,13 @@ import ( "agent-squad/squad" "agent-squad/store" + + "error-code-system/errcode" + "error-code-system/httpx" + "logging-system/logsys" + + // 导入 domains 才会把种子码填进 errcode.Default。 + _ "error-code-system/domains" ) // Server HTTP 服务。 @@ -24,6 +31,15 @@ type Server struct { mu sync.Mutex latest map[string]*squad.RunResult // 最近一次推进结果（供轮询） + log *logsys.Logger +} + +// SetLogger 设置组合根日志器。nil 表示不写日志。 +func (s *Server) SetLogger(log *logsys.Logger) { + if s == nil { + return + } + s.log = log } // New 组装服务。 @@ -58,7 +74,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /v1/wakeups/sweep", s.handleSweep) mux.HandleFunc("GET /v1/runs/{id}/blackboard", s.handleBlackboard) mux.HandleFunc("GET /v1/runs/{id}/events", s.handleEvents) - return mux + return httpx.Trace(mux) } func writeJSON(w http.ResponseWriter, code int, v any) { @@ -67,8 +83,28 @@ func writeJSON(w http.ResponseWriter, code int, v any) { _ = json.NewEncoder(w).Encode(v) } -func writeErr(w http.ResponseWriter, code int, err error) { - writeJSON(w, code, map[string]string{"error": err.Error()}) +func (s *Server) writeErr(w http.ResponseWriter, r *http.Request, err error) { + if errcode.From(err) == nil { + err = errcode.Wrap(err, errcode.SysInternal, "内部错误") + } + s.logErr(r, err) + httpx.WriteError(errcode.Default, w, r, err) +} + +func (s *Server) logErr(r *http.Request, err error) { + if s == nil || s.log == nil || r == nil { + return + } + ctx := logsys.WithTrace(r.Context(), httpx.TraceID(r)) + s.log.Module("server").WithContext(ctx).Error("http error", err) +} + +func (s *Server) logRun(r *http.Request, msg string, res *squad.RunResult) { + if s == nil || s.log == nil || r == nil || res == nil { + return + } + ctx := logsys.WithTrace(r.Context(), httpx.TraceID(r)) + s.log.Module("server").WithContext(ctx).With("run_id", res.RunID).Info(msg, "status", string(res.Status)) } func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { @@ -84,14 +120,14 @@ 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 { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) return } if key := r.PathValue("key"); key != "" && spec.Role == "" { spec.Role = key } if err := s.engine.Registry().Register(spec); err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } writeJSON(w, http.StatusOK, spec) @@ -100,7 +136,7 @@ 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 !s.engine.Registry().Remove(key) { - writeErr(w, http.StatusNotFound, fmt.Errorf("role %s not found", key)) + s.writeErr(w, r, errcode.New(errcode.SquadRoleNotFound, fmt.Sprintf("role %s not found", key))) return } s.st.DeleteRole(key) @@ -110,12 +146,12 @@ 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 { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) return } spec, err := s.engine.Registry().Derive(req) if err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } writeJSON(w, http.StatusCreated, spec) @@ -135,11 +171,11 @@ 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 { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) return } if err := s.engine.Blueprints().Register(bp); err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } writeJSON(w, http.StatusCreated, bp) @@ -148,7 +184,7 @@ 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 !s.engine.Blueprints().Remove(key) { - writeErr(w, http.StatusNotFound, fmt.Errorf("blueprint %s not found", key)) + s.writeErr(w, r, errcode.New(errcode.SquadBlueprintNotFound, fmt.Sprintf("blueprint %s not found", key))) return } s.st.DeleteBlueprint(key) @@ -168,19 +204,19 @@ 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 { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) return } for name, b64 := range req.MaterialsB64 { data, err := base64.StdEncoding.DecodeString(b64) if err != nil { - writeErr(w, http.StatusBadRequest, fmt.Errorf("materials_b64[%s]: %w", name, err)) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, fmt.Sprintf("materials_b64[%s]", name))) return } req.Goal.Materials = append(req.Goal.Materials, squad.MaterialFile{Filename: name, Content: data}) } if req.Goal.Text == "" { - writeErr(w, http.StatusBadRequest, fmt.Errorf("goal.text is required")) + s.writeErr(w, r, errcode.New(errcode.SquadGoalRequired, "goal.text is required")) return } // 同步发起：Run 在首次挂起或终态前返回；长时间执行由调用方带超时/异步包装。 @@ -188,10 +224,11 @@ func (s *Server) handleCreateRun(w http.ResponseWriter, r *http.Request) { // 因此采用同步调用 + 流式等待由调用方控制超时。 res, err := s.engine.Run(r.Context(), req.Goal, squad.RunOptions{TaskKey: req.TaskKey, Backend: req.Backend}) if err != nil { - writeErr(w, http.StatusInternalServerError, err) + s.writeErr(w, r, err) return } s.record(res) + s.logRun(r, "run finished", res) code := http.StatusOK if res.Status.Terminal() { code = http.StatusOK @@ -211,7 +248,7 @@ func (s *Server) record(res *squad.RunResult) { func (s *Server) handleListRuns(w http.ResponseWriter, r *http.Request) { runs, err := s.engine.ListRuns() if err != nil { - writeErr(w, http.StatusInternalServerError, err) + s.writeErr(w, r, err) return } writeJSON(w, http.StatusOK, runs) @@ -220,7 +257,7 @@ 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")) if err != nil { - writeErr(w, http.StatusNotFound, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SquadRunNotFound, "运行不存在")) return } writeJSON(w, http.StatusOK, res) @@ -238,32 +275,34 @@ 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 { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) return } res, err := s.engine.Respond(r.Context(), r.PathValue("id"), decision) if err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } s.record(res) + s.logRun(r, "run responded", res) writeJSON(w, http.StatusOK, res) } func (s *Server) handleResume(w http.ResponseWriter, r *http.Request) { res, err := s.engine.Resume(r.Context(), r.PathValue("id")) if err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } s.record(res) + s.logRun(r, "run resumed", res) writeJSON(w, http.StatusOK, res) } func (s *Server) handleBlackboard(w http.ResponseWriter, r *http.Request) { bb, err := s.engine.Blackboard(r.PathValue("id")) if err != nil { - writeErr(w, http.StatusNotFound, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SquadRunNotFound, "运行不存在")) return } writeJSON(w, http.StatusOK, bb) @@ -274,11 +313,11 @@ func (s *Server) handleSteer(w http.ResponseWriter, r *http.Request) { Text string `json:"text"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) return } if err := s.engine.Steer(r.PathValue("id"), req.Text); err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } writeJSON(w, http.StatusAccepted, map[string]string{"status": "queued_for_next_step"}) @@ -290,11 +329,11 @@ func (s *Server) handleSetField(w http.ResponseWriter, r *http.Request) { Value string `json:"value"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, "请求 JSON 无法解析")) return } if err := s.engine.SetBlackboardField(r.PathValue("id"), req.Field, req.Value); err != nil { - writeErr(w, http.StatusBadRequest, err) + s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error())) return } writeJSON(w, http.StatusOK, map[string]string{"status": "set"}) @@ -331,7 +370,7 @@ func (s *Server) StartWakeupSweeper(interval time.Duration) (stop func()) { func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) { flusher, ok := w.(http.Flusher) if !ok { - writeErr(w, http.StatusInternalServerError, fmt.Errorf("streaming unsupported")) + s.writeErr(w, r, errcode.New(errcode.SysInternal, "streaming unsupported")) return } cursor := 0 diff --git a/server/server_test.go b/server/server_test.go index 162d852..8785192 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -9,12 +9,16 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" "agent-native/exec" "agent-squad/squad" "agent-squad/store" + + "error-code-system/httpx" + "logging-system/logsys" ) type scriptBackend struct{ script map[string]map[string]string } @@ -196,6 +200,88 @@ func collectSSEReadable(t *testing.T, resp *http.Response) string { } } +type memSink struct { + mu sync.Mutex + entries []logsys.Entry +} + +func (m *memSink) Write(e logsys.Entry) error { + m.mu.Lock() + m.entries = append(m.entries, e) + m.mu.Unlock() + return nil +} + +func (m *memSink) Sync() error { return nil } +func (m *memSink) Close() error { return nil } + +func (m *memSink) has(traceID, code string) bool { + m.mu.Lock() + defer m.mu.Unlock() + for _, e := range m.entries { + if e.TraceID == traceID && e.Code == code { + return true + } + } + return false +} + +func TestErrorEnvelopeAndTrace(t *testing.T) { + dir := t.TempDir() + st, err := store.New(dir) + if err != nil { + t.Fatal(err) + } + reg := squad.NewRoleRegistry() + bps := squad.NewBlueprintRegistry() + backends := squad.NewBackendResolver() + sink := squad.NewFanoutSink(16) + engine, err := squad.NewEngine(squad.Options{DataDir: dir, Registry: reg, Blueprints: bps, Backends: backends}) + if err != nil { + t.Fatal(err) + } + mem := &memSink{} + logger := logsys.New(logsys.Options{Level: logsys.LevelInfo, Sinks: []logsys.Sink{mem}}) + srv := New(engine, sink, st) + srv.SetLogger(logger) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + + req, err := http.NewRequest(http.MethodPost, ts.URL+"/v1/runs", strings.NewReader(`{"goal":{}}`)) + if err != nil { + t.Fatal(err) + } + const traceID = "trace-test-1234" + req.Header.Set(httpx.TraceHeader, traceID) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status %d", resp.StatusCode) + } + if got := resp.Header.Get(httpx.TraceHeader); got != traceID { + t.Fatalf("header trace %q", got) + } + var body struct { + Error struct { + Code string `json:"code"` + TraceID string `json:"trace_id"` + } `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if body.Error.Code != "squad.goal_required" || body.Error.TraceID != traceID { + t.Fatalf("envelope %+v", body.Error) + } + _ = logger.Sync() + if !mem.has(traceID, "squad.goal_required") { + t.Fatalf("log missing trace/code: %+v", mem.entries) + } +} + func TestHealthAndBlueprints(t *testing.T) { ts, _ := newTestServer(t) code, out := doJSON(t, "GET", ts.URL+"/v1/health", nil) diff --git a/squad/backend.go b/squad/backend.go index 08c5eed..0d2aa7b 100644 --- a/squad/backend.go +++ b/squad/backend.go @@ -8,6 +8,8 @@ import ( "agent-native/adapters/native" "agent-native/exec" "agent-native/runtime/llm" + + "error-code-system/errcode" ) // BackendFactory 按角色构造后端：角色级 Model/ToolProfile 在工厂闭包层绑定 @@ -136,7 +138,7 @@ func (r *BackendResolver) Resolve(role RoleSpec, defaultBackend string) (exec.Ba return nil, name, err } } else { - return nil, name, fmt.Errorf("unknown backend %q (registered: %v); refusing to silently fall back", name, names) + return nil, name, errcode.New(errcode.SquadUnknownBackend, fmt.Sprintf("unknown backend %q (registered: %v); refusing to silently fall back", name, names)) } // MaxConcurrentTasks>=1 即包装限流（1 = 对象级串行；0 = 不限）。 diff --git a/squad/progress.go b/squad/progress.go index d9b9c07..87cfdec 100644 --- a/squad/progress.go +++ b/squad/progress.go @@ -8,19 +8,19 @@ import ( // Event 类型常量（SSE 事件名与此一致）。 const ( - EvRunStarted = "run_started" - EvPhase = "phase" - EvStepStarted = "step_started" - EvStepCompleted = "step_completed" - EvStepFailed = "step_failed" - EvStepSkipped = "step_skipped" - EvDetail = "detail" // 成员执行过程帧（agent-native ProcessEvent 投影） - EvDeliverable = "deliverable" - EvHumanInputRequired = "human_input_required" - EvHumanResponded = "human_responded" - EvRunCompleted = "run_completed" - EvRunFailed = "run_failed" - EvRunCancelled = "run_cancelled" + EvRunStarted = "run_started" + EvPhase = "phase" + EvStepStarted = "step_started" + EvStepCompleted = "step_completed" + EvStepFailed = "step_failed" + EvStepSkipped = "step_skipped" + EvDetail = "detail" // 成员执行过程帧（agent-native ProcessEvent 投影） + EvDeliverable = "deliverable" + EvHumanInputRequired = "human_input_required" + EvHumanResponded = "human_responded" + EvRunCompleted = "run_completed" + EvRunFailed = "run_failed" + EvRunCancelled = "run_cancelled" ) // Event 引擎对外事件（可 JSON 序列化，SSE 直接下发）。 @@ -30,6 +30,7 @@ type Event struct { StepID string `json:"step_id,omitempty"` Role string `json:"role,omitempty"` Text string `json:"text,omitempty"` + Code string `json:"code,omitempty"` Data json.RawMessage `json:"data,omitempty"` At time.Time `json:"at"` } diff --git a/squad/workflow.go b/squad/workflow.go index a43775b..26542de 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -10,6 +10,8 @@ import ( "time" "agent-native/exec" + + "error-code-system/errcode" ) // flowDriver 工作流驱动器：推进 runState 直到终态或 HITL 挂起边界。 @@ -262,7 +264,7 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *runState, idx int) (*Ru step.DurationMs = time.Since(started).Milliseconds() if step.Status == "failed" { - d.emit(st, Event{Type: EvStepFailed, StepID: step.ID, Role: step.Role, Text: step.Error}) + d.emit(st, Event{Type: EvStepFailed, StepID: step.ID, Role: step.Role, Text: step.Error, Code: string(errcode.SquadStepFailed)}) st.setStep(idx, step) return d.failStep(ctx, st, idx) } @@ -271,7 +273,7 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *runState, idx int) (*Ru missing := missingTargets(subDir, targets) if len(missing) > 0 { step.Status, step.Error = "failed", fmt.Sprintf("missing deliverables: %v", missing) - d.emit(st, Event{Type: EvStepFailed, StepID: step.ID, Role: step.Role, Text: step.Error}) + d.emit(st, Event{Type: EvStepFailed, StepID: step.ID, Role: step.Role, Text: step.Error, Code: string(errcode.AgentMissingDeliverable)}) st.setStep(idx, step) return d.failStep(ctx, st, idx) } @@ -352,7 +354,7 @@ func (d *flowDriver) checkBlockingGaps(st *runState) *hitlGate { req := &HumanRequest{ ID: newRequestID(), RunID: st.Meta.ID, Kind: HitlGap, Severity: SeverityActionRequired, WaitCondition: waitCond, - Question: fmt.Sprintf("发现 %d 个阻塞缺口，需要人工补充信息后才能继续：", len(gaps)), + Question: fmt.Sprintf("发现 %d 个阻塞缺口，需要人工补充信息后才能继续：", len(gaps)), Options: []HumanOption{ {Value: "resolved", Label: "已在回答中补充信息，继续执行"}, {Value: "skip", Label: "跳过缺口，按现状继续"}, @@ -701,7 +703,9 @@ func firstTarget(targets []string) string { return "output.md" } -func allTargetsPresent(dir string, targets []string) bool { return len(missingTargets(dir, targets)) == 0 } +func allTargetsPresent(dir string, targets []string) bool { + return len(missingTargets(dir, targets)) == 0 +} func missingTargets(dir string, targets []string) []string { var missing []string diff --git a/store/eventsink.go b/store/eventsink.go index e49841c..ddd220d 100644 --- a/store/eventsink.go +++ b/store/eventsink.go @@ -20,9 +20,9 @@ type FileEventSink struct { DataDir string Next squad.ProgressSink - mu sync.Mutex - files map[string]*os.File - seqs map[string]int64 + mu sync.Mutex + files map[string]*os.File + seqs map[string]int64 } // NewFileEventSink 创建事件落盘 sink。 @@ -61,7 +61,7 @@ func (s *FileEventSink) Emit(ev squad.Event) { } s.seqs[ev.RunID]++ line := persistedEvent{Seq: s.seqs[ev.RunID], Type: ev.Type, StepID: ev.StepID, - Role: ev.Role, Text: Redact(ev.Text), At: ev.At} + Role: ev.Role, Text: Redact(ev.Text), Code: ev.Code, At: ev.At} if len(ev.Data) > 0 { line.Data = json.RawMessage(Redact(string(ev.Data))) } @@ -88,16 +88,17 @@ type persistedEvent struct { StepID string `json:"step_id,omitempty"` Role string `json:"role,omitempty"` Text string `json:"text,omitempty"` + Code string `json:"code,omitempty"` Data json.RawMessage `json:"data,omitempty"` At time.Time `json:"at"` } // secret 形状：API key（sk-/api key 风格长串）与 Bearer 凭证。 var ( - reSKKey = regexp.MustCompile(`(?:sk|rk|pk)-[A-Za-z0-9_\-]{12,}`) - reBearer = regexp.MustCompile(`(?i)bearer\s+[A-Za-z0-9._\-]{12,}`) - reAPIKey = regexp.MustCompile(`(?i)(api[_-]?key["':=\s]+)[A-Za-z0-9_\-\.]{12,}`) - reJWT = regexp.MustCompile(`eyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}`) + reSKKey = regexp.MustCompile(`(?:sk|rk|pk)-[A-Za-z0-9_\-]{12,}`) + reBearer = regexp.MustCompile(`(?i)bearer\s+[A-Za-z0-9._\-]{12,}`) + reAPIKey = regexp.MustCompile(`(?i)(api[_-]?key["':=\s]+)[A-Za-z0-9_\-\.]{12,}`) + reJWT = regexp.MustCompile(`eyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}`) ) // Redact 脱敏：把秘密形状子串替换为 ***（宁可多脱，不漏一个）。