From 698b742cb3429b6613b0fead1c8b54ec34515542 Mon Sep 17 00:00:00 2001 From: Raphael Westphal Date: Wed, 19 Aug 2026 10:07:52 +0200 Subject: [PATCH] =?UTF-8?q?daemon:=20integrate=20QoL=20slices=201-5=20(mod?= =?UTF-8?q?el/rename/catalog,=20history=20deletes,=20stats,=20repo=20prepa?= =?UTF-8?q?re+imageUsed+built)=20=E2=80=94=2095.3%=20cover=20green?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- daemon/api.go | 191 ++++++++++++++++++++++++++++- daemon/api_extra_test.go | 16 +-- daemon/api_prepare_test.go | 168 +++++++++++++++++++++++++ daemon/api_stats_test.go | 103 ++++++++++++++++ daemon/docker.go | 50 +++++--- daemon/docker_test.go | 6 + daemon/hub.go | 33 +++++ daemon/model_rename_test.go | 238 ++++++++++++++++++++++++++++++++++++ daemon/spawner_test.go | 6 +- daemon/store.go | 121 ++++++++++++++++++ daemon/store_extra_test.go | 82 +++++++++++++ daemon/store_stats_test.go | 85 +++++++++++++ plugin/lvmh-agent.ts | 92 ++++++++++++++ plugin/smoke.ts | 111 +++++++++++++++++ web/.pi-merge-task.md | 46 +++++++ web/src/ChatView.tsx | 194 ++++++++++++++++++++++++++++- 16 files changed, 1506 insertions(+), 36 deletions(-) create mode 100644 daemon/api_prepare_test.go create mode 100644 daemon/api_stats_test.go create mode 100644 daemon/model_rename_test.go create mode 100644 daemon/store_stats_test.go create mode 100644 web/.pi-merge-task.md diff --git a/daemon/api.go b/daemon/api.go index d5f9111..bd38e8b 100644 --- a/daemon/api.go +++ b/daemon/api.go @@ -9,10 +9,12 @@ import ( "errors" "fmt" "io/fs" + "log" "net/http" "os" "path/filepath" "regexp" + "sort" "strconv" "strings" ) @@ -21,14 +23,21 @@ import ( const ( envToken string = "LVMH_TOKEN" + 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/" - reposPathSuffix string = "/image" + 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 @@ -63,12 +72,18 @@ func (s *Server) Routes(webdist string) http.Handler { 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) // 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) @@ -268,6 +283,60 @@ func (s *Server) handleAbort(w http.ResponseWriter, r *http.Request) { 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 @@ -334,6 +403,14 @@ func (s *Server) handleRepoImages(w http.ResponseWriter, r *http.Request) { 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) } @@ -424,6 +501,62 @@ func (s *Server) handleGitLabRepos(w http.ResponseWriter, r *http.Request) { 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 +} + +func (s *Server) handleModelCatalog(w http.ResponseWriter, r *http.Request) { + data, err := os.ReadFile(envOr(envModelsFile, defaultModelsFile)) + 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 + } + 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 { @@ -463,3 +596,57 @@ func (s *Server) webHandler(webdist string) http.Handler { // 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"` +} diff --git a/daemon/api_extra_test.go b/daemon/api_extra_test.go index 0140c65..7214ba8 100644 --- a/daemon/api_extra_test.go +++ b/daemon/api_extra_test.go @@ -83,7 +83,7 @@ func TestAPISpawnStatusAndContainerLifecycle(t *testing.T) { if err := json.Unmarshal([]byte(body), &res); err != nil { t.Fatalf("decode spawn result: %v", err) } - if res.SessionID == "" || res.ContainerID != "" { + if res.SessionID == "" || res.ImageUsed != imageRefWorker { t.Fatalf("spawn result = %+v", res) } @@ -428,7 +428,7 @@ func TestAPIWebHandlerEmbedded(t *testing.T) { } func TestAPIRepoImagesCRUD(t *testing.T) { - ts, store := newTestServer(t) + ts, _ := newSpawnAPIServer(t) auth := testToken // auth required on the new namespace @@ -450,21 +450,23 @@ func TestAPIRepoImagesCRUD(t *testing.T) { if code != http.StatusOK || !strings.Contains(body, `"ok":true`) { t.Fatalf("put image = %d %s", code, body) } - // a second repo, then listing is sorted by repo + // a second repo, then listing is sorted by repo; built reflects the fake + // docker store (legacy mode: only the default worker image exists) if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/alpha/repo/image", auth, `{"image":"lvmh-worker-alpha:1.2.3"}`); code != http.StatusOK { t.Fatalf("put tagged image = %d", code) } code, body = apiReq(t, http.MethodGet, ts.URL+"/api/repos", auth, "") - if code != http.StatusOK || body != `[{"repo":"alpha/repo","image":"lvmh-worker-alpha:1.2.3"},{"repo":"group/project","image":"lvmh-worker-group--project-ab12cd"}]`+"\n" { + if code != http.StatusOK || body != `[{"repo":"alpha/repo","image":"lvmh-worker-alpha:1.2.3","built":false},{"repo":"group/project","image":"lvmh-worker-group--project-ab12cd","built":false}]`+"\n" { t.Fatalf("repos = %d %s", code, body) } - // upsert via PUT + // upsert via PUT, verified through the listing if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/group/project/image", auth, `{"image":"lvmh-worker-other"}`); code != http.StatusOK { t.Fatalf("upsert = %d", code) } - if img, ok, _ := store.GetRepoImage("group/project"); !ok || img != "lvmh-worker-other" { - t.Fatalf("after upsert = %q %v", img, ok) + code, body = apiReq(t, http.MethodGet, ts.URL+"/api/repos", auth, "") + if code != http.StatusOK || !strings.Contains(body, `{"repo":"group/project","image":"lvmh-worker-other","built":false}`) { + t.Fatalf("after upsert = %d %s", code, body) } // delete → gone; idempotent delete diff --git a/daemon/api_prepare_test.go b/daemon/api_prepare_test.go new file mode 100644 index 0000000..920dc0c --- /dev/null +++ b/daemon/api_prepare_test.go @@ -0,0 +1,168 @@ +package main + +// api_prepare_test.go — POST /api/repos/{repo}/prepare (ops prompt routing), +// GET /api/repos built flags, POST /api/spawn imageUsed resolution. + +import ( + "encoding/json" + "net/http" + "strings" + "testing" + "time" +) + +// TestAPIPrepareRepoRoutesToOps: with a live ops agent WS the prepare route +// delivers the prompt; without it the route 409s. +func TestAPIPrepareRepoRoutesToOps(t *testing.T) { + ts, _ := newTestServer(t) + auth := testToken + prepareURL := ts.URL + "/api/repos/g/p/prepare" + + // ops offline → 409 + if code, body := apiReq(t, http.MethodPost, prepareURL, auth, ""); code != http.StatusConflict { + t.Fatalf("prepare offline = %d %s, want 409", code, body) + } + + // path validation + for _, path := range []string{"/api/repos/noslash/prepare", "/api/repos/g/p/notprepare", "/api/repos/g/p/image"} { + if code, _ := apiReq(t, http.MethodPost, ts.URL+path, auth, ""); code != http.StatusBadRequest { + t.Fatalf("POST %s = %d, want 400", path, code) + } + } + if code, _ := apiReq(t, http.MethodPost, prepareURL, "", ""); code != http.StatusUnauthorized { + t.Fatal("prepare must require auth") + } + + // %2F-encoded repo path (web encodeURIComponent) decodes to the same route + ws := dialAgent(t, ts) + if err := ws.WriteJSON(helloFrame(opsSessionID)); err != nil { + t.Fatalf("hello: %v", err) + } + if welcome := readFrame(t, ws); welcome["type"] != evWelcome { + t.Fatalf("welcome = %v", welcome) + } + code, body := apiReq(t, http.MethodPost, ts.URL+"/api/repos/g%2Fp/prepare", auth, "") + if code != http.StatusOK || !strings.Contains(body, `"ok":true`) { + t.Fatalf("prepare = %d %s, want 200 ok", code, body) + } + prompt := readFrame(t, ws) + if prompt["type"] != evPrompt { + t.Fatalf("prompt frame = %v", prompt) + } + if prompt["sessionId"] != opsSessionID { + t.Fatalf("prompt sessionId = %v, want ops session", prompt["sessionId"]) + } + want := "prepare g/p: clone, build a worker image, register it" + if prompt["message"] != want { + t.Fatalf("prompt message = %q, want %q", prompt["message"], want) + } +} + +// TestAPIRepoImagesBuiltFlag: each row carries built=true when the image +// exists on the docker host, false when registered but missing. +func TestAPIRepoImagesBuiltFlag(t *testing.T) { + ts, f := newSpawnAPIServer(t) + auth := testToken + custom := "lvmh-worker-group--project-ab12cd" + f.imageTags = map[string]bool{imageRefWorker: true, custom: true} + if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/group/project/image", auth, `{"image":"`+custom+`"}`); code != http.StatusOK { + t.Fatal("register custom image failed") + } + if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/other/repo/image", auth, `{"image":"lvmh-worker-other"}`); code != http.StatusOK { + t.Fatal("register second image failed") + } + + code, body := apiReq(t, http.MethodGet, ts.URL+"/api/repos", auth, "") + if code != http.StatusOK { + t.Fatalf("repos = %d %s", code, body) + } + var rows []RepoImageRow + if err := json.Unmarshal([]byte(body), &rows); err != nil { + t.Fatalf("decode: %v", err) + } + if len(rows) != 2 { + t.Fatalf("rows = %+v", rows) + } + for _, r := range rows { + wantBuilt := r.Image == custom + if r.Built != wantBuilt { + t.Fatalf("row %+v built = %v, want %v", r, r.Built, wantBuilt) + } + } + + // docker unreachable → 500, not a silent built=false + f.failImages = true + if code, body := apiReq(t, http.MethodGet, ts.URL+"/api/repos", auth, ""); code != http.StatusInternalServerError { + t.Fatalf("repos with dead docker = %d %s, want 500", code, body) + } +} + +// TestAPISpawnImageUsed: the spawn response names the image the job will use +// (registry lookup), default or custom; the running job line carries it too. +func TestAPISpawnImageUsed(t *testing.T) { + ts, f := newSpawnAPIServer(t) + auth := testToken + custom := "lvmh-worker-group--project-ab12cd" + f.imageTags = map[string]bool{imageRefWorker: true, custom: true} + + // unregistered repo → default worker image + code, body := apiReq(t, http.MethodPost, ts.URL+"/api/spawn", auth, `{"repo":"plain/repo"}`) + if code != http.StatusCreated { + t.Fatalf("default spawn = %d %s", code, body) + } + var res SpawnResult + if err := json.Unmarshal([]byte(body), &res); err != nil { + t.Fatalf("decode: %v", err) + } + if res.ImageUsed != imageRefWorker { + t.Fatalf("default imageUsed = %q, want %q", res.ImageUsed, imageRefWorker) + } + + // registered + built custom image → custom in response and job line + if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/group/project/image", auth, `{"image":"`+custom+`"}`); code != http.StatusOK { + t.Fatal("register custom failed") + } + code, body = apiReq(t, http.MethodPost, ts.URL+"/api/spawn", auth, `{"repo":"group/project"}`) + if code != http.StatusCreated { + t.Fatalf("custom spawn = %d %s", code, body) + } + if err := json.Unmarshal([]byte(body), &res); err != nil { + t.Fatalf("decode: %v", err) + } + if res.ImageUsed != custom { + t.Fatalf("custom imageUsed = %q, want %q", res.ImageUsed, custom) + } + + // running job snapshot reports the image + var runningJob SpawnJob + waitFor(t, 5*time.Second, func() bool { + _, body := apiReq(t, http.MethodGet, ts.URL+"/api/spawn/status", auth, "") + var jobs []SpawnJob + _ = json.Unmarshal([]byte(body), &jobs) + for _, j := range jobs { + if j.SessionID == res.SessionID && j.State == stateRunning { + runningJob = j + return true + } + } + return false + }) + if runningJob.Message != "image "+custom { + t.Fatalf("running message = %q, want %q", runningJob.Message, "image "+custom) + } + + // registered but NOT built: response still names the custom image (the + // job itself errors later with the ask-ops message) + if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/other/repo/image", auth, `{"image":"lvmh-worker-unbuilt"}`); code != http.StatusOK { + t.Fatal("register unbuilt failed") + } + code, body = apiReq(t, http.MethodPost, ts.URL+"/api/spawn", auth, `{"repo":"other/repo"}`) + if code != http.StatusCreated { + t.Fatalf("unbuilt spawn = %d %s", code, body) + } + var res2 SpawnResult + _ = json.Unmarshal([]byte(body), &res2) + if res2.ImageUsed != "lvmh-worker-unbuilt" { + t.Fatalf("unbuilt imageUsed = %q, want lvmh-worker-unbuilt", res2.ImageUsed) + } +} diff --git a/daemon/api_stats_test.go b/daemon/api_stats_test.go new file mode 100644 index 0000000..d6d1b50 --- /dev/null +++ b/daemon/api_stats_test.go @@ -0,0 +1,103 @@ +package main + +// api_stats_test.go — REST shape of /api/sessions/:id/stats and /api/stats. + +import ( + "encoding/json" + "net/http" + "testing" +) + +func TestAPISessionStatsShape(t *testing.T) { + ts, store := newTestServer(t) + appendAgentEnd(t, store, "s1", 1, `{"usage":{"inputTokens":1200,"outputTokens":340,"totalCost":0.25}}`) + appendAgentEnd(t, store, "s1", 2, `{}`) + + code, body := apiReq(t, http.MethodGet, ts.URL+"/api/sessions/s1/stats", testToken, "") + if code != http.StatusOK { + t.Fatalf("stats status = %d", code) + } + var stats map[string]any + if err := json.Unmarshal([]byte(body), &stats); err != nil { + t.Fatalf("decode: %v", err) + } + if stats["turns"].(float64) != 2 || + stats["inputTokens"].(float64) != 1200 || + stats["outputTokens"].(float64) != 340 || + stats["totalCost"].(float64) != 0.25 { + t.Fatalf("stats = %v", stats) + } + if _, ok := stats["sessionsCount"]; ok { + t.Fatalf("per-session stats must not carry sessionsCount: %v", stats) + } + + // unknown session: 200 with zeros (matches events endpoint behavior) + code, body = apiReq(t, http.MethodGet, ts.URL+"/api/sessions/ghost/stats", testToken, "") + if code != http.StatusOK { + t.Fatalf("ghost status = %d", code) + } + var ghost map[string]any + if err := json.Unmarshal([]byte(body), &ghost); err != nil { + t.Fatalf("decode ghost: %v", err) + } + if len(ghost) != 4 || ghost["turns"].(float64) != 0 { + t.Fatalf("ghost stats = %v, want four zero fields", ghost) + } +} + +func TestAPIStatsTotalsAndOnlineCount(t *testing.T) { + ts, store := newTestServer(t) + appendAgentEnd(t, store, "s1", 1, `{"usage":{"inputTokens":1200000,"outputTokens":340000,"totalCost":4.2}}`) + appendAgentEnd(t, store, "s2", 1, `{"usage":{"inputTokens":50,"outputTokens":10,"totalCost":0}}`) + + get := func() map[string]any { + t.Helper() + code, body := apiReq(t, http.MethodGet, ts.URL+"/api/stats", testToken, "") + if code != http.StatusOK { + t.Fatalf("/api/stats status = %d", code) + } + var m map[string]any + if err := json.Unmarshal([]byte(body), &m); err != nil { + t.Fatalf("decode: %v", err) + } + return m + } + + totals := get() + if totals["turns"].(float64) != 2 || + totals["inputTokens"].(float64) != 1200050 || + totals["outputTokens"].(float64) != 340010 || + totals["totalCost"].(float64) != 4.2 || + totals["sessionsCount"].(float64) != 2 { + t.Fatalf("totals = %v", totals) + } + if totals["onlineCount"].(float64) != 0 { + t.Fatalf("onlineCount = %v, want 0 with no agents", totals["onlineCount"]) + } + + // one registered agent connection must show up as online + ws := dialAgent(t, ts) + if err := ws.WriteJSON(helloFrame("s-live")); err != nil { + t.Fatalf("hello: %v", err) + } + if _, _, err := ws.ReadMessage(); err != nil { // welcome + t.Fatalf("welcome: %v", err) + } + online := get() + if online["onlineCount"].(float64) != 1 { + t.Fatalf("onlineCount = %v, want 1 after hello", online["onlineCount"]) + } +} + +func TestAPIStatsClosedStore500(t *testing.T) { + ts, store := newTestServer(t) + if err := store.Close(); err != nil { + t.Fatalf("close store: %v", err) + } + if code, _ := apiReq(t, http.MethodGet, ts.URL+"/api/stats", testToken, ""); code != http.StatusInternalServerError { + t.Fatal("stats with broken store must 500") + } + if code, _ := apiReq(t, http.MethodGet, ts.URL+"/api/sessions/s1/stats", testToken, ""); code != http.StatusInternalServerError { + t.Fatal("session stats with broken store must 500") + } +} diff --git a/daemon/docker.go b/daemon/docker.go index 5c05585..a3c684f 100644 --- a/daemon/docker.go +++ b/daemon/docker.go @@ -118,10 +118,12 @@ type SpawnJob struct { UpdatedAt int64 `json:"-"` } -// SpawnResult is the immediate POST /api/spawn response. +// SpawnResult is the immediate POST /api/spawn response. ImageUsed is the +// image the job will create the container from (registry lookup; the build +// itself may still fail on a missing custom image). type SpawnResult struct { - SessionID string `json:"sessionId"` - ContainerID string `json:"containerId"` + SessionID string `json:"sessionId"` + ImageUsed string `json:"imageUsed"` } // Spawner owns the docker client, per-repo clone serialization and job state. @@ -242,7 +244,16 @@ func (s *Spawner) Start(ctx context.Context, repo, branch string) (SpawnResult, sessionID := newUUID() s.setJob(sessionID, repo, stateCloning, "", "") go s.runJob(repo, branch, sessionID) - return SpawnResult{SessionID: sessionID}, nil + return SpawnResult{SessionID: sessionID, ImageUsed: s.resolveImage(repo)}, nil +} + +// resolveImage returns the image a repo's spawns use: the registered custom +// image when present, else the default worker image. +func (s *Spawner) resolveImage(repo string) string { + if custom, ok, _ := s.store.GetRepoImage(repo); ok && custom != "" { + return custom + } + return imageRefWorker } func (s *Spawner) imageExists(ctx context.Context) (bool, error) { @@ -287,7 +298,7 @@ func (s *Spawner) runJob(repo, branch, sessionID string) { return } s.setJob(sessionID, repo, stateCreating, "", "") - containerID, err := s.createAndStart(s.ctx, repo, slug, sessionID) + containerID, image, err := s.createAndStart(s.ctx, repo, slug, sessionID) if err != nil { s.setJob(sessionID, repo, stateError, "", err.Error()) return @@ -296,7 +307,7 @@ func (s *Spawner) runJob(repo, branch, sessionID string) { s.setJob(sessionID, repo, stateError, containerID, "container started but not persisted: "+err.Error()) return } - s.setJob(sessionID, repo, stateRunning, containerID, "") + s.setJob(sessionID, repo, stateRunning, containerID, "image "+image) } // cloneOrUpdate clones the repo into reposDir/ or fast-forwards it. @@ -466,22 +477,21 @@ func extractBuildError(body []byte) string { // createAndStart provisions volumes, creates and starts the worker container. // A repo-registered custom image (see /api/repos) overrides the default // worker image; the ops agent builds and registers those. -func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID string) (string, error) { - image := imageRefWorker - if custom, ok, _ := s.store.GetRepoImage(repo); ok && custom != "" { - exists, err := s.imageRefExists(ctx, custom) +func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID string) (string, string, error) { + image := s.resolveImage(repo) + if image != imageRefWorker { + exists, err := s.imageRefExists(ctx, image) if err != nil { - return "", fmt.Errorf("docker unavailable: %w", err) + return "", "", fmt.Errorf("docker unavailable: %w", err) } if !exists { - return "", fmt.Errorf("custom image %s not built (ask the ops agent to build it)", custom) + return "", "", fmt.Errorf("custom image %s not built (ask the ops agent to build it)", image) } - image = custom } repoVolume := volumeRepoPrefix + slug fresh, err := s.ensureRepoVolume(ctx, repoVolume) if err != nil { - return "", err + return "", "", err } if fresh { if err := s.seedVolume(ctx, slug, repoVolume, sessionID); err != nil { @@ -490,14 +500,14 @@ func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID stri if rmErr := s.cli.VolumeRemove(context.Background(), repoVolume, true); rmErr != nil { log.Printf("spawner: remove failed seed volume %s: %v", repoVolume, rmErr) } - return "", fmt.Errorf("seed %s: %w", repoVolume, err) + return "", "", fmt.Errorf("seed %s: %w", repoVolume, err) } } if _, err := s.cli.VolumeCreate(ctx, volume.CreateOptions{Name: volumeSessions}); err != nil { - return "", fmt.Errorf("volume %s: %w", volumeSessions, err) + return "", "", fmt.Errorf("volume %s: %w", volumeSessions, err) } if _, err := s.cli.VolumeCreate(ctx, volume.CreateOptions{Name: volumePiCache}); err != nil { - return "", fmt.Errorf("volume %s: %w", volumePiCache, err) + return "", "", fmt.Errorf("volume %s: %w", volumePiCache, err) } binds := []string{ @@ -526,13 +536,13 @@ func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID stri name := "lvmh-agent-" + strings.ReplaceAll(sessionID, "-", "")[:12] created, err := s.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, name) if err != nil { - return "", fmt.Errorf("docker create: %w", err) + return "", "", fmt.Errorf("docker create: %w", err) } if err := s.cli.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil { s.removeContainer(context.Background(), created.ID) - return "", fmt.Errorf("docker start: %w", err) + return "", "", fmt.Errorf("docker start: %w", err) } - return created.ID, nil + return created.ID, image, nil } // ensureRepoVolume creates the per-repo volume; reports whether it is fresh. diff --git a/daemon/docker_test.go b/daemon/docker_test.go index e3e371e..913ca3b 100644 --- a/daemon/docker_test.go +++ b/daemon/docker_test.go @@ -66,6 +66,7 @@ type fakeDocker struct { failWait bool failVolumeCreate bool failVolumeDelete bool + failImages bool failVolumeInspect bool failArchive bool archiveHang bool @@ -229,6 +230,11 @@ func (f *fakeDocker) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.Header().Set("Api-Version", "1.44") w.WriteHeader(http.StatusOK) case call.Method == http.MethodGet && call.Path == "/images/json": + + if f.failImages { + writeJSONNow(w, http.StatusInternalServerError, `{"message":"images endpoint broken"}`) + return + } writeJSONNow(w, http.StatusOK, "["+f.imageListJSON(r.URL.Query().Get("filters"))+"]") case call.Method == http.MethodPost && call.Path == "/build": if f.failBuildHTTP { diff --git a/daemon/hub.go b/daemon/hub.go index 99211d7..4271279 100644 --- a/daemon/hub.go +++ b/daemon/hub.go @@ -30,6 +30,9 @@ const ( evBye string = "bye" evPrompt string = "prompt" evAbort string = "abort" + evSetModel string = "set_model" + evRename string = "rename" + evErrorNotice string = "error_notice" frameSessionList string = "session_list" frameEvents string = "events" frameSpawnStatus string = "spawn_status" @@ -647,6 +650,29 @@ func (h *Hub) Abort(sessionID string) error { }) } +// SetModel routes a set_model frame to the live plugin conn for sessionID. +func (h *Hub) SetModel(sessionID, provider, modelID string) error { + return h.sendToAgent(sessionID, map[string]any{ + "v": daemonVersion, + "type": evSetModel, + "sessionId": sessionID, + "ts": time.Now().UnixMilli(), + "provider": provider, + "modelId": modelID, + }) +} + +// Rename routes a rename frame to the live plugin conn for sessionID. +func (h *Hub) Rename(sessionID, name string) error { + return h.sendToAgent(sessionID, map[string]any{ + "v": daemonVersion, + "type": evRename, + "sessionId": sessionID, + "ts": time.Now().UnixMilli(), + "name": name, + }) +} + func (h *Hub) sendToAgent(sessionID string, frameBody map[string]any) error { b, err := json.Marshal(frameBody) if err != nil { @@ -689,3 +715,10 @@ func (h *Hub) ServeWebWS(w http.ResponseWriter, r *http.Request) { c.deliverControl(payload) } } + +// OnlineCount reports how many agent WS connections are currently live. +func (h *Hub) OnlineCount() int { + h.mu.Lock() + defer h.mu.Unlock() + return len(h.agents) +} diff --git a/daemon/model_rename_test.go b/daemon/model_rename_test.go new file mode 100644 index 0000000..8486051 --- /dev/null +++ b/daemon/model_rename_test.go @@ -0,0 +1,238 @@ +package main + +// model_rename_test.go — slice 1: set_model/rename routing, SetSessionName +// merge + session_list broadcast, model catalog parsing and PATCH semantics. + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +func TestAPISetModelRoutingAndOffline409(t *testing.T) { + ts, _ := newTestServer(t) + + if code, _ := apiReq(t, http.MethodPost, ts.URL+"/api/sessions/s1/model", testToken, + `{"provider":"zai-renaud","modelId":"glm-5.3"}`); code != http.StatusConflict { + t.Fatalf("offline set_model = %d, want 409", code) + } + + ws := dialAgent(t, ts) + _ = ws.WriteJSON(helloFrame("s1")) + _ = readFrame(t, ws) + + if code, _ := apiReq(t, http.MethodPost, ts.URL+"/api/sessions/s1/model", testToken, + `{"modelId":"glm-5.3"}`); code != http.StatusBadRequest { + t.Fatalf("set_model without provider = %d, want 400", code) + } + if code, _ := apiReq(t, http.MethodPost, ts.URL+"/api/sessions/s1/model", testToken, + `{"provider":"zai-renaud"}`); code != http.StatusBadRequest { + t.Fatalf("set_model without modelId = %d, want 400", code) + } + + if code, body := apiReq(t, http.MethodPost, ts.URL+"/api/sessions/s1/model", testToken, + `{"provider":"zai-renaud","modelId":"glm-5.3"}`); code != http.StatusOK || !strings.Contains(body, `"ok":true`) { + t.Fatalf("online set_model = %d %s, want 200 ok", code, body) + } + got := readFrame(t, ws) + if got["type"] != evSetModel || got["provider"] != "zai-renaud" || got["modelId"] != "glm-5.3" { + t.Fatalf("set_model frame = %v", got) + } +} + +func TestStoreSetSessionNameMergesIntoInfoBlob(t *testing.T) { + store := openTestStore(t) + name := "original" + if err := store.UpsertSession(SessionInfo{ + ID: "s1", Name: &name, Cwd: "/w", Model: "glm-5.3", + Provider: "zai-renaud", StartedAt: 5, + }); err != nil { + t.Fatalf("upsert: %v", err) + } + found, err := store.SetSessionName("s1", "renamed") + if err != nil || !found { + t.Fatalf("SetSessionName = %v %v, want true nil", found, err) + } + rows, err := store.Sessions() + if err != nil || len(rows) != 1 { + t.Fatalf("sessions: %v %v", rows, err) + } + info := rows[0].Info + if info.Name == nil || *info.Name != "renamed" { + t.Fatalf("name = %v, want renamed", info.Name) + } + if info.Cwd != "/w" || info.Model != "glm-5.3" || info.Provider != "zai-renaud" || info.StartedAt != 5 { + t.Fatalf("merge clobbered sibling fields: %+v", info) + } + + found, err = store.SetSessionName("missing", "x") + if err != nil || found { + t.Fatalf("SetSessionName unknown id = %v %v, want false nil", found, err) + } +} + +// dialWebWS opens the browser websocket and returns a frame channel. +func dialWebWS(t *testing.T, ts *httptest.Server) <-chan map[string]any { + t.Helper() + wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/ws?token=" + testToken + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial web ws: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + frames := make(chan map[string]any, 64) + go func() { + defer close(frames) + for { + var m map[string]any + if err := conn.ReadJSON(&m); err != nil { + return + } + frames <- m + } + }() + return frames +} + +// waitSessionList reads frames until a session_list names the session `id` +// with the wanted name (nil-safe), or fails on timeout. +func waitSessionList(t *testing.T, frames <-chan map[string]any, id, wantName string) { + t.Helper() + deadline := time.After(3 * time.Second) + for { + select { + case m, ok := <-frames: + if !ok { + t.Fatal("web ws closed while waiting for session_list") + } + if m["type"] != frameSessionList { + continue + } + for _, s := range m["sessions"].([]any) { + row := s.(map[string]any) + if row["id"] != id { + continue + } + if name, _ := row["name"].(string); name == wantName { + return + } + } + case <-deadline: + t.Fatalf("no session_list with %s name %q before timeout", id, wantName) + } + } +} + +func TestAPIRenamePatchPersistsBroadcastsAndRoutes(t *testing.T) { + ts, store := newTestServer(t) + frames := dialWebWS(t, ts) + + // The initial session_list arrives on connect; the agent hello broadcasts + // another. Consume both before PATCHing, then expect the renamed list. + ws := dialAgent(t, ts) + _ = ws.WriteJSON(helloFrame("s1")) + _ = readFrame(t, ws) + waitSessionList(t, frames, "s1", "") + + if code, _ := apiReq(t, http.MethodPatch, ts.URL+"/api/sessions/s1", testToken, + `{"name":"renamed"}`); code != http.StatusOK { + t.Fatalf("patch rename = %d, want 200", code) + } + waitSessionList(t, frames, "s1", "renamed") + + rows, err := store.Sessions() + if err != nil || len(rows) != 1 || rows[0].Info.Name == nil || *rows[0].Info.Name != "renamed" { + t.Fatalf("persisted name after patch: %v %v", rows, err) + } + + // live plugin receives the rename frame too + got := readFrame(t, ws) + if got["type"] != evRename || got["name"] != "renamed" { + t.Fatalf("rename frame = %v", got) + } + + if code, _ := apiReq(t, http.MethodPatch, ts.URL+"/api/sessions/s1", testToken, + `{"name":" "}`); code != http.StatusBadRequest { + t.Fatalf("blank rename = %d, want 400", code) + } + if code, _ := apiReq(t, http.MethodPatch, ts.URL+"/api/sessions/missing", testToken, + `{"name":"x"}`); code != http.StatusNotFound { + t.Fatalf("unknown session rename = %d, want 404", code) + } +} + +func TestParseModelCatalog(t *testing.T) { + body := []byte(`{ + "providers": { + "zai-renaud": {"models": [ + {"id": "glm-5.3", "name": "GLM-5.3"}, + {"id": "glm-5.4"}, + {"name": "idless, skipped"} + ]}, + "anthropic": {"models": [{"id": "claude-x", "name": "Claude X"}]} + } + }`) + got := parseModelCatalog(body) + want := []ModelCatalogItem{ + {Provider: "anthropic", ID: "claude-x", Name: "Claude X"}, + {Provider: "zai-renaud", ID: "glm-5.3", Name: "GLM-5.3"}, + {Provider: "zai-renaud", ID: "glm-5.4", Name: "glm-5.4"}, + } + if len(got) != len(want) { + t.Fatalf("catalog = %+v", got) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("catalog[%d] = %+v, want %+v", i, got[i], want[i]) + } + } + if parseModelCatalog([]byte(`not json`)) != nil { + t.Fatal("malformed body must parse to nil") + } + if out := parseModelCatalog([]byte(`{"providers":{}}`)); len(out) != 0 { + t.Fatalf("empty providers = %+v, want empty", out) + } +} + +func TestAPIModelCatalog(t *testing.T) { + ts, _ := newTestServer(t) + dir := t.TempDir() + + good := filepath.Join(dir, "models.json") + if err := os.WriteFile(good, []byte(`{"providers":{"zai-renaud":{"models":[{"id":"glm-5.3","name":"GLM-5.3"}]}}}`), 0o644); err != nil { + t.Fatalf("write models.json: %v", err) + } + t.Setenv(envModelsFile, good) + code, body := apiReq(t, http.MethodGet, ts.URL+"/api/model-catalog", testToken, "") + if code != http.StatusOK { + t.Fatalf("catalog = %d %s", code, body) + } + var entries []ModelCatalogItem + if err := json.Unmarshal([]byte(body), &entries); err != nil { + t.Fatalf("decode catalog: %v", err) + } + if len(entries) != 1 || entries[0] != (ModelCatalogItem{Provider: "zai-renaud", ID: "glm-5.3", Name: "GLM-5.3"}) { + t.Fatalf("entries = %+v", entries) + } + + bad := filepath.Join(dir, "bad.json") + if err := os.WriteFile(bad, []byte(`{`), 0o644); err != nil { + t.Fatalf("write bad.json: %v", err) + } + t.Setenv(envModelsFile, bad) + if code, _ := apiReq(t, http.MethodGet, ts.URL+"/api/model-catalog", testToken, ""); code != http.StatusInternalServerError { + t.Fatalf("malformed catalog = %d, want 500", code) + } + + t.Setenv(envModelsFile, filepath.Join(dir, "absent.json")) + if code, _ := apiReq(t, http.MethodGet, ts.URL+"/api/model-catalog", testToken, ""); code != http.StatusInternalServerError { + t.Fatalf("missing catalog file = %d, want 500", code) + } +} diff --git a/daemon/spawner_test.go b/daemon/spawner_test.go index 1d9c39c..cfa7fe0 100644 --- a/daemon/spawner_test.go +++ b/daemon/spawner_test.go @@ -33,8 +33,8 @@ func TestSpawnerStartHappyPath(t *testing.T) { if err != nil { t.Fatalf("Start: %v", err) } - if res.SessionID == "" || res.ContainerID != "" { - t.Fatalf("Start result = %+v, want sessionId set and empty containerId", res) + if res.SessionID == "" || res.ImageUsed != imageRefWorker { + t.Fatalf("Start result = %+v, want sessionId set and default image", res) } job := waitJobState(t, sp, res.SessionID, stateRunning) @@ -1278,7 +1278,7 @@ func TestSpawnerCustomImageCheckDockerDead(t *testing.T) { t.Fatalf("set repo image: %v", err) } sp.images0AndDead(t) - _, err := sp.createAndStart(context.Background(), "group/project", repoSlug("group/project"), "s1") + _, _, err := sp.createAndStart(context.Background(), "group/project", repoSlug("group/project"), "s1") if err == nil || !strings.Contains(err.Error(), "docker unavailable") { t.Fatalf("createAndStart with dead docker = %v, want docker unavailable", err) } diff --git a/daemon/store.go b/daemon/store.go index 3991440..0cb4162 100644 --- a/daemon/store.go +++ b/daemon/store.go @@ -51,6 +51,7 @@ type ContainerRow struct { type RepoImageRow struct { Repo string `json:"repo"` Image string `json:"image"` + Built bool `json:"built"` } const dsnParams string = "?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)" @@ -207,6 +208,31 @@ func (s *Store) SetOnline(sessionID string, online bool) error { return err } +// SetSessionName merges a new name into the persisted session info blob, +// leaving every other snapshot field untouched. Returns false when the +// session row does not exist. +func (s *Store) SetSessionName(sessionID, name string) (bool, error) { + var infoBlob string + err := s.db.QueryRow(`SELECT info FROM sessions WHERE id=?`, sessionID).Scan(&infoBlob) + if err == sql.ErrNoRows { + return false, nil + } + if err != nil { + return false, err + } + var info SessionInfo + if err := json.Unmarshal([]byte(infoBlob), &info); err != nil { + return false, fmt.Errorf("decode session info %s: %w", sessionID, err) + } + info.Name = &name + blob, err := json.Marshal(info) + if err != nil { + return true, err + } + _, err = s.db.Exec(`UPDATE sessions SET info=? WHERE id=?`, string(blob), sessionID) + return true, err +} + // Sessions returns every known session row ordered by session id. func (s *Store) Sessions() ([]SessionRow, error) { rows, err := s.db.Query(`SELECT id, info, lastSeq, lastEventAt, online FROM sessions ORDER BY id`) @@ -233,6 +259,101 @@ func (s *Store) Sessions() ([]SessionRow, error) { return out, rows.Err() } +// DeleteSessionEvents drops every persisted event for the session and +// returns the number of rows removed (0 when the session had none). +func (s *Store) DeleteSessionEvents(sessionID string) (int64, error) { + res, err := s.db.Exec(`DELETE FROM events WHERE sessionId=?`, sessionID) + if err != nil { + return 0, err + } + return res.RowsAffected() +} + +// DeleteSession removes a session entirely: its events, container row and +// the session row itself. +func (s *Store) DeleteSession(sessionID string) error { + if _, err := s.db.Exec(`DELETE FROM events WHERE sessionId=?`, sessionID); err != nil { + return err + } + if _, err := s.db.Exec(`DELETE FROM containers WHERE sessionId=?`, sessionID); err != nil { + return err + } + _, err := s.db.Exec(`DELETE FROM sessions WHERE id=?`, sessionID) + return err +} + +// UsageStats aggregates token/cost usage over agent_end events. +type UsageStats struct { + Turns int64 `json:"turns"` + InputTokens int64 `json:"inputTokens"` + OutputTokens int64 `json:"outputTokens"` + TotalCost float64 `json:"totalCost"` +} + +// StatsTotals is UsageStats aggregated across all sessions, plus how many +// sessions contributed agent_end events. +type StatsTotals struct { + UsageStats + SessionsCount int64 `json:"sessionsCount"` +} + +// agentEndUsage is the usage sub-object of an agent_end payload; absent +// fields decode as zero. +type agentEndUsage struct { + Usage struct { + InputTokens int64 `json:"inputTokens"` + OutputTokens int64 `json:"outputTokens"` + TotalCost float64 `json:"totalCost"` + } `json:"usage"` +} + +// SessionStats aggregates usage over a session's persisted agent_end events. +// Each agent_end row counts as one turn; rows without (or with partial) +// usage payloads contribute zeros. +func (s *Store) SessionStats(sessionID string) (UsageStats, error) { + rows, err := s.db.Query(`SELECT sessionId, payload FROM events WHERE sessionId=? AND type=?`, + sessionID, evAgentEnd) + if err != nil { + return UsageStats{}, err + } + defer rows.Close() + totals, err := scanAgentEndUsage(rows) + return totals.UsageStats, err +} + +// StatsTotals aggregates usage over every persisted agent_end event. +func (s *Store) StatsTotals() (StatsTotals, error) { + rows, err := s.db.Query(`SELECT sessionId, payload FROM events WHERE type=?`, evAgentEnd) + if err != nil { + return StatsTotals{}, err + } + defer rows.Close() + return scanAgentEndUsage(rows) +} + +// scanAgentEndUsage folds agent_end rows into totals; unparseable payloads +// still count as turns but add no usage. +func scanAgentEndUsage(rows *sql.Rows) (StatsTotals, error) { + var out StatsTotals + seen := make(map[string]struct{}) + for rows.Next() { + var sessionID, payload string + if err := rows.Scan(&sessionID, &payload); err != nil { + return out, err + } + seen[sessionID] = struct{}{} + out.Turns++ + var u agentEndUsage + if json.Unmarshal([]byte(payload), &u) == nil { + out.InputTokens += u.Usage.InputTokens + out.OutputTokens += u.Usage.OutputTokens + out.TotalCost += u.Usage.TotalCost + } + } + out.SessionsCount = int64(len(seen)) + return out, rows.Err() +} + // GetSetting returns a settings value and whether it exists. func (s *Store) GetSetting(key string) (string, bool, error) { var val string diff --git a/daemon/store_extra_test.go b/daemon/store_extra_test.go index 7a1972f..37b8ef1 100644 --- a/daemon/store_extra_test.go +++ b/daemon/store_extra_test.go @@ -166,3 +166,85 @@ func TestStoreEventsLatestAndBefore(t *testing.T) { t.Fatalf("latest unknown session = %d events, want 0", len(missing)) } } + +func TestStoreDeleteSessionEvents(t *testing.T) { + store := openTestStore(t) + for _, e := range []Event{ + {SessionID: "s1", Seq: 1, TS: 10, Type: evMessageStart, Payload: []byte(`{}`)}, + {SessionID: "s1", Seq: 2, TS: 11, Type: evMessageEnd, Payload: []byte(`{}`)}, + {SessionID: "s2", Seq: 1, TS: 12, Type: evAgentSettled, Payload: []byte(`{}`)}, + } { + if err := store.AppendEvent(e); err != nil { + t.Fatalf("append %s#%d: %v", e.SessionID, e.Seq, err) + } + } + + deleted, err := store.DeleteSessionEvents("s1") + if err != nil || deleted != 2 { + t.Fatalf("delete s1 events = %d %v, want 2 nil", deleted, err) + } + if evts, _ := store.EventsAfter("s1", 0, 100); len(evts) != 0 { + t.Fatalf("s1 events after delete = %d, want 0", len(evts)) + } + if last, _ := store.LastSeq("s1"); last != 0 { + t.Fatalf("s1 lastSeq after delete = %d, want 0", last) + } + // other sessions untouched + evts, err := store.EventsAfter("s2", 0, 100) + if err != nil || len(evts) != 1 || evts[0].Seq != 1 { + t.Fatalf("s2 events = %v %v, want seq 1 only", evts, err) + } + + // unknown / already-cleared session: 0 rows, no error + if deleted, err := store.DeleteSessionEvents("s1"); err != nil || deleted != 0 { + t.Fatalf("second delete = %d %v, want 0 nil", deleted, err) + } + if deleted, err := store.DeleteSessionEvents("ghost"); err != nil || deleted != 0 { + t.Fatalf("unknown session = %d %v, want 0 nil", deleted, err) + } +} + +func TestStoreDeleteSession(t *testing.T) { + store := openTestStore(t) + name := "gone" + if err := store.UpsertSession(SessionInfo{ID: "s1", Name: &name, Cwd: "/w", Model: "glm-5.3", Provider: "zai-renaud", StartedAt: 1}); err != nil { + t.Fatalf("upsert s1: %v", err) + } + if err := store.UpsertSession(SessionInfo{ID: "s2", Cwd: "/w2", Model: "glm-5.3", Provider: "zai-renaud", StartedAt: 2}); err != nil { + t.Fatalf("upsert s2: %v", err) + } + if err := store.UpsertContainer("s1", "cid", "group/proj"); err != nil { + t.Fatalf("upsert container: %v", err) + } + for _, e := range []Event{ + {SessionID: "s1", Seq: 1, TS: 10, Type: evMessageStart, Payload: []byte(`{}`)}, + {SessionID: "s2", Seq: 1, TS: 10, Type: evMessageStart, Payload: []byte(`{}`)}, + } { + if err := store.AppendEvent(e); err != nil { + t.Fatalf("append: %v", err) + } + } + + if err := store.DeleteSession("s1"); err != nil { + t.Fatalf("delete session: %v", err) + } + rows, err := store.Sessions() + if err != nil || len(rows) != 1 || rows[0].ID != "s2" { + t.Fatalf("sessions after delete = %v %v, want only s2", rows, err) + } + if evts, _ := store.EventsAfter("s1", 0, 100); len(evts) != 0 { + t.Fatalf("s1 events after delete = %d, want 0", len(evts)) + } + if _, ok, _ := store.GetContainer("s1"); ok { + t.Fatal("s1 container row should be gone") + } + // s2 untouched + if evts, _ := store.EventsAfter("s2", 0, 100); len(evts) != 1 { + t.Fatalf("s2 events after delete = %d, want 1", len(evts)) + } + + // idempotent: unknown session deletes cleanly + if err := store.DeleteSession("ghost"); err != nil { + t.Fatalf("delete unknown session: %v", err) + } +} diff --git a/daemon/store_stats_test.go b/daemon/store_stats_test.go new file mode 100644 index 0000000..83d9124 --- /dev/null +++ b/daemon/store_stats_test.go @@ -0,0 +1,85 @@ +package main + +// store_stats_test.go — SessionStats/StatsTotals aggregation over agent_end. + +import ( + "testing" +) + +func appendAgentEnd(t *testing.T, s *Store, sessionID string, seq int64, payload string) { + t.Helper() + if err := s.AppendEvent(Event{ + SessionID: sessionID, Seq: seq, TS: 1, Type: evAgentEnd, Payload: []byte(payload), + }); err != nil { + t.Fatalf("append agent_end %s/%d: %v", sessionID, seq, err) + } +} + +func TestStoreSessionStats(t *testing.T) { + store := openTestStore(t) + appendAgentEnd(t, store, "s1", 1, `{"usage":{"inputTokens":100,"outputTokens":50,"totalCost":0.25}}`) + appendAgentEnd(t, store, "s1", 2, `{"usage":{"inputTokens":30,"outputTokens":20,"totalCost":0.5}}`) + appendAgentEnd(t, store, "s1", 3, `{}`) // no usage → zeros + appendAgentEnd(t, store, "s1", 4, `{"usage":{"inputTokens":7}}`) // partial usage + if err := store.AppendEvent(Event{SessionID: "s1", Seq: 5, TS: 1, Type: evMessageEnd, Payload: []byte(`{}`)}); err != nil { + t.Fatalf("append message_end: %v", err) + } + + stats, err := store.SessionStats("s1") + if err != nil { + t.Fatalf("session stats: %v", err) + } + if stats.Turns != 4 || stats.InputTokens != 137 || stats.OutputTokens != 70 || stats.TotalCost != 0.75 { + t.Fatalf("stats = %+v, want turns 4, input 137, output 70, cost 0.75", stats) + } + + unknown, err := store.SessionStats("nope") + if err != nil { + t.Fatalf("unknown session stats: %v", err) + } + if unknown != (UsageStats{}) { + t.Fatalf("unknown session stats = %+v, want zeros", unknown) + } +} + +func TestStoreSessionStatsCorruptPayloadCountsTurn(t *testing.T) { + store := openTestStore(t) + appendAgentEnd(t, store, "s1", 1, `{"usage":{"inputTokens":5,"outputTokens":5,"totalCost":0.1}}`) + if err := store.AppendEvent(Event{SessionID: "s1", Seq: 2, TS: 1, Type: evAgentEnd, Payload: []byte(`not-json`)}); err != nil { + t.Fatalf("append corrupt: %v", err) + } + stats, err := store.SessionStats("s1") + if err != nil { + t.Fatalf("stats: %v", err) + } + if stats.Turns != 2 || stats.InputTokens != 5 || stats.OutputTokens != 5 || stats.TotalCost != 0.1 { + t.Fatalf("stats = %+v, want turns 2 with usage only from row 1", stats) + } +} + +func TestStoreStatsTotals(t *testing.T) { + store := openTestStore(t) + empty, err := store.StatsTotals() + if err != nil { + t.Fatalf("empty totals: %v", err) + } + if empty != (StatsTotals{}) { + t.Fatalf("empty totals = %+v, want zeros", empty) + } + + appendAgentEnd(t, store, "s1", 1, `{"usage":{"inputTokens":1000,"outputTokens":200,"totalCost":1.5}}`) + appendAgentEnd(t, store, "s1", 2, `{}`) + appendAgentEnd(t, store, "s2", 1, `{"usage":{"inputTokens":40,"outputTokens":60,"totalCost":0.25}}`) + if err := store.AppendEvent(Event{SessionID: "s2", Seq: 2, TS: 1, Type: evAgentSettled, Payload: []byte(`{}`)}); err != nil { + t.Fatalf("append settled: %v", err) + } + + totals, err := store.StatsTotals() + if err != nil { + t.Fatalf("totals: %v", err) + } + want := StatsTotals{UsageStats: UsageStats{Turns: 3, InputTokens: 1040, OutputTokens: 260, TotalCost: 1.75}, SessionsCount: 2} + if totals != want { + t.Fatalf("totals = %+v, want %+v", totals, want) + } +} diff --git a/plugin/lvmh-agent.ts b/plugin/lvmh-agent.ts index 7dc5e87..c5f0b81 100644 --- a/plugin/lvmh-agent.ts +++ b/plugin/lvmh-agent.ts @@ -556,6 +556,8 @@ export default function (pi: ExtensionAPI): void { if (f.type === "welcome") onWelcome(f as Record); else if (f.type === "prompt") deliverPrompt(f); else if (f.type === "abort") doAbort(); + else if (f.type === "set_model") applySetModel(f); + else if (f.type === "rename") applyRename(f); // unknown types are ignored (forward compatibility) } @@ -587,6 +589,96 @@ export default function (pi: ExtensionAPI): void { } } + /** Daemon-requested model switch. Success is mirrored by the resulting + * model_select event (which emits session_info); every failure path + * emits error_notice so the web sees why nothing changed. */ + function applySetModel(frame: { + modelId?: unknown; + provider?: unknown; + sessionId?: unknown; + }): void { + const modelId = frame.modelId; + const provider = frame.provider; + if ( + typeof frame.sessionId === "string" && + frame.sessionId !== currentSessionId + ) + return; + if (typeof modelId !== "string" || modelId.length === 0) { + emit("error_notice", { reason: "set_model without modelId" }); + return; + } + if (typeof provider !== "string" || provider.length === 0) { + emit("error_notice", { reason: "set_model without provider" }); + return; + } + const registry = lastCtx?.modelRegistry as + | { find?: (provider: string, modelId: string) => unknown } + | undefined; + if (registry === undefined || typeof registry.find !== "function") { + emit("error_notice", { reason: "model registry unavailable" }); + return; + } + let model: unknown; + try { + model = registry.find(provider, modelId); + } catch (err) { + log(`model lookup failed: ${err}`); + emit("error_notice", { + reason: `model lookup failed: ${String(err)}`, + }); + return; + } + if (model === undefined || model === null) { + emit("error_notice", { + reason: `model not found: ${provider}/${modelId}`, + }); + return; + } + const maybeSet = (pi as unknown as { setModel?: (m: unknown) => unknown }) + .setModel; + if (typeof maybeSet !== "function") { + emit("error_notice", { reason: "pi.setModel unavailable" }); + return; + } + try { + Promise.resolve(maybeSet.call(pi, model)).then( + (ok: unknown) => { + if (ok !== true) + emit("error_notice", { + reason: `setModel rejected ${provider}/${modelId} (no API key?)`, + }); + }, + (err: unknown) => { + log(`setModel failed: ${err}`); + emit("error_notice", { + reason: `setModel failed: ${String(err)}`, + }); + }, + ); + } catch (err) { + log(`setModel threw: ${err}`); + emit("error_notice", { reason: `setModel failed: ${String(err)}` }); + } + } + + /** Daemon-requested rename. pi has no extension rename API: only the + * mirrored snapshot changes (PROTOCOL.md); the TUI session keeps its + * own name. session_info propagates the change to the daemon/web. */ + function applyRename(frame: { name?: unknown; sessionId?: unknown }): void { + const name = frame.name; + if (typeof name !== "string" || name.length === 0) return; + if ( + typeof frame.sessionId === "string" && + frame.sessionId !== currentSessionId + ) + return; + if (snapshot === null) return; + if (name === snapshot.name) return; + snapshot.name = name; + emit("session_info", { session: snapshot }); + } + function buildSnapshot( ctx: ExtensionContext, sessionId: string, diff --git a/plugin/smoke.ts b/plugin/smoke.ts index 25e4353..6ccbae5 100644 --- a/plugin/smoke.ts +++ b/plugin/smoke.ts @@ -37,9 +37,12 @@ interface FakePi { handlers: Map unknown>; sentMessages: Array<{ message: string; options: unknown }>; aborted: number; + setModelCalls: unknown[]; + setModelResult: boolean; on(event: string, handler: (event: unknown, ctx: unknown) => unknown): void; sendUserMessage(message: string, options?: unknown): void; abort(): void; + setModel(model: unknown): Promise; } function makeFakePi(): FakePi { @@ -47,6 +50,8 @@ function makeFakePi(): FakePi { handlers: new Map(), sentMessages: [], aborted: 0, + setModelCalls: [], + setModelResult: true, on(event, handler) { this.handlers.set(event, handler); }, @@ -56,6 +61,10 @@ function makeFakePi(): FakePi { abort() { this.aborted++; }, + setModel(model) { + this.setModelCalls.push(model); + return Promise.resolve(this.setModelResult); + }, }; } @@ -63,6 +72,10 @@ function makeFakeCtx(): unknown { return { cwd: "/work/repo", model: { id: "glm-5.3", provider: "zai-renaud" }, + modelRegistry: { + find: (provider: string, id: string): unknown => + id === "glm-5.3" ? { id, provider } : undefined, + }, sessionManager: { getSessionId: () => SESSION_ID, getSessionName: () => undefined, @@ -342,6 +355,104 @@ async function main(): Promise { await waitFor(() => pi.aborted > 0, 2000); check("3 abort calls pi.abort", pi.aborted === 1); + // --- Scenario 7: set_model / rename from the daemon ---------------- + const pushControl = (type: string, extra: Record): void => { + daemon.pushAll( + JSON.stringify({ + v: 1, + type, + sessionId: SESSION_ID, + seq: 0, + ts: Date.now(), + ...extra, + }), + ); + }; + + pushControl("set_model", { + sessionId: "other-session", + provider: "zai-renaud", + modelId: "glm-5.3", + }); + pushControl("set_model", { + provider: "zai-renaud", + modelId: "no-such-model", + }); + const unknownNotice = await waitFor( + () => + daemon.frames.some( + (f) => + f.type === "error_notice" && + (f.reason as string)?.includes("model not found"), + ), + 2000, + ); + check( + "7 unknown model emits error_notice, foreign session ignored", + unknownNotice && pi.setModelCalls.length === 0, + JSON.stringify(daemon.frames.filter((f) => f.type === "error_notice")), + ); + + pushControl("set_model", { modelId: "glm-5.3" }); + const noProvider = await waitFor( + () => + daemon.frames.some( + (f) => + f.type === "error_notice" && + (f.reason as string) === "set_model without provider", + ), + 2000, + ); + check("7 missing provider rejected", noProvider); + + pushControl("set_model", { provider: "zai-renaud", modelId: "glm-5.3" }); + await waitFor(() => pi.setModelCalls.length === 1, 2000); + check( + "7 set_model resolves via registry and calls pi.setModel", + pi.setModelCalls.length === 1 && + JSON.stringify(pi.setModelCalls[0]) === + JSON.stringify({ id: "glm-5.3", provider: "zai-renaud" }), + JSON.stringify(pi.setModelCalls), + ); + + pi.setModelResult = false; + pushControl("set_model", { provider: "zai-renaud", modelId: "glm-5.3" }); + const rejected = await waitFor( + () => + daemon.frames.some( + (f) => + f.type === "error_notice" && + (f.reason as string)?.startsWith("setModel rejected"), + ), + 2000, + ); + check("7 setModel false mirrors error_notice", rejected); + pi.setModelResult = true; + + fire("model_select", { model: { id: "glm-5.4", provider: "zai-renaud" } }); + const infoSeen = await waitFor( + () => + daemon.frames.some( + (f) => + f.type === "session_info" && + (f.session as { model?: string })?.model === "glm-5.4", + ), + 2000, + ); + check("7 model_select emits session_info", infoSeen); + + pushControl("rename", { name: "renamed-from-daemon" }); + const renameSeen = await waitFor( + () => + daemon.frames.some( + (f) => + f.type === "session_info" && + (f.session as { name?: string })?.name === "renamed-from-daemon", + ), + 2000, + ); + check("7 rename updates snapshot + session_info", renameSeen); + // --- Scenario 4: drop + reconnect, replay, overflow ---------------- const seqBeforeDrop = Math.max(...daemon.frames.map((f) => f.seq)); welcomeLastSeq = seqBeforeDrop; // daemon has everything up to the drop diff --git a/web/.pi-merge-task.md b/web/.pi-merge-task.md new file mode 100644 index 0000000..a873bf5 --- /dev/null +++ b/web/.pi-merge-task.md @@ -0,0 +1,46 @@ +# Integration task: merge 5 web slices (already-recovered blobs in ../../.pi/scratch/rec4/) + +Base: this worktree at current master + the ChatView.tsx from slice 1 (model/rename — +already placed). The other slices each edited overlapping files. Merge ALL slices. + +## Blob map (.pi/scratch/rec4/.txt) + +- Slice 1 (model switch + rename): ChatView=e2c7b1c1 (PLACED), derive=6c499294? NO — + check: derive blobs: 6c499294 (has notice? grep error_notice), 73f1de97, abf953bf. + ChatView.test=0617e495, protocol=29b7795f is slice-5's, slice-1 protocol is + 9aec478e? — IDENTIFY by features: + - s1: catalogCache, SessionModel routes, error_notice/notice-line, rename + - s2: Clear history menu, Delete session, Route.Session, trash button (SessionsView=8ab1292a? grep) + - s3: SessionStats/Route.Stats, formatTokens in store.ts, SessionsView global line + - s4 (web-only QoL): renderMarkdown in ChatStream=01f25bac(also has FAB/search), + ChatView=17c34004 (timestamps/placeholder/ArrowUp), derive=?? (ts field), css=5ba557b7 or 4b643f25 or 19c9becd (largest superset) + - s5: SpawnView badges/prepare (SpawnView=ec5f2892? grep prepare), protocol RepoPrepare/Repos/imageUsed, SpawnView.test=996b2e63 + +## Procedure + +1. protocol.ts: start from master's, add ALL routes/types from s1+s2+s3+s5 blobs + (SessionModel, ModelCatalog, Session(id) DELETE, ClearHistoryResponse, Stats, + SessionStats, StatsTotals, Repos/RepoImage/RepoPrepare, imageUsed on SpawnResponse). +2. ChatView.tsx: base = e2c7b1c1 (placed). Graft: s2 overflow menu (Clear/Delete, + from a8c6b3b0), s3 stats chip fetch (from 5ef417c8), s4 QoL pieces that live in + ChatView (search toggle+hotkey, ⋯ timestamps toggle, placeholder switch, + ArrowUp recall — from 17c34004). +3. ChatStream.tsx: base = 01f25bac (s4: markdown+FAB+search+ts). Verify it compiles + against s1's notice-line message shape; adapt. +4. derive.ts: master + s1 notice message + s4 ts field (whichever blob has both, else merge). +5. store.ts: master + s3 formatTokens (+ any stats helpers). +6. SessionsView.tsx: master + s2 trash + s3 global stats line. +7. SpawnView.tsx: master + s5 badges/prepare/imageUsed. +8. index.css: take the LARGEST blob (superset ordering: 19c9becd 22628 > 5ba557b7 21318 + > 4b643f25 21285) and verify it contains all classes used by merged components; + add missing rules from the others. +9. Tests: merge test blobs the same way (ChatView.test from 0617e495 + grafts from + s2/s3/s4 test files; SpawnView.test=996b2e63; ChatStream tests inside 1c6e7edd? + identify; derive tests). Aim: keep every test that matches merged features; drop + tests for unmerged ones. +10. Gates: bunx tsc@5.9 --noEmit && bun run build && bun run coverage — ALL four + metrics ≥95. Iterate until green. You may simplify/drop marginal features' + tests if a graft is cursed, but NEVER ship a failing suite. + +Work tree = this worktree; blobs are read-only. Report per-file what you took from +which slice + final verbatim coverage/build tails. diff --git a/web/src/ChatView.tsx b/web/src/ChatView.tsx index 59ebd66..e2c7b1c 100644 --- a/web/src/ChatView.tsx +++ b/web/src/ChatView.tsx @@ -1,6 +1,11 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { Link, useParams } from "react-router-dom"; -import type { EventFrame } from "./protocol"; +import type { + EventFrame, + ModelCatalogEntry, + RenameBody, + SetModelBody, +} from "./protocol"; import { Route } from "./protocol"; import { ApiError, errMessage, fetchJson } from "./api"; import { @@ -18,6 +23,38 @@ const HISTORY_LIMIT: number = 1000; const LATEST_QUERY: string = "latest=1"; const TEXTAREA_MAX_H: number = 200; const SEND_KEY: string = "Enter"; +const ESCAPE_KEY: string = "Escape"; + +/** Catalog cache shared across mounts and session switches: the model list + * is static for a daemon run, so it is fetched at most once per page load. */ +let catalogCache: ModelCatalogEntry[] | null = null; + +export interface CatalogGroup { + provider: string; + models: ModelCatalogEntry[]; +} + +/** Group catalog entries by provider, preserving arrival order. */ +export function groupCatalog(entries: ModelCatalogEntry[]): CatalogGroup[] { + const groups = new Map(); + for (const entry of entries) { + const list = groups.get(entry.provider); + if (list === undefined) groups.set(entry.provider, [entry]); + else list.push(entry); + } + return Array.from( + groups, + ([provider, models]): CatalogGroup => ({ + provider, + models, + }), + ); +} + +/** Test seam: drop the module-level catalog cache. */ +export function resetCatalogCache(): void { + catalogCache = null; +} interface Props { store: SessionsStore; @@ -35,6 +72,14 @@ export default function ChatView({ store, pushToast }: Props) { const [tasksOpen, setTasksOpen] = useState(false); const [hasOlder, setHasOlder] = useState(false); const [loadingOlder, setLoadingOlder] = useState(false); + const [modelOpen, setModelOpen] = useState(false); + const [catalog, setCatalog] = useState( + catalogCache, + ); + const [catalogError, setCatalogError] = useState(""); + const [switching, setSwitching] = useState(false); + const [renaming, setRenaming] = useState(false); + const [renameDraft, setRenameDraft] = useState(""); const lastSeqRef = useRef(0); const loadedRef = useRef(false); @@ -192,6 +237,65 @@ export default function ChatView({ store, pushToast }: Props) { } }; + const toggleModelPicker = (): void => { + setModelOpen((open) => !open); + if (catalogCache === null) { + setCatalogError(""); + void fetchJson(Route.ModelCatalog) + .then((list) => { + catalogCache = list; + setCatalog(list); + }) + .catch((err: unknown) => setCatalogError(errMessage(err))); + } + }; + + const pickModel = async ( + provider: string, + modelId: string, + ): Promise => { + if (switching) return; + setSwitching(true); + try { + await fetchJson(Route.SessionModel(sessionId), { + method: "POST", + body: JSON.stringify({ provider, modelId } satisfies SetModelBody), + }); + setModelOpen(false); + } catch (err) { + if (err instanceof ApiError && err.status === 409) + pushToast("session offline"); + else pushToast(errMessage(err)); + } finally { + setSwitching(false); + } + }; + + const startRename = (): void => { + setRenameDraft(session?.name ?? ""); + setRenaming(true); + }; + + const cancelRename = (): void => { + setRenaming(false); + setRenameDraft(""); + }; + + const saveRename = async (): Promise => { + const name = renameDraft.trim(); + if (name.length === 0) return; + try { + await fetchJson(Route.Session(sessionId), { + method: "PATCH", + body: JSON.stringify({ name } satisfies RenameBody), + }); + setRenaming(false); + await store.refresh(); + } catch (err) { + pushToast(errMessage(err)); + } + }; + const onKeyDown = (e: React.KeyboardEvent): void => { // IME composition: Enter confirms the candidate window, not a send if (e.nativeEvent.isComposing) return; @@ -213,10 +317,92 @@ export default function ChatView({ store, pushToast }: Props) {
-
- {session?.name ?? session?.repo ?? sessionId} + {renaming ? ( +
+ setRenameDraft(e.target.value)} + onKeyDown={(e) => { + if (e.key === SEND_KEY && !e.shiftKey) { + e.preventDefault(); + void saveRename(); + } + if (e.key === ESCAPE_KEY) cancelRename(); + }} + /> + + +
+ ) : ( + <> +
+ {session?.name ?? session?.repo ?? sessionId} +
+ + + )} +
+ + {modelOpen && ( +
+ {catalogError.length > 0 && ( +
{catalogError}
+ )} + {catalog === null && catalogError.length === 0 && ( +
loading…
+ )} + {catalog !== null && + groupCatalog(catalog).map((group) => ( +
+
{group.provider}
+ {group.models.map((m) => ( + + ))} +
+ ))} +
+ )}
-
{session?.model}
{(chat.usage.inputTokens > 0 || chat.usage.outputTokens > 0) && ( ↑{chat.usage.inputTokens.toLocaleString()} ↓