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" ) // 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" // opsPreparePrompt is sent to the ops control session by // POST /api/repos//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"` } if !decodeBody(w, r, &body) { return } if !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 } res, err := s.spawn.Start(r.Context(), body.Repo, body.Branch) 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//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 _, 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//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 } if err := s.hub.Prompt(opsSessionID, fmt.Sprintf(opsPreparePrompt, repo)); err != nil { if errors.Is(err, ErrOffline) { writeError(w, http.StatusConflict, "ops session offline") return } writeError(w, http.StatusInternalServerError, err.Error()) 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"` }