weichenandCursor
5cbac00b99
Why: Squad runs were one file workspace and one JSON file the leader had to emit, so agents, issues, and retries had no durable queue. Evidence: GOWORK=off go test -count=1 -timeout 180s ./... Impact: /v1/runs stays. New routes mount only when the orchestration database opens. SSE for a run id now skips other runs. External CLIs are not a sandbox. Refs: feature/orchestration-platform S1-S7 Co-authored-by: Cursor <cursoragent@cursor.com>
427 lines
13 KiB
Go
427 lines
13 KiB
Go
// Package server 提供 agent-squad 的 HTTP API：角色/蓝图 CRUD、发起运行、
|
||
// SSE 事件流与 HITL pending/respond。无 UI；产品接入方自行适配展示层。
|
||
package server
|
||
|
||
import (
|
||
"context"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"fmt"
|
||
"net/http"
|
||
"strconv"
|
||
"sync"
|
||
"time"
|
||
|
||
"agent-squad/platform"
|
||
"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 服务。
|
||
type Server struct {
|
||
engine *squad.SquadEngine
|
||
sink *squad.FanoutSink
|
||
st *store.FileStore
|
||
|
||
mu sync.Mutex
|
||
latest map[string]*squad.RunResult // 最近一次推进结果（供轮询）
|
||
log *logsys.Logger
|
||
plat *platform.Store
|
||
}
|
||
|
||
// SetLogger 设置组合根日志器。nil 表示不写日志。
|
||
func (s *Server) SetLogger(log *logsys.Logger) {
|
||
if s == nil {
|
||
return
|
||
}
|
||
s.log = log
|
||
}
|
||
|
||
// SetPlatform attaches the orchestration database. Nil leaves the goal engine only.
|
||
func (s *Server) SetPlatform(st *platform.Store) {
|
||
if s == nil {
|
||
return
|
||
}
|
||
s.plat = st
|
||
}
|
||
|
||
// New 组装服务。
|
||
func New(engine *squad.SquadEngine, sink *squad.FanoutSink, st *store.FileStore) *Server {
|
||
return &Server{engine: engine, sink: sink, st: st, latest: map[string]*squad.RunResult{}}
|
||
}
|
||
|
||
// Handler 返回路由。
|
||
func (s *Server) Handler() http.Handler {
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("GET /v1/health", s.handleHealth)
|
||
|
||
mux.HandleFunc("GET /v1/roles", s.handleListRoles)
|
||
mux.HandleFunc("POST /v1/roles", s.handlePutRole)
|
||
mux.HandleFunc("PUT /v1/roles/{key}", s.handlePutRole)
|
||
mux.HandleFunc("DELETE /v1/roles/{key}", s.handleDeleteRole)
|
||
mux.HandleFunc("POST /v1/roles/derive", s.handleDeriveRole)
|
||
mux.HandleFunc("GET /v1/roles/recommend", s.handleRecommendRole)
|
||
|
||
mux.HandleFunc("GET /v1/blueprints", s.handleListBlueprints)
|
||
mux.HandleFunc("POST /v1/blueprints", s.handlePutBlueprint)
|
||
mux.HandleFunc("DELETE /v1/blueprints/{key}", s.handleDeleteBlueprint)
|
||
|
||
mux.HandleFunc("POST /v1/runs", s.handleCreateRun)
|
||
mux.HandleFunc("GET /v1/runs", s.handleListRuns)
|
||
mux.HandleFunc("GET /v1/runs/{id}", s.handleInspectRun)
|
||
mux.HandleFunc("GET /v1/runs/{id}/pending", s.handlePending)
|
||
mux.HandleFunc("POST /v1/runs/{id}/respond", s.handleRespond)
|
||
mux.HandleFunc("POST /v1/runs/{id}/resume", s.handleResume)
|
||
mux.HandleFunc("POST /v1/runs/{id}/steer", s.handleSteer)
|
||
mux.HandleFunc("POST /v1/runs/{id}/blackboard", s.handleSetField)
|
||
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)
|
||
if s.plat != nil {
|
||
platform.Mount(mux, s.plat)
|
||
}
|
||
return httpx.Trace(mux)
|
||
}
|
||
|
||
func writeJSON(w http.ResponseWriter, code int, v any) {
|
||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||
w.WriteHeader(code)
|
||
_ = json.NewEncoder(w).Encode(v)
|
||
}
|
||
|
||
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) {
|
||
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
|
||
}
|
||
|
||
// ---- 角色 ----
|
||
|
||
func (s *Server) handleListRoles(w http.ResponseWriter, r *http.Request) {
|
||
writeJSON(w, http.StatusOK, s.engine.Registry().List())
|
||
}
|
||
|
||
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 无法解析"))
|
||
return
|
||
}
|
||
if key := r.PathValue("key"); key != "" && spec.Role == "" {
|
||
spec.Role = key
|
||
}
|
||
if err := s.engine.Registry().Register(spec); err != nil {
|
||
s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, spec)
|
||
}
|
||
|
||
func (s *Server) handleDeleteRole(w http.ResponseWriter, r *http.Request) {
|
||
key := r.PathValue("key")
|
||
if !s.engine.Registry().Remove(key) {
|
||
s.writeErr(w, r, errcode.New(errcode.SquadRoleNotFound, fmt.Sprintf("role %s not found", key)))
|
||
return
|
||
}
|
||
s.st.DeleteRole(key)
|
||
writeJSON(w, http.StatusOK, map[string]bool{"deleted": true})
|
||
}
|
||
|
||
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 无法解析"))
|
||
return
|
||
}
|
||
spec, err := s.engine.Registry().Derive(req)
|
||
if err != nil {
|
||
s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusCreated, spec)
|
||
}
|
||
|
||
func (s *Server) handleRecommendRole(w http.ResponseWriter, r *http.Request) {
|
||
spec, ok := s.engine.Registry().Recommend(r.URL.Query().Get("q"))
|
||
writeJSON(w, http.StatusOK, map[string]any{"role": spec, "matched": ok})
|
||
}
|
||
|
||
// ---- 蓝图 ----
|
||
|
||
func (s *Server) handleListBlueprints(w http.ResponseWriter, r *http.Request) {
|
||
writeJSON(w, http.StatusOK, s.engine.Blueprints().List())
|
||
}
|
||
|
||
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 无法解析"))
|
||
return
|
||
}
|
||
if err := s.engine.Blueprints().Register(bp); err != nil {
|
||
s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusCreated, bp)
|
||
}
|
||
|
||
func (s *Server) handleDeleteBlueprint(w http.ResponseWriter, r *http.Request) {
|
||
key := r.PathValue("key")
|
||
if !s.engine.Blueprints().Remove(key) {
|
||
s.writeErr(w, r, errcode.New(errcode.SquadBlueprintNotFound, fmt.Sprintf("blueprint %s not found", key)))
|
||
return
|
||
}
|
||
s.st.DeleteBlueprint(key)
|
||
writeJSON(w, http.StatusOK, map[string]bool{"deleted": true})
|
||
}
|
||
|
||
// ---- 运行 ----
|
||
|
||
type createRunRequest struct {
|
||
Goal squad.Goal `json:"goal"`
|
||
TaskKey string `json:"task_key,omitempty"`
|
||
Backend string `json:"backend,omitempty"`
|
||
// Mode 编排模式：standard（默认）| leader（leader-delegation）。
|
||
Mode string `json:"mode,omitempty"`
|
||
// LeaderRole / MaxLeaderTurns leader 模式参数（CLI 等价项）。
|
||
LeaderRole string `json:"leader_role,omitempty"`
|
||
MaxLeaderTurns int `json:"max_leader_turns,omitempty"`
|
||
// MaterialsB64 便捷通道：filename → base64 内容。
|
||
MaterialsB64 map[string]string `json:"materials_b64,omitempty"`
|
||
}
|
||
|
||
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 无法解析"))
|
||
return
|
||
}
|
||
for name, b64 := range req.MaterialsB64 {
|
||
data, err := base64.StdEncoding.DecodeString(b64)
|
||
if err != nil {
|
||
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 == "" {
|
||
s.writeErr(w, r, errcode.New(errcode.SquadGoalRequired, "goal.text is required"))
|
||
return
|
||
}
|
||
// 同步发起：Run 在首次挂起或终态前返回；长时间执行由调用方带超时/异步包装。
|
||
// 这里为 HTTP 体验做异步：先返回 run_id 无法预知（ID 在 Run 内生成），
|
||
// 因此采用同步调用 + 流式等待由调用方控制超时。
|
||
res, err := s.engine.Run(r.Context(), req.Goal, squad.RunOptions{
|
||
TaskKey: req.TaskKey, Backend: req.Backend,
|
||
Mode: req.Mode, LeaderRole: req.LeaderRole, MaxLeaderTurns: req.MaxLeaderTurns,
|
||
})
|
||
if err != nil {
|
||
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
|
||
}
|
||
writeJSON(w, code, res)
|
||
}
|
||
|
||
func (s *Server) record(res *squad.RunResult) {
|
||
if res == nil {
|
||
return
|
||
}
|
||
s.mu.Lock()
|
||
s.latest[res.RunID] = res
|
||
s.mu.Unlock()
|
||
}
|
||
|
||
func (s *Server) handleListRuns(w http.ResponseWriter, r *http.Request) {
|
||
runs, err := s.engine.ListRuns()
|
||
if err != nil {
|
||
s.writeErr(w, r, err)
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, runs)
|
||
}
|
||
|
||
func (s *Server) handleInspectRun(w http.ResponseWriter, r *http.Request) {
|
||
res, err := s.engine.Inspect(r.PathValue("id"))
|
||
if err != nil {
|
||
s.writeErr(w, r, errcode.Wrap(err, errcode.SquadRunNotFound, "运行不存在"))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, res)
|
||
}
|
||
|
||
func (s *Server) handlePending(w http.ResponseWriter, r *http.Request) {
|
||
req, err := s.engine.Pending(r.PathValue("id"))
|
||
if err != nil || req == nil {
|
||
writeJSON(w, http.StatusOK, map[string]any{"pending": nil})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"pending": req})
|
||
}
|
||
|
||
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 无法解析"))
|
||
return
|
||
}
|
||
res, err := s.engine.Respond(r.Context(), r.PathValue("id"), decision)
|
||
if err != nil {
|
||
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 {
|
||
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 {
|
||
s.writeErr(w, r, errcode.Wrap(err, errcode.SquadRunNotFound, "运行不存在"))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, bb)
|
||
}
|
||
|
||
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 无法解析"))
|
||
return
|
||
}
|
||
if err := s.engine.Steer(r.PathValue("id"), req.Text); err != nil {
|
||
s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]string{"status": "queued_for_next_step"})
|
||
}
|
||
|
||
func (s *Server) handleSetField(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
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 无法解析"))
|
||
return
|
||
}
|
||
if err := s.engine.SetBlackboardField(r.PathValue("id"), req.Field, req.Value); err != nil {
|
||
s.writeErr(w, r, errcode.Wrap(err, errcode.SysInvalidArgument, err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]string{"status": "set"})
|
||
}
|
||
|
||
func (s *Server) handleSweep(w http.ResponseWriter, r *http.Request) {
|
||
n := s.engine.SweepWakeups(r.Context())
|
||
writeJSON(w, http.StatusOK, map[string]int{"woken": n})
|
||
}
|
||
|
||
// StartWakeupSweeper 周期扫描等待条件（等事件自动唤醒）；返回停止函数。
|
||
// 间隔默认 15s。
|
||
func (s *Server) StartWakeupSweeper(interval time.Duration) (stop func()) {
|
||
if interval <= 0 {
|
||
interval = 15 * time.Second
|
||
}
|
||
done := make(chan struct{})
|
||
go func() {
|
||
t := time.NewTicker(interval)
|
||
defer t.Stop()
|
||
for {
|
||
select {
|
||
case <-done:
|
||
return
|
||
case <-t.C:
|
||
s.engine.SweepWakeups(context.Background())
|
||
}
|
||
}
|
||
}()
|
||
return func() { close(done) }
|
||
}
|
||
|
||
// handleEvents SSE：cursor 之后重放缓冲再实时推送。
|
||
func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) {
|
||
flusher, ok := w.(http.Flusher)
|
||
if !ok {
|
||
s.writeErr(w, r, errcode.New(errcode.SysInternal, "streaming unsupported"))
|
||
return
|
||
}
|
||
cursor := 0
|
||
if c := r.URL.Query().Get("cursor"); c != "" {
|
||
if n, err := strconv.Atoi(c); err == nil {
|
||
cursor = n
|
||
}
|
||
}
|
||
ch, next, cancel := s.sink.Subscribe(cursor)
|
||
defer cancel()
|
||
w.Header().Set("Content-Type", "text/event-stream")
|
||
w.Header().Set("Cache-Control", "no-cache")
|
||
w.WriteHeader(http.StatusOK)
|
||
fmt.Fprintf(w, "event: cursor\ndata: %d\n\n", next)
|
||
flusher.Flush()
|
||
for {
|
||
select {
|
||
case <-r.Context().Done():
|
||
return
|
||
case ev, ok := <-ch:
|
||
if !ok {
|
||
return
|
||
}
|
||
if id := r.PathValue("id"); id != "" && ev.RunID != id {
|
||
continue
|
||
}
|
||
data, _ := json.Marshal(ev)
|
||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, data)
|
||
flusher.Flush()
|
||
}
|
||
}
|
||
}
|