package run import ( "context" "encoding/json" "net/http" "strconv" "time" "github.com/go-chi/chi/v5" "github.com/google/uuid" ws "github.com/coder/websocket" "github.com/coder/websocket/wsjson" "github.com/yan1h/agent-coding-workflow/internal/infra/errs" httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http" "github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware" "github.com/yan1h/agent-coding-workflow/internal/workspace" ) const defaultLogsLimit = 500 // AdminLookup resolves whether a user is an admin(mirrors acp/workspace pattern)。 type AdminLookup interface { IsAdmin(ctx context.Context, userID uuid.UUID) (bool, error) } // WSConfig controls WebSocket heartbeat and buffering for the log stream. type WSConfig struct { PingInterval time.Duration PongTimeout time.Duration SendBuffer int } // Handler 暴露 run profiles 的 HTTP/WS 端点。 type Handler struct { svc *Service resolver middleware.SessionResolver adminLookup AdminLookup cfg WSConfig } // NewHandler 构造 run.Handler。 func NewHandler(svc *Service, resolver middleware.SessionResolver, al AdminLookup, cfg WSConfig) *Handler { return &Handler{svc: svc, resolver: resolver, adminLookup: al, cfg: cfg} } // Mount 注册 run profiles 路由。分两组:workspace 作用域(列表/创建)与 // id 作用域(按 id 操作,内部经 profile→workspace 解析做 ACL)。 func (h *Handler) Mount(r chi.Router) { r.Route("/api/v1/workspaces/{wsID}/run-profiles", func(r chi.Router) { r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{})) r.Get("/", h.list) r.Post("/", h.create) }) r.Route("/api/v1/run-profiles", func(r chi.Router) { r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{})) r.Patch("/{id}", h.update) r.Delete("/{id}", h.remove) r.Post("/{id}/start", h.start) r.Post("/{id}/stop", h.stop) r.Post("/{id}/restart", h.restart) r.Get("/{id}/status", h.status) r.Get("/{id}/logs", h.logs) r.Get("/{id}/logs/stream", h.logsStream) }) } func (h *Handler) caller(r *http.Request) (workspace.Caller, error) { uid, ok := middleware.UserIDFromContext(r.Context()) if !ok { return workspace.Caller{}, errs.New(errs.CodeUnauthorized, "not authenticated") } admin, err := h.adminLookup.IsAdmin(r.Context(), uid) if err != nil { return workspace.Caller{}, err } return workspace.Caller{UserID: uid, IsAdmin: admin}, nil } func uuidParam(r *http.Request, key string) (uuid.UUID, error) { id, err := uuid.Parse(chi.URLParam(r, key)) if err != nil { return uuid.Nil, errs.New(errs.CodeInvalidInput, "bad "+key) } return id, nil } func (h *Handler) list(w http.ResponseWriter, r *http.Request) { c, err := h.caller(r) if err != nil { writeErr(w, r, err) return } wsID, err := uuidParam(r, "wsID") if err != nil { writeErr(w, r, err) return } out, err := h.svc.List(r.Context(), c, wsID) if err != nil { writeErr(w, r, err) return } httpx.WriteJSON(w, http.StatusOK, out) } func (h *Handler) create(w http.ResponseWriter, r *http.Request) { c, err := h.caller(r) if err != nil { writeErr(w, r, err) return } wsID, err := uuidParam(r, "wsID") if err != nil { writeErr(w, r, err) return } var req CreateRunProfileReq if derr := json.NewDecoder(r.Body).Decode(&req); derr != nil { writeErr(w, r, errs.New(errs.CodeInvalidInput, "bad json")) return } out, err := h.svc.Create(r.Context(), c, wsID, req) if err != nil { writeErr(w, r, err) return } httpx.WriteJSON(w, http.StatusCreated, out) } func (h *Handler) update(w http.ResponseWriter, r *http.Request) { c, err := h.caller(r) if err != nil { writeErr(w, r, err) return } id, err := uuidParam(r, "id") if err != nil { writeErr(w, r, err) return } var req UpdateRunProfileReq if derr := json.NewDecoder(r.Body).Decode(&req); derr != nil { writeErr(w, r, errs.New(errs.CodeInvalidInput, "bad json")) return } out, err := h.svc.Update(r.Context(), c, id, req) if err != nil { writeErr(w, r, err) return } httpx.WriteJSON(w, http.StatusOK, out) } func (h *Handler) remove(w http.ResponseWriter, r *http.Request) { c, err := h.caller(r) if err != nil { writeErr(w, r, err) return } id, err := uuidParam(r, "id") if err != nil { writeErr(w, r, err) return } if err := h.svc.Delete(r.Context(), c, id); err != nil { writeErr(w, r, err) return } w.WriteHeader(http.StatusNoContent) } func (h *Handler) start(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Start) } func (h *Handler) stop(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Stop) } func (h *Handler) restart(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Restart) } func (h *Handler) status(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Status) } // lifecycle 复用 start/stop/restart/status 的公共骨架(鉴权 + 解析 id + 调 svc + 写 RunStatusResp)。 func (h *Handler) lifecycle(w http.ResponseWriter, r *http.Request, fn func(context.Context, workspace.Caller, uuid.UUID) (RunStatusResp, error)) { c, err := h.caller(r) if err != nil { writeErr(w, r, err) return } id, err := uuidParam(r, "id") if err != nil { writeErr(w, r, err) return } out, err := fn(r.Context(), c, id) if err != nil { writeErr(w, r, err) return } httpx.WriteJSON(w, http.StatusOK, out) } func (h *Handler) logs(w http.ResponseWriter, r *http.Request) { c, err := h.caller(r) if err != nil { writeErr(w, r, err) return } id, err := uuidParam(r, "id") if err != nil { writeErr(w, r, err) return } since := parseInt64(r.URL.Query().Get("since"), 0) limit := int(parseInt64(r.URL.Query().Get("limit"), defaultLogsLimit)) out, err := h.svc.Logs(r.Context(), c, id, since, limit) if err != nil { writeErr(w, r, err) return } httpx.WriteJSON(w, http.StatusOK, out) } // logsStream 升级为 WebSocket,推送内存日志(含 since 续传)与实时新行。只读流, // 不接收客户端消息。鉴权经 middleware.Auth(WS 用 ?token= 传 token)。 func (h *Handler) logsStream(w http.ResponseWriter, r *http.Request) { c, err := h.caller(r) if err != nil { writeErr(w, r, err) return } id, err := uuidParam(r, "id") if err != nil { writeErr(w, r, err) return } since := parseInt64(r.URL.Query().Get("since"), 0) sub, relay, running, err := h.svc.SubscribeLogs(r.Context(), c, id, c.UserID) if err != nil { writeErr(w, r, err) return } conn, aerr := ws.Accept(w, r, &ws.AcceptOptions{}) if aerr != nil { return } defer conn.Close(ws.StatusInternalError, "internal") ctx, cancel := context.WithCancel(r.Context()) defer cancel() // 初始 state 帧。 status := StatusStopped if running { status = StatusRunning } if werr := wsjson.Write(ctx, conn, LogEnvelope{Kind: "state", Status: string(status), TS: time.Now()}); werr != nil { return } // profile 未运行:无 live relay,推完即关。 if !running || relay == nil { conn.Close(ws.StatusNormalClosure, "not running") return } defer relay.Unsubscribe(sub.ID) // 推送历史缓冲(ID > since)。 lastSent := since for _, e := range relay.since(since, 0) { if werr := wsjson.Write(ctx, conn, e); werr != nil { return } lastSent = e.ID } // 心跳。 pingTicker := time.NewTicker(h.cfg.PingInterval) defer pingTicker.Stop() go func() { for { select { case <-ctx.Done(): return case <-pingTicker.C: pingCtx, pcancel := context.WithTimeout(ctx, h.cfg.PongTimeout) perr := conn.Ping(pingCtx) pcancel() if perr != nil { cancel() return } } } }() for { select { case <-ctx.Done(): return case env, ok := <-sub.Send: if !ok { conn.Close(ws.StatusNormalClosure, "relay closed") return } // 历史已推到 lastSent,避免重复推送同 ID 的日志行(控制帧 ID=0 放行)。 if env.Kind == "log" && env.ID <= lastSent && env.ID != 0 { continue } if werr := wsjson.Write(ctx, conn, env); werr != nil { return } if env.ID > lastSent { lastSent = env.ID } } } } func parseInt64(s string, def int64) int64 { if s == "" { return def } n, err := strconv.ParseInt(s, 10, 64) if err != nil { return def } return n } func writeErr(w http.ResponseWriter, r *http.Request, err error) { httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err) }