diff --git a/CHANGELOG.md b/CHANGELOG.md index f52387a..c630311 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ has a matching GitHub release whose body comes from its section below. ### Added +- Gateway: `POST /v1/berth/switch` and `POST /v1/berth/stop` switch the card to an engine or give it back without + sending a prompt. Stop never touches a pinned holder. The `/gui/` board's Launch now uses the same switch. - Linux AMD support (experimental), contributed by @jenerestain. `berth daemon --gpu-backend amd` governs a discrete RDNA3 or RDNA4 card, or a Phoenix APU (Radeon 780M), read from sysfs and DRM `fdinfo` with no ROCm. `--gpu N` names DRM cardN. On an APU, reserved VRAM and GTT (shared host diff --git a/docs/gateway.md b/docs/gateway.md index ae6dd77..e921eb6 100644 --- a/docs/gateway.md +++ b/docs/gateway.md @@ -247,6 +247,22 @@ back to, which are worth parking in RAM instead of stopping and reloading. An un or an engine `berth.toml` does not name is a 400 naming the step; a missing or wrong key and any browser `Origin` are refused, as on every other route. +## Switch or stop an engine directly + +A request naming a model switches the card on its own. To switch without sending a prompt, or to give the card back, +name the engine: + +```sh +curl -s http://127.0.0.1:9799/v1/berth/switch -H "Authorization: Bearer $(berth key)" -d '{"engine":"strata"}' +curl -s http://127.0.0.1:9799/v1/berth/stop -H "Authorization: Bearer $(berth key)" -d '{"engine":"strata"}' +``` + +- **Switch** answers `202` once the start is under way, and starts the engine under a lease the same way a request + would: the holder drains at its boundary first. It takes any engine with a `start` command; another is a `404`. +- **Stop** answers `202` with the leases berthd asked to give the card back, the same drain as `berth stop`. A + pinned holder (a game, a render, `berth hold`) is a `409`: Berth never stops it from the network. An engine that is + not running under a lease is a `409` too, and berthd not answering is a `503`. + ## Open a stopped engine's web UI A bookmark to ComfyUI or Strata's chat page is a dead link while that engine is stopped. Give the engine a diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index bebbc2a..800e411 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -207,6 +207,8 @@ func (g *Gateway) Handler() http.Handler { mux.HandleFunc("GET /v1/berth/wait", g.wait) mux.HandleFunc("GET /v1/berth/cost", g.cost) mux.HandleFunc("POST /v1/berth/plan", g.plan) + mux.HandleFunc("POST /v1/berth/switch", g.apiSwitch) + mux.HandleFunc("POST /v1/berth/stop", g.apiStop) for _, p := range []string{"/v1/chat/completions", "/v1/completions", "/v1/embeddings"} { mux.HandleFunc("POST "+p, g.forward) } diff --git a/internal/gateway/gui.go b/internal/gateway/gui.go index 4a03cc9..bd18fb3 100644 --- a/internal/gateway/gui.go +++ b/internal/gateway/gui.go @@ -350,40 +350,15 @@ func names(card []guiHold) string { return strings.Join(out, " and ") } -// guiWake starts the engine in the background and answers at once; the page -// polls /state. A second ask while a start is under way joins it. Only the -// Start button asks: opening the page never takes the card. +// guiWake is the page's Launch: a switch to the engine. Only the button +// asks; opening the page never takes the card. func (g *Gateway) guiWake(w http.ResponseWriter, r *http.Request) { e, ok := g.launchable(r.PathValue("name")) if !ok { apiError(w, http.StatusNotFound, "no engine of that name has a start command in berth.toml") return } - g.mu.Lock() - if g.wakes == nil { - g.wakes = map[string]*wake{} - } - already := g.wakes[e.Name] != nil && g.wakes[e.Name].starting - if !already { - g.wakes[e.Name] = &wake{starting: true} - } - g.mu.Unlock() - if !already { - g.log().Info("gui plane starting engine", "engine", e.Name) - go func() { - // The start outlives the page's request: a closed tab still - // leaves the engine coming up, bounded by the gateway's ready. - ctx, cancel := context.WithTimeout(context.Background(), g.ready()) - defer cancel() - err := g.ensure(ctx, e) - if err != nil { - g.log().Warn("gui plane start failed", "engine", e.Name, "err", err) - } - g.mu.Lock() - g.wakes[e.Name] = &wake{err: err} - g.mu.Unlock() - }() - } + g.switchTo(e) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusAccepted) _ = json.NewEncoder(w).Encode(guiStatus{State: "waiting", Message: "Asking berth for the card"}) diff --git a/internal/gateway/switch.go b/internal/gateway/switch.go new file mode 100644 index 0000000..7cf943f --- /dev/null +++ b/internal/gateway/switch.go @@ -0,0 +1,132 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package gateway + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + + "github.com/T-Crypt/berth/internal/config" + "github.com/T-Crypt/berth/internal/proto" + "github.com/T-Crypt/berth/internal/server" +) + +// The switchboard's actions on the keyed API: switch the card to an engine, +// or stop one. The gui plane's Launch and the harbor's board call the same +// two paths, so every surface switches the card the same way. + +// switchTo starts the engine in the background under a lease, the way a +// request for one of its models would; the caller reads the engine's state +// to follow it. A second ask while a start is under way joins it. +func (g *Gateway) switchTo(e *config.Engine) { + g.mu.Lock() + if g.wakes == nil { + g.wakes = map[string]*wake{} + } + already := g.wakes[e.Name] != nil && g.wakes[e.Name].starting + if !already { + g.wakes[e.Name] = &wake{starting: true} + } + g.mu.Unlock() + if already { + return + } + g.log().Info("switching the card", "engine", e.Name) + go func() { + // The start outlives the request that asked for it: a closed tab + // still leaves the engine coming up, bounded by the gateway's ready. + ctx, cancel := context.WithTimeout(context.Background(), g.ready()) + defer cancel() + err := g.ensure(ctx, e) + if err != nil { + g.log().Warn("switch failed", "engine", e.Name, "err", err) + } + g.mu.Lock() + g.wakes[e.Name] = &wake{err: err} + g.mu.Unlock() + }() +} + +// engineReq reads the actions' body, {"engine":"NAME"}. +func engineReq(w http.ResponseWriter, r *http.Request) (string, bool) { + var req struct { + Engine string `json:"engine"` + } + body, err := io.ReadAll(io.LimitReader(r.Body, 1<<16)) + if err != nil || json.Unmarshal(body, &req) != nil || req.Engine == "" { + apiError(w, http.StatusBadRequest, `the body must be JSON like {"engine":"NAME"}`) + return "", false + } + return req.Engine, true +} + +// apiSwitch is POST /v1/berth/switch: 202 once the start is under way. +func (g *Gateway) apiSwitch(w http.ResponseWriter, r *http.Request) { + name, ok := engineReq(w, r) + if !ok { + return + } + e, ok := g.launchable(name) + if !ok { + apiError(w, http.StatusNotFound, fmt.Sprintf("no engine %q with a start command in berth.toml", name)) + return + } + g.switchTo(e) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(guiStatus{State: "waiting", Message: "Asking berth for the card"}) +} + +// apiStop is POST /v1/berth/stop: berthd asks the engine's leases to give +// the card back at their request boundary, as `berth stop` does. Unlike the +// terminal, a remote stop never touches a pinned lease (a game, a render, +// berth hold): those end only when their owner ends them. +func (g *Gateway) apiStop(w http.ResponseWriter, r *http.Request) { + name, ok := engineReq(w, r) + if !ok { + return + } + if g.engineByName(name) == nil { + apiError(w, http.StatusNotFound, fmt.Sprintf("no engine %q in berth.toml", name)) + return + } + st := g.nodeStatus() + if st == nil { + apiError(w, http.StatusServiceUnavailable, "berthd is not answering; nothing was stopped") + return + } + for _, l := range st.Leases { + if (l.Engine == name || l.Group == name) && l.Pinned { + apiError(w, http.StatusConflict, name+" is pinned; Berth never stops it") + return + } + } + conn, raw, err := server.Dial(g.Socket) + if err != nil { + apiError(w, http.StatusServiceUnavailable, "berthd is not answering; nothing was stopped") + return + } + defer raw.Close() + if err := conn.Send(proto.Msg{Op: proto.OpStop, Engine: name}); err != nil { + apiError(w, http.StatusServiceUnavailable, "berthd: "+err.Error()) + return + } + m, err := conn.Recv() + switch { + case err != nil: + apiError(w, http.StatusServiceUnavailable, "no stop reply from berthd: "+err.Error()) + case m.Op == proto.OpRefused: + apiError(w, http.StatusConflict, m.Reason) + default: + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(struct { + Engine string `json:"engine"` + Leases []string `json:"leases"` + }{name, m.Leases}) + } +} diff --git a/internal/gateway/switch_test.go b/internal/gateway/switch_test.go new file mode 100644 index 0000000..f08e077 --- /dev/null +++ b/internal/gateway/switch_test.go @@ -0,0 +1,111 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +//go:build unix + +package gateway + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + "github.com/T-Crypt/berth/internal/broker" + "github.com/T-Crypt/berth/internal/config" +) + +// switchGateway is a gateway with one API-only engine the test binary +// serves, the shape of Strata or a bare llama-server. +func switchGateway(t *testing.T) (*httptest.Server, *broker.Broker) { + t.Helper() + sock, b := daemon(t) + addr := freeAddr(t) + t.Setenv("BERTH_GW_ENGINE", addr+",strata-model,300ms") + g := &Gateway{Socket: sock, Ready: 10 * time.Second, Key: testKey, Log: slog.New(slog.NewTextHandler(io.Discard, nil)), + Engines: []config.Engine{ + {Name: "strata", Type: "llamacpp", URL: "http://" + addr, Start: []string{os.Args[0]}, Serves: []string{"strata-model"}}, + {Name: "blender", Type: "custom", URL: "http://127.0.0.1:1"}, + }} + srv := httptest.NewServer(g.Handler()) + t.Cleanup(srv.Close) + return srv, b +} + +func until(t *testing.T, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for !ok() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(50 * time.Millisecond) + } +} + +// A switch starts the engine under a lease and answers before it is up; a +// stop gives the card back through berthd's own drain. +func TestSwitchThenStop(t *testing.T) { + srv, b := switchGateway(t) + + resp := call(t, "POST", srv.URL+"/v1/berth/switch", `{"engine":"strata"}`) + resp.Body.Close() + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("switch = %d", resp.StatusCode) + } + until(t, "strata to hold the card", func() bool { + st := b.Status() + return len(st.Leases) == 1 && st.Leases[0].Engine == "strata" && st.Leases[0].State == "active" + }) + + resp = call(t, "POST", srv.URL+"/v1/berth/stop", `{"engine":"strata"}`) + var got struct { + Engine string `json:"engine"` + Leases []string `json:"leases"` + } + _ = json.NewDecoder(resp.Body).Decode(&got) + resp.Body.Close() + if resp.StatusCode != http.StatusAccepted || got.Engine != "strata" || len(got.Leases) != 1 { + t.Fatalf("stop = %d %+v", resp.StatusCode, got) + } + until(t, "the card to be free", func() bool { return len(b.Status().Leases) == 0 }) +} + +func TestSwitchStopRefusals(t *testing.T) { + srv, b := switchGateway(t) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + l, err := b.Acquire(ctx, broker.Request{Engine: "blender", Priority: broker.Hold, Pinned: true}, nil) + if err != nil { + t.Fatal(err) + } + defer l.Release() + l.Attach(os.Getpid()) + + for _, c := range []struct { + path, body string + want int + }{ + {"/v1/berth/switch", `{"engine":"nope"}`, http.StatusNotFound}, + {"/v1/berth/switch", `{"engine":"blender"}`, http.StatusNotFound}, // no start command + {"/v1/berth/switch", `not json`, http.StatusBadRequest}, + {"/v1/berth/stop", `{"engine":"nope"}`, http.StatusNotFound}, + {"/v1/berth/stop", `{"engine":"blender"}`, http.StatusConflict}, // pinned: never stopped from here + {"/v1/berth/stop", `{"engine":"strata"}`, http.StatusConflict}, // configured, not running + } { + resp := call(t, "POST", srv.URL+c.path, c.body) + resp.Body.Close() + if resp.StatusCode != c.want { + t.Errorf("%s %s = %d, want %d", c.path, c.body, resp.StatusCode, c.want) + } + } + if st := b.Status(); len(st.Leases) != 1 || st.Leases[0].State != "active" { + t.Fatalf("a refused stop touched the pinned lease: %+v", st.Leases) + } +}