Files

748 lines
22 KiB
Go

package main
// api.go — REST /api/* handlers, bearer auth, static web serving.
import (
"crypto/subtle"
"embed"
"encoding/json"
"errors"
"fmt"
"io/fs"
"log"
"net/http"
"os"
"path/filepath"
"regexp"
"sort"
"strconv"
"strings"
"time"
)
// Route names, event query defaults and limits.
const (
envToken string = "LVMH_TOKEN"
envSettingsFile string = "LVMH_SETTINGS_FILE"
defaultSettingsFile string = "/app/build/docker/pi-agent/settings.json"
defaultBakedModelsFile string = "/app/build/docker/pi-agent/models.json"
envModelsFile string = "LVMH_MODELS_FILE"
defaultModelsFile string = "/app/build/docker/worker-models.json"
defaultEventsAfter int64 = 0
defaultEventsLimit int = 1000
maxEventsLimit int = 10000
maxBodyBytes int64 = 1 << 20
webIndexFallback string = "index.html"
reposPathPrefix string = "/api/repos/"
reposPrepareSuffix string = "/prepare"
prepOpsBootWait time.Duration = 90 * time.Second
// opsPreparePrompt is sent to the ops control session by
// POST /api/repos/<repo>/prepare.
opsPreparePrompt string = "prepare %s: clone, build a worker image, register it"
reposPathSuffix string = "/image"
)
// imageNameRe pins registered images to the lvmh-worker- namespace so agents
// can only point spawns at images we built.
var imageNameRe = regexp.MustCompile(`^lvmh-worker-[a-z0-9][a-z0-9.-]*:?[a-zA-Z0-9._-]*$`)
//go:embed webdist
var embeddedWeb embed.FS
// Server wires store, hub, spawner and gitlab client into HTTP handlers.
type Server struct {
store *Store
hub *Hub
spawn *Spawner
gitlab *GitLab
}
// NewServer builds the Server and cross-links hub callbacks.
func NewServer(store *Store, hub *Hub, spawn *Spawner, gl *GitLab) *Server {
s := &Server{store: store, hub: hub, spawn: spawn, gitlab: gl}
hub.SpawnStatus = spawn.JobsSnapshot
return s
}
// Routes assembles the full mux with auth middleware on /api and /agent.
func (s *Server) Routes(webdist string) http.Handler {
mux := http.NewServeMux()
mux.Handle("GET /agent/ws", s.bearerAuth(http.HandlerFunc(s.hub.ServeAgentWS)))
mux.HandleFunc("GET /ws", s.handleWebWS)
api := http.NewServeMux()
api.HandleFunc("GET /api/sessions", s.handleSessions)
api.HandleFunc("GET /api/sessions/{id}/events", s.handleSessionEvents)
api.HandleFunc("GET /api/sessions/{id}/stats", s.handleSessionStats)
api.HandleFunc("GET /api/stats", s.handleStats)
api.HandleFunc("POST /api/sessions/{id}/prompt", s.handlePrompt)
api.HandleFunc("POST /api/sessions/{id}/abort", s.handleAbort)
api.HandleFunc("POST /api/sessions/{id}/model", s.handleSetModel)
api.HandleFunc("PATCH /api/sessions/{id}", s.handleRenameSession)
api.HandleFunc("DELETE /api/sessions/{id}/container", s.handleDeleteContainer)
api.HandleFunc("GET /api/model-catalog", s.handleModelCatalog)
api.HandleFunc("POST /api/spawn", s.handleSpawn)
api.HandleFunc("GET /api/spawn/status", s.handleSpawnStatus)
api.HandleFunc("GET /api/repos", s.handleRepoImages)
api.HandleFunc("POST /api/repos/", s.handlePrepareRepo)
api.HandleFunc("POST /api/pi-config/resync", s.handlePiConfigResync)
// repo paths contain "/" (group/project), which a single {repo} wildcard
// segment cannot match — subtree routes with manual path parsing instead.
api.HandleFunc("PUT /api/repos/", s.handleSetRepoImage)
api.HandleFunc("DELETE /api/repos/", s.handleDeleteRepoImage)
api.HandleFunc("GET /api/gitlab/status", s.handleGitLabStatus)
api.HandleFunc("POST /api/gitlab/connect", s.handleGitLabConnect)
api.HandleFunc("DELETE /api/gitlab/connect", s.handleGitLabDisconnect)
api.HandleFunc("GET /api/gitlab/repos", s.handleGitLabRepos)
mux.Handle("/api/", s.bearerAuth(api))
mux.Handle("/", s.webHandler(webdist))
return recoverMiddleware(mux)
}
// recoverMiddleware converts panics into 500s instead of killing the server.
func recoverMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
defer func() {
if rec := recover(); rec != nil {
writeError(w, http.StatusInternalServerError, fmt.Sprintf("internal error: %v", rec))
}
}()
next.ServeHTTP(w, r)
})
}
// bearerAuth rejects non-matching bearer tokens with 401.
func (s *Server) bearerAuth(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if !tokenMatches(r.Header.Get("Authorization")) {
writeError(w, http.StatusUnauthorized, "unauthorized")
return
}
next.ServeHTTP(w, r)
})
}
// TokenMatches validates an Authorization header value against LVMH_TOKEN.
func tokenMatches(header string) bool {
const prefix = "Bearer "
if !strings.HasPrefix(header, prefix) {
return false
}
token := strings.TrimSpace(strings.TrimPrefix(header, prefix))
want := strings.TrimSpace(daemonToken)
if want == "" || token == "" {
return false
}
return subtle.ConstantTimeCompare([]byte(token), []byte(want)) == 1
}
// handleWebWS upgrades /ws with token from query param (?token=).
func (s *Server) handleWebWS(w http.ResponseWriter, r *http.Request) {
token := r.URL.Query().Get("token")
want := strings.TrimSpace(daemonToken)
if want == "" || token == "" ||
subtle.ConstantTimeCompare([]byte(token), []byte(want)) != 1 {
// Reject before upgrade: browsers can't read WS close codes well.
writeError(w, http.StatusUnauthorized, "unauthorized")
return
}
s.hub.ServeWebWS(w, r)
}
func writeError(w http.ResponseWriter, status int, msg string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(map[string]string{"error": msg})
}
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(v)
}
func decodeBody(w http.ResponseWriter, r *http.Request, dst any) bool {
r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
dec := json.NewDecoder(r.Body)
if err := dec.Decode(dst); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON body: "+err.Error())
return false
}
return true
}
func (s *Server) handleSessions(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.hub.SessionsView())
}
func (s *Server) handleSessionEvents(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
after := defaultEventsAfter
if raw := r.URL.Query().Get("after"); raw != "" {
v, err := strconv.ParseInt(raw, 10, 64)
if err != nil || v < 0 {
writeError(w, http.StatusBadRequest, "invalid after")
return
}
after = v
}
limit := defaultEventsLimit
if raw := r.URL.Query().Get("limit"); raw != "" {
v, err := strconv.Atoi(raw)
if err != nil || v <= 0 {
writeError(w, http.StatusBadRequest, "invalid limit")
return
}
limit = v
}
if limit > maxEventsLimit {
limit = maxEventsLimit
}
var (
events []Event
err error
)
switch {
case r.URL.Query().Get("latest") != "":
if r.URL.Query().Get("latest") != "1" {
writeError(w, http.StatusBadRequest, "invalid latest")
return
}
events, err = s.store.EventsLatest(id, limit)
case r.URL.Query().Get("before") != "":
before, perr := strconv.ParseInt(r.URL.Query().Get("before"), 10, 64)
if perr != nil || before < 0 {
writeError(w, http.StatusBadRequest, "invalid before")
return
}
events, err = s.store.EventsBefore(id, before, limit)
default:
events, err = s.store.EventsAfter(id, after, limit)
}
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if events == nil {
events = []Event{}
}
out := make([]map[string]any, 0, len(events))
for _, e := range events {
var payload map[string]any
if len(e.Payload) > 0 {
if err := json.Unmarshal(e.Payload, &payload); err != nil {
payload = map[string]any{}
}
}
frameMap := map[string]any{}
for k, v := range payload {
frameMap[k] = v
}
frameMap["v"] = daemonVersion
frameMap["sessionId"] = e.SessionID
frameMap["seq"] = e.Seq
frameMap["ts"] = e.TS
frameMap["type"] = e.Type
out = append(out, frameMap)
}
writeJSON(w, http.StatusOK, out)
}
func (s *Server) handlePrompt(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var body struct {
Message string `json:"message"`
}
if !decodeBody(w, r, &body) {
return
}
if strings.TrimSpace(body.Message) == "" {
writeError(w, http.StatusBadRequest, "message required")
return
}
if err := s.hub.Prompt(id, body.Message); err != nil {
if errors.Is(err, ErrOffline) {
writeError(w, http.StatusConflict, "session offline")
return
}
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleAbort(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
if err := s.hub.Abort(id); err != nil {
if errors.Is(err, ErrOffline) {
writeError(w, http.StatusConflict, "session offline")
return
}
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleSetModel(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var body struct {
Provider string `json:"provider"`
ModelID string `json:"modelId"`
}
if !decodeBody(w, r, &body) {
return
}
if strings.TrimSpace(body.ModelID) == "" || strings.TrimSpace(body.Provider) == "" {
writeError(w, http.StatusBadRequest, "provider and modelId required")
return
}
if err := s.hub.SetModel(id, body.Provider, body.ModelID); err != nil {
if errors.Is(err, ErrOffline) {
writeError(w, http.StatusConflict, "session offline")
return
}
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleRenameSession(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var body struct {
Name string `json:"name"`
}
if !decodeBody(w, r, &body) {
return
}
if strings.TrimSpace(body.Name) == "" {
writeError(w, http.StatusBadRequest, "name required")
return
}
found, err := s.store.SetSessionName(id, body.Name)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if !found {
writeError(w, http.StatusNotFound, "unknown session")
return
}
// best-effort mirror into the live session; offline renames persist
// daemon-side only (documented edge in PROTOCOL.md)
if err := s.hub.Rename(id, body.Name); err != nil && !errors.Is(err, ErrOffline) {
log.Printf("api: rename frame %s: %v", id, err)
}
s.hub.BroadcastSessionList()
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleDeleteContainer(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var err error
if id == opsSessionID {
// the ops container has no containers-table row; remove by name
err = s.spawn.RemoveOps(r.Context())
} else {
err = s.spawn.RemoveSession(r.Context(), id)
}
if err != nil {
if errors.Is(err, errNoContainer) {
writeError(w, http.StatusNotFound, "no container for session")
return
}
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleSpawn(w http.ResponseWriter, r *http.Request) {
var body struct {
Repo string `json:"repo"`
Branch string `json:"branch"`
Model string `json:"model"`
Empty bool `json:"empty"`
}
if !decodeBody(w, r, &body) {
return
}
// Blank-container spawns carry no repo; everything else must look like
// group/project.
if body.Repo == "" && !body.Empty {
writeError(w, http.StatusBadRequest, "repo required (or set empty for a blank container)")
return
}
if body.Repo != "" && !validRepoPath(body.Repo) {
writeError(w, http.StatusBadRequest, "repo must look like group/project")
return
}
if body.Branch != "" && !branchRe.MatchString(body.Branch) {
writeError(w, http.StatusBadRequest, "invalid branch name")
return
}
// Optional initial model "provider/model-id" (validated loosely here; the
// bridge logs and falls back when the registry cannot resolve it).
if body.Model != "" && !modelSpecRe.MatchString(body.Model) {
writeError(w, http.StatusBadRequest, "model must look like provider/model-id")
return
}
res, err := s.spawn.Start(r.Context(), body.Repo, body.Branch, body.Model, body.Empty)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusCreated, res)
}
func (s *Server) handleSpawnStatus(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.spawn.JobsSnapshot())
}
// repoFromImagePath extracts the repo path from /api/repos/<repo>/image.
func repoFromImagePath(path string) (string, bool) {
rest, ok := strings.CutSuffix(strings.TrimPrefix(path, reposPathPrefix), reposPathSuffix)
if !ok || !validRepoPath(rest) {
return "", false
}
return rest, true
}
func (s *Server) handleRepoImages(w http.ResponseWriter, r *http.Request) {
rows, err := s.store.ListRepoImages()
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if rows == nil {
rows = []RepoImageRow{}
}
for i := range rows {
built, err := s.spawn.imageRefExists(r.Context(), rows[i].Image)
if err != nil {
writeError(w, http.StatusInternalServerError, "docker: "+err.Error())
return
}
rows[i].Built = built
}
writeJSON(w, http.StatusOK, rows)
}
func (s *Server) handleSetRepoImage(w http.ResponseWriter, r *http.Request) {
repo, ok := repoFromImagePath(r.URL.Path)
if !ok {
writeError(w, http.StatusBadRequest, "repo must look like group/project")
return
}
var body struct {
Image string `json:"image"`
}
if !decodeBody(w, r, &body) {
return
}
if !imageNameRe.MatchString(body.Image) {
writeError(w, http.StatusBadRequest, "invalid image name")
return
}
if err := s.store.SetRepoImage(repo, body.Image); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleDeleteRepoImage(w http.ResponseWriter, r *http.Request) {
repo, ok := repoFromImagePath(r.URL.Path)
if !ok {
writeError(w, http.StatusBadRequest, "repo must look like group/project")
return
}
if err := s.store.DeleteRepoImage(repo); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleGitLabStatus(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.gitlab.Status())
}
func (s *Server) handleGitLabConnect(w http.ResponseWriter, r *http.Request) {
var body struct {
Token string `json:"token"`
}
if !decodeBody(w, r, &body) {
return
}
if strings.TrimSpace(body.Token) == "" {
writeError(w, http.StatusBadRequest, "token required")
return
}
username, err := s.gitlab.Connect(r.Context(), body.Token)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]string{"username": username})
}
func (s *Server) handleGitLabDisconnect(w http.ResponseWriter, r *http.Request) {
if err := s.gitlab.Disconnect(); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleGitLabRepos(w http.ResponseWriter, r *http.Request) {
repos, err := s.gitlab.Repos(r.Context())
if err != nil {
status := http.StatusInternalServerError
var httpErr *GitLabError
if errors.As(err, &httpErr) {
status = http.StatusBadGateway
}
if errors.Is(err, errNotConnected) {
status = http.StatusConflict
}
writeError(w, status, err.Error())
return
}
if repos == nil {
repos = []GitLabRepo{}
}
writeJSON(w, http.StatusOK, repos)
}
// ModelCatalogItem is one selectable model in GET /api/model-catalog.
type ModelCatalogItem struct {
Provider string `json:"provider"`
ID string `json:"id"`
Name string `json:"name"`
}
// parseModelCatalog extracts provider/id/name entries from a pi models.json
// file body, grouped per provider (file order preserved within a provider).
func parseModelCatalog(data []byte) []ModelCatalogItem {
var doc struct {
Providers map[string]struct {
Models []struct {
ID string `json:"id"`
Name string `json:"name"`
} `json:"models"`
} `json:"providers"`
}
if err := json.Unmarshal(data, &doc); err != nil {
return nil
}
providers := make([]string, 0, len(doc.Providers))
for p := range doc.Providers {
providers = append(providers, p)
}
sort.Strings(providers)
out := []ModelCatalogItem{}
for _, p := range providers {
for _, m := range doc.Providers[p].Models {
if m.ID == "" {
continue
}
name := m.Name
if name == "" {
name = m.ID
}
out = append(out, ModelCatalogItem{Provider: p, ID: m.ID, Name: name})
}
}
return out
}
// parseEnabledModels turns settings.json "enabledModels" entries
// ("provider/model-id") into catalog items, so built-in providers
// (anthropic, openai, google, ...) show up next to custom ones.
func parseEnabledModels(data []byte) []ModelCatalogItem {
var doc struct {
EnabledModels []string `json:"enabledModels"`
}
if err := json.Unmarshal(data, &doc); err != nil {
return nil
}
out := []ModelCatalogItem{}
for _, entry := range doc.EnabledModels {
provider, id, ok := strings.Cut(entry, "/")
if !ok || provider == "" || id == "" {
continue
}
out = append(out, ModelCatalogItem{Provider: provider, ID: id, Name: id})
}
return out
}
func (s *Server) handleModelCatalog(w http.ResponseWriter, r *http.Request) {
// Prefer the baked dotfiles models.json (full provider set, synced by
// rsync-pi-agent.sh); fall back to the minimal frozen worker list.
path := envOr(envModelsFile, "")
if path == "" {
path = defaultBakedModelsFile
if t := os.Getenv("LVMH_TEST_BAKED_MODELS"); t != "" {
path = t
} else if _, err := os.Stat(path); err != nil {
path = defaultModelsFile
}
}
data, err := os.ReadFile(path)
if err != nil {
writeError(w, http.StatusInternalServerError, "model catalog unavailable: "+err.Error())
return
}
catalog := parseModelCatalog(data)
if catalog == nil {
writeError(w, http.StatusInternalServerError, "model catalog unreadable")
return
}
// Merge in enabled built-in models from the baked pi settings; custom
// providers (models.json) win on duplicates.
if sdata, serr := os.ReadFile(envOr(envSettingsFile, defaultSettingsFile)); serr == nil {
seen := make(map[string]bool, len(catalog))
for _, m := range catalog {
seen[m.Provider+"/"+m.ID] = true
}
for _, m := range parseEnabledModels(sdata) {
if !seen[m.Provider+"/"+m.ID] {
catalog = append(catalog, m)
}
}
}
writeJSON(w, http.StatusOK, catalog)
}
// webHandler serves the embedded web/dist, or an override directory when
// webdist is non-empty (dev mode / mounted volume).
func (s *Server) webHandler(webdist string) http.Handler {
if webdist != "" {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Guard against path traversal outside the override dir.
cleaned := filepath.Clean("/" + r.URL.Path)
full := filepath.Join(webdist, cleaned)
if !strings.HasPrefix(full, filepath.Clean(webdist)) {
http.NotFound(w, r)
return
}
st, err := os.Stat(full)
if err == nil && !st.IsDir() {
http.ServeFile(w, r, full)
return
}
http.ServeFile(w, r, filepath.Join(webdist, webIndexFallback))
})
}
sub, err := fs.Sub(embeddedWeb, "webdist")
if err != nil {
panic(err)
}
fileServer := http.FileServer(http.FS(sub))
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
path := strings.TrimPrefix(r.URL.Path, "/")
if path == "" {
path = webIndexFallback
}
if _, err := fs.Stat(sub, path); err != nil {
r.URL.Path = "/" + webIndexFallback // SPA fallback
}
fileServer.ServeHTTP(w, r)
})
}
// daemonToken is set by main from LVMH_TOKEN (single process, read-only).
var daemonToken string
// repoFromPreparePath extracts the repo path from /api/repos/<repo>/prepare.
func repoFromPreparePath(path string) (string, bool) {
return repoFromSubPath(path, reposPrepareSuffix)
}
func (s *Server) handlePrepareRepo(w http.ResponseWriter, r *http.Request) {
repo, ok := repoFromPreparePath(r.URL.Path)
if !ok {
writeError(w, http.StatusBadRequest, "repo must look like group/project")
return
}
// Fresh ops context per prepare: when ops is running, restart it (wiping
// its transcript) so the prompt lands in a clean session — no history
// buildup across prepares. When offline, try to boot it once.
if s.hub.IsOnline(opsSessionID) {
if err := s.spawn.RemoveOps(r.Context()); err != nil {
writeError(w, http.StatusInternalServerError, "reset ops: "+err.Error())
return
}
_, _ = s.store.DeleteSessionEvents(opsSessionID)
_, _ = s.store.SetSessionName(opsSessionID, "")
}
sent := false
wait := prepOpsBootWait
if w := os.Getenv("LVMH_TEST_OPS_BOOT_WAIT"); w != "" {
if d, err := time.ParseDuration(w); err == nil {
wait = d
}
}
deadline := time.Now().Add(wait)
for time.Now().Before(deadline) {
if s.hub.IsOnline(opsSessionID) ||
(s.spawn.EnsureOps(r.Context()) == nil && s.hub.IsOnline(opsSessionID)) {
if s.hub.Prompt(opsSessionID, fmt.Sprintf(opsPreparePrompt, repo)) == nil {
sent = true
}
break
}
select {
case <-r.Context().Done():
writeError(w, http.StatusRequestTimeout, "client gone while booting ops")
return
case <-time.After(2 * time.Second):
}
}
if !sent {
writeError(w, http.StatusConflict, "ops session did not come back online")
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func repoFromSubPath(path, suffix string) (string, bool) {
rest, ok := strings.CutSuffix(strings.TrimPrefix(path, reposPathPrefix), suffix)
if !ok || !validRepoPath(rest) {
return "", false
}
return rest, true
}
func (s *Server) handleSessionStats(w http.ResponseWriter, r *http.Request) {
stats, err := s.store.SessionStats(r.PathValue("id"))
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, stats)
}
func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) {
totals, err := s.store.StatsTotals()
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, statsResponse{StatsTotals: totals, OnlineCount: s.hub.OnlineCount()})
}
// statsResponse is the /api/stats shape: daemon-wide totals plus hub liveness.
type statsResponse struct {
StatsTotals
OnlineCount int `json:"onlineCount"`
}