From 3ff2a9340aa8f9458f66befdb03c045d823372f4 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 14:15:58 +0400 Subject: [PATCH] phraser: gate every llm.Client call on the swap drain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The drain counted only the phrasing paths in internal/phraser. The router, the replier, the mail extractor and the memory evaluator reach llama-server through llm.Client, so quiesce could report zero requests in flight while the router was mid-generation, and the old server was killed under it. The turn then finished on the new model, which is the split turn the swap exists to prevent. llm.Client now enters an optional Gate before every completion and LLMPhraser implements it, so one counter covers every holder of the base URL. A total failure also reported itself as a rollback. Swap set RolledBack on the path where the rollback failed too, so the page rendered "rolled back to — she is still answering, with the old model" over an empty model name and a daemon with no model at all. The total failure has its own flag now, LiveModel stops naming a gguf that is not loaded, and the log says another attempt can recover without a restart, which is true. The swap also ran on the connection every other page shares. ipc.Client holds its mutex for a whole roundtrip with no read deadline on either side, so a load froze /dash, /history and /notifications for minutes. mavweb dials a second connection for /models alone. POST /models joins the route table, and the load settings no longer come off a form that renders no input for them. Found in review of #68. --- cmd/mavend/modelswap.go | 6 ++ cmd/mavweb/main.go | 14 +++- cmd/mavweb/models.go | 30 +++++-- cmd/mavweb/models_test.go | 60 +++++++++++++- internal/ipc/api.go | 6 ++ internal/llm/client.go | 55 +++++++++++-- internal/phraser/swap.go | 31 ++++++- internal/phraser/swap_gate_test.go | 127 +++++++++++++++++++++++++++++ 8 files changed, 312 insertions(+), 17 deletions(-) create mode 100644 internal/phraser/swap_gate_test.go diff --git a/cmd/mavend/modelswap.go b/cmd/mavend/modelswap.go index fae8cb7..be5bfef 100644 --- a/cmd/mavend/modelswap.go +++ b/cmd/mavend/modelswap.go @@ -58,6 +58,7 @@ func wireModelSwap(srv *ipc.Server, phr phraser.Phraser, cfg *config.Config) { ModelPath: res.ModelPath, BaseURL: res.BaseURL, RolledBack: res.RolledBack, + NoBackend: res.NoBackend, TookMs: res.Took.Milliseconds(), } if err != nil { @@ -106,8 +107,13 @@ func wireModelSwap(srv *ipc.Server, phr phraser.Phraser, cfg *config.Config) { // the port of a server that no longer exists, and the daemon would degrade to // the classifier permanently after the first swap. The client is re-pointed, not // rebuilt, so nothing that holds it has to know a swap happened. +// SetGate is the other half, and on the deploy shape it is the load-bearing one: +// llama-server is relaunched on the same fixed port, so SetBaseURL is usually a +// no-op, while the gate is what makes the swap's drain count these callers at +// all. Without it a swap can kill the server mid-routing-decision. func llmClientFor(lp *phraser.LLMPhraser, timeout time.Duration) *llm.Client { c := llm.New(lp.BaseURL(), timeout) lp.OnSwap(func(base string) { c.SetBaseURL(base) }) + c.SetGate(lp) return c } diff --git a/cmd/mavweb/main.go b/cmd/mavweb/main.go index 17e49af..c9f03e7 100644 --- a/cmd/mavweb/main.go +++ b/cmd/mavweb/main.go @@ -363,6 +363,11 @@ func main() { flag.Parse() var core ipc.CoreAPI + // swapConn — a second connection, for /models and nothing else. A model swap + // is a multi-minute IPC call and ipc.Client serialises everything on one + // mutex, so sharing the connection would freeze every other page for the + // length of the load. See handleModels. + var swapConn modelController if *coreSock != "" { c, err := ipc.DialWait(*coreSock, 60*time.Second) if err != nil { @@ -370,6 +375,12 @@ func main() { } defer c.Close() core = c + if sc, err := ipc.Dial(*coreSock); err != nil { + log.Printf("models: second core connection failed (%v) — /models will share the main one and a swap will block the other pages", err) + } else { + defer sc.Close() + swapConn = sc + } } mux := http.NewServeMux() @@ -499,13 +510,14 @@ func main() { // /tools, and for a comparable reason: which model is loaded decides how every // utterance is routed and how every reply is worded. GET is read-only. mux.HandleFunc("/models", func(w http.ResponseWriter, r *http.Request) { - handleModels(w, r, core, stepUpSession, *requireStepUp) + handleModels(w, r, core, swapConn, stepUpSession, *requireStepUp) }) // State-changing routes on this server, and their gate (Vikunja #317): // // POST /tools step-up — defines argv that internal/tool executes // POST /routines step-up — accepting schedules recurring firing + // POST /models step-up — replaces the model that routes and phrases // POST /api/revert step-up — voids the latest fact for a key // POST /api/chat step-up — reaches the router, LLM and the act path // POST /api/signal none — appends a presence fact, no argv, no act diff --git a/cmd/mavweb/models.go b/cmd/mavweb/models.go index 11c4d41..28b598a 100644 --- a/cmd/mavweb/models.go +++ b/cmd/mavweb/models.go @@ -36,6 +36,7 @@ var modelsTmpl = template.Must(template.New("models").Funcs(shellFuncs()).Parse( const modelsHTML = `{{template "shellTop" "models"}}

Resident model

swapping requires step-up — assert a passkey first. The old model is unloaded before the new one is loaded (one model fits the iGPU at a time), so turns during the load are refused and fall back to the classifier.

+

a swap is not remembered. Nothing writes it down, so the next restart of the daemon — including the one mavupdate does — comes back on phraser.model_path from the config. Make it stick by editing that.

{{if .Msg}}
{{.Msg}}
{{end}} {{if .Err}}
{{.Err}}
{{end}} {{if .Off}} @@ -80,12 +81,22 @@ type modelsPage struct { // A failed swap is reported as a failure with the model that is still serving // named, because that is the state the operator needs: the daemon rolled back // and is answering turns, it just is not answering them with what he asked for. -func handleModels(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI, session *webauthn.PasskeySession, requireStepUp bool) { +// swapConn, when non-nil, is a SECOND connection to the same core, used for +// nothing but this page. ipc.Client holds its mutex for a whole roundtrip and +// neither side sets a read deadline, so a swap on the shared connection blocks +// /dash, /history, /notifications and everything else for as long as the load +// takes: a 90s drain plus a 60s launch plus a 30s probe, doubled if it rolls +// back. No browser timeout frees them, because the server side keeps reading +// the reply. On its own connection the swap only blocks the swap. +func handleModels(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI, swapConn modelController, session *webauthn.PasskeySession, requireStepUp bool) { if core == nil { http.Error(w, "models disabled (no -core)", http.StatusServiceUnavailable) return } - mc, ok := core.(modelController) + mc, ok := swapConn, swapConn != nil + if !ok { + mc, ok = core.(modelController) + } if !ok { http.Error(w, "models unavailable: core connection does not support model swap", http.StatusServiceUnavailable) return @@ -103,11 +114,13 @@ func handleModels(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI, sess http.Error(w, "model_path required", http.StatusBadRequest) return } - req := ipc.SwapModelReq{ModelPath: path} - if v, err := strconv.Atoi(r.FormValue("n_ctx")); err == nil { - req.NCtx = v - } - res, err := mc.SwapModel(ctx, req) + // Only the path comes off the form. n_ctx and n_gpu_layers are load + // settings the daemon keeps from what is live, and the resident model is + // a Thinking variant whose 4096-token window is sized for reasoning + // tokens (CLAUDE.md). A field nothing renders, that a hand-crafted POST + // could use to shrink the window under the router, is not worth having. + // Changing them is a config edit and a restart. + res, err := mc.SwapModel(ctx, ipc.SwapModelReq{ModelPath: path}) switch { case err == nil: page.Msg = "loaded " + res.Model + " (" + strconv.FormatInt(res.TookMs, 10) + "ms)" @@ -118,6 +131,9 @@ func handleModels(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI, sess case errors.Is(err, ipc.ErrUnknownMethod): http.Error(w, "swap not configured on this core", http.StatusServiceUnavailable) return + case res.NoBackend: + page.Err = "swap failed AND the rollback failed — no model is loaded. She is answering from templates and routing on the classifier. Try loading a model again; a restart is not needed." + log.Printf("models: swap to %s failed and the rollback failed, no model loaded: %v", path, err) case res.RolledBack: page.Err = "swap failed, rolled back to " + res.Model + " — she is still answering, with the old model" log.Printf("models: swap to %s failed, rolled back: %v", path, err) diff --git a/cmd/mavweb/models_test.go b/cmd/mavweb/models_test.go index 1115e9e..7810b5f 100644 --- a/cmd/mavweb/models_test.go +++ b/cmd/mavweb/models_test.go @@ -36,7 +36,7 @@ func (f *fakeModelCore) SwapModel(ctx context.Context, req ipc.SwapModelReq) (ip func modelsGET(t *testing.T, core ipc.CoreAPI) *httptest.ResponseRecorder { t.Helper() w := httptest.NewRecorder() - handleModels(w, httptest.NewRequest(http.MethodGet, "/models", nil), core, nil, false) + handleModels(w, httptest.NewRequest(http.MethodGet, "/models", nil), core, nil, nil, false) return w } @@ -45,7 +45,7 @@ func modelsPOST(t *testing.T, core ipc.CoreAPI, session *webauthn.PasskeySession r := httptest.NewRequest(http.MethodPost, "/models", strings.NewReader("model_path="+path)) r.Header.Set("Content-Type", "application/x-www-form-urlencoded") w := httptest.NewRecorder() - handleModels(w, r, core, session, requireStepUp) + handleModels(w, r, core, nil, session, requireStepUp) return w } @@ -148,3 +148,59 @@ func TestModels_CoreWithoutTheMethodsIs503(t *testing.T) { type errBrokenModel struct{} func (errBrokenModel) Error() string { return "llm: server did not start" } + +func TestModels_TotalFailureDoesNotSaySheIsStillAnswering(t *testing.T) { + // The load failed and so did the rollback: nothing is loaded. The page used + // to branch on RolledBack first and render "rolled back to — she is still + // answering, with the old model" over an empty model name. + core := &fakeModelCore{ + swapResp: ipc.SwapModelResp{NoBackend: true}, + swapErr: errBrokenModel{}, + status: ipc.ModelStatusResp{Model: "unknown"}, + } + w := modelsPOST(t, core, nil, false, "/m/cpt.gguf") + if w.Code != http.StatusOK { + t.Fatalf("POST /models after a total failure = %d; want 200 with the failure rendered", w.Code) + } + body := w.Body.String() + if strings.Contains(body, "still answering") { + t.Errorf("the page claims she is still answering while no model is loaded:\n%s", body) + } + if !strings.Contains(body, "no model is loaded") { + t.Errorf("the page does not name the state the operator is in:\n%s", body) + } +} + +func TestModels_POSTIgnoresLoadSettingsOffTheForm(t *testing.T) { + // n_ctx was read off a form that renders no such input, so only a + // hand-crafted POST could set it. The resident model is a Thinking variant + // whose window is sized for reasoning tokens; shrinking it from the wire is + // not a capability this page offers. + core := &fakeModelCore{swapResp: ipc.SwapModelResp{Model: "qwen3-cpt"}} + r := httptest.NewRequest(http.MethodPost, "/models", strings.NewReader("model_path=/m/cpt.gguf&n_ctx=512&n_gpu_layers=0")) + r.Header.Set("Content-Type", "application/x-www-form-urlencoded") + w := httptest.NewRecorder() + handleModels(w, r, core, nil, nil, false) + if len(core.swapped) != 1 { + t.Fatalf("SwapModel calls = %v; want one", core.swapped) + } + if got := core.swapped[0]; got.NCtx != 0 || got.NGpuLayers != 0 { + t.Errorf("swap request = %+v; want the load settings left to the daemon", got) + } +} + +func TestModels_SwapUsesItsOwnConnection(t *testing.T) { + // A swap is a multi-minute IPC call and ipc.Client serialises everything on + // one mutex, so it must not run on the connection every other page shares. + shared := &fakeModelCore{status: ipc.ModelStatusResp{Model: "qwen3"}} + swapConn := &fakeModelCore{swapResp: ipc.SwapModelResp{Model: "qwen3-cpt"}} + r := httptest.NewRequest(http.MethodPost, "/models", strings.NewReader("model_path=/m/cpt.gguf")) + r.Header.Set("Content-Type", "application/x-www-form-urlencoded") + handleModels(httptest.NewRecorder(), r, shared, swapConn, nil, false) + if len(shared.swapped) != 0 { + t.Errorf("the swap went out on the shared connection: %v", shared.swapped) + } + if len(swapConn.swapped) != 1 { + t.Errorf("the swap did not use the dedicated connection: %v", swapConn.swapped) + } +} diff --git a/internal/ipc/api.go b/internal/ipc/api.go index db71b6c..711c2aa 100644 --- a/internal/ipc/api.go +++ b/internal/ipc/api.go @@ -370,11 +370,17 @@ type SwapModelReq struct { // RolledBack is true when the requested model failed to load or would not answer // and the previous one was put back. In that case the call also returns an error // — the swap did not happen — and Model names the model still serving. +// +// NoBackend is the other failure and it is not a milder one: the rollback failed +// too, no model is loaded, and every phrasing path is on its template fallback +// with routing on the classifier. It is a separate field from RolledBack because +// the two need opposite words on the page. type SwapModelResp struct { Model string `json:"model"` ModelPath string `json:"model_path"` BaseURL string `json:"base_url"` RolledBack bool `json:"rolled_back,omitempty"` + NoBackend bool `json:"no_backend,omitempty"` TookMs int64 `json:"took_ms"` } diff --git a/internal/llm/client.go b/internal/llm/client.go index a43748f..86ceae6 100644 --- a/internal/llm/client.go +++ b/internal/llm/client.go @@ -14,17 +14,57 @@ import ( "time" ) +// Gate — admission control for a completion. Enter blocks or refuses while the +// resident model is being swapped, and the returned release says the request is +// done. The phraser implements it: a swap kills the running llama-server, so +// every holder of a base URL has to be counted before the kill, not just the +// phrasing paths. +// +// Without this the drain saw only the phraser's own calls. The LLM router, the +// replier, the mail extractor and the memory evaluator all reach llama-server +// through this client, so a swap could report zero requests in flight and kill +// the server out from under a routing decision. The turn then finished on the +// new model, which is the "half of one model and half of another" the swap is +// supposed to make impossible. +type Gate interface { + Enter() (release func(), err error) +} + type Client struct { - // mu guards base only. The base URL changes when the daemon swaps the - // resident model (Vikunja #250): llama-server is relaunched on a fresh - // port, and every holder of this client — the LLM router, the replier, the - // mail extractor — must follow without being rebuilt. One mutexed field is - // the whole mechanism; a swap re-points the client, it does not replace it. + // mu guards base and gate. The base URL can change when the daemon swaps the + // resident model (Vikunja #250) and every holder of this client — the LLM + // router, the replier, the mail extractor — must follow without being + // rebuilt. A swap re-points the client, it does not replace it. + // + // On the deploy shape the new server binds the same fixed port the killed + // one released (startLlamaProc passes the port out of phraser.listen), so + // SetBaseURL is normally a no-op and the gate is the part doing the work. + // The re-pointing stays because nothing guarantees the port: a phraser + // listening on :0, or a future swap that moves the server, changes the base. mu sync.RWMutex base string + gate Gate http *http.Client } +// SetGate installs the admission gate. Nil (the default, and what the eval +// harness and the tests use) means no gating. +func (c *Client) SetGate(g Gate) { + c.mu.Lock() + c.gate = g + c.mu.Unlock() +} + +func (c *Client) enter() (func(), error) { + c.mu.RLock() + g := c.gate + c.mu.RUnlock() + if g == nil { + return func() {}, nil + } + return g.Enter() +} + func New(baseURL string, timeout time.Duration) *Client { return &Client{base: baseURL, http: &http.Client{Timeout: timeout}} } @@ -81,6 +121,11 @@ type resp struct { } func (c *Client) Complete(ctx context.Context, r Req) (string, error) { + release, err := c.enter() + if err != nil { + return "", err + } + defer release() b, _ := json.Marshal(body{ Messages: []msg{{Role: "system", Content: r.System}, {Role: "user", Content: r.User}}, MaxTokens: r.MaxTokens, diff --git a/internal/phraser/swap.go b/internal/phraser/swap.go index 22d9aeb..bf4d679 100644 --- a/internal/phraser/swap.go +++ b/internal/phraser/swap.go @@ -28,6 +28,10 @@ import ( // fallback (templates, "вот что я нашла", the classifier for routing), so a // fast refusal degrades one turn instead of hanging it for the length of a // model load. No turn ever gets half of one model and half of another. +// "Every path" means every path: the router, the replier, the mail +// extractor and the memory evaluator do not call acquire, they call +// llm.Client.Complete, so LLMPhraser implements llm.Gate and the client +// enters through the same counter. // // 3. A failed load rolls back to the model that was working. The new server is // probed (it must say which model it loaded) before it is published; if the @@ -68,11 +72,16 @@ type SwapSpec struct { // SwapResult — what happened. Model is the identity the NEW server reported, so // it is evidence rather than an echo of the request: if the file at ModelPath is // not what the operator thought it was, this is where that shows up. +// RolledBack is true only when a model is serving again. NoBackend is the other +// failure, and it is the worse one: the rollback failed too and nothing is +// loaded. They are separate flags because the operator surface reads them, and +// "rolled back" spelled over a dead daemon reads as reassurance. type SwapResult struct { Model string BaseURL string ModelPath string RolledBack bool + NoBackend bool Took time.Duration } @@ -140,6 +149,19 @@ func (p *LLMPhraser) acquire() (string, func(), error) { }, nil } +// Enter implements llm.Gate so every holder of an *llm.Client is drained by a +// swap, not only the phrasing paths in this package. +// +// The router, the replier, the mail extractor and the memory evaluator do not +// call acquire; they call llm.Client.Complete. Before this existed quiesce could +// see zero requests in flight while the router was mid-generation and kill the +// server under it. A refusal here is the same ErrSwapping the phrasing paths +// get, and every caller of Complete already falls back. +func (p *LLMPhraser) Enter() (func(), error) { + _, release, err := p.acquire() + return release, err +} + // Swap loads another model in place of the live one. See the file comment for // the properties it guarantees. Returns the new model's reported identity, or // an error plus RolledBack=true when the old model was put back. @@ -195,8 +217,13 @@ func (p *LLMPhraser) Swap(ctx context.Context, spec SwapSpec) (SwapResult, error log.Printf("phraser: swap to %s FAILED (%v) — rolling back to %s", newLive.ModelPath, err, oldLive.ModelPath) rb, rbErr := p.loadAndProbe(ctx, oldLive) if rbErr != nil { - log.Printf("phraser: ROLLBACK to %s ALSO FAILED (%v) — no model is loaded, every phrasing path is on its fallback and routing is on the classifier until the daemon is restarted", oldLive.ModelPath, rbErr) - return SwapResult{RolledBack: true, Took: time.Since(started)}, + // Nothing is loaded, so LiveModel must stop naming a gguf: the page + // would show a file next to an unknown model and read as half-working. + p.mu.Lock() + p.live = liveModel{} + p.mu.Unlock() + log.Printf("phraser: ROLLBACK to %s ALSO FAILED (%v) — no model is loaded, every phrasing path is on its fallback and routing is on the classifier. Swap is still wired, so another attempt can recover without restarting the daemon", oldLive.ModelPath, rbErr) + return SwapResult{NoBackend: true, Took: time.Since(started)}, fmt.Errorf("phraser: swap failed (%w) and rollback failed too: %v", err, rbErr) } p.publish(rb, oldLive) diff --git a/internal/phraser/swap_gate_test.go b/internal/phraser/swap_gate_test.go new file mode 100644 index 0000000..c2deda0 --- /dev/null +++ b/internal/phraser/swap_gate_test.go @@ -0,0 +1,127 @@ +package phraser + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/kami/maven/internal/llm" +) + +// The drain has to cover every holder of the base URL, not only the phrasing +// paths in this package. The LLM router, the replier, the mail extractor and the +// memory evaluator all reach llama-server through llm.Client, and a swap that +// does not count them kills the server mid-turn. + +// blockingLLM — a completion endpoint that does not answer until the test says +// so. It stands in for a router call that is generating when the swap arrives. +func blockingLLM(t *testing.T, release <-chan struct{}) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-release + w.Write([]byte(`{"choices":[{"message":{"content":"ok"}}]}`)) + })) + t.Cleanup(srv.Close) + return srv +} + +func (p *LLMPhraser) inflightCount() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.inflight +} + +func TestSwap_WaitsForARouterCallThatWentThroughLLMClient(t *testing.T) { + fl := &fakeFleet{models: map[string]string{"/m/old.gguf": "old", "/m/new.gguf": "new"}} + p := newSwapPhraser(t, fl, "/m/old.gguf") + + release := make(chan struct{}) + c := llm.New(blockingLLM(t, release).URL, 5*time.Second) + c.SetGate(p) + + completed := make(chan error, 1) + go func() { + _, err := c.Complete(context.Background(), llm.Req{System: "s", User: "u"}) + completed <- err + }() + deadline := time.Now().Add(2 * time.Second) + for p.inflightCount() == 0 { + if time.Now().After(deadline) { + t.Fatal("the llm.Client request never registered with the phraser gate") + } + time.Sleep(5 * time.Millisecond) + } + + swapped := make(chan error, 1) + go func() { _, e := p.Swap(context.Background(), SwapSpec{ModelPath: "/m/new.gguf"}); swapped <- e }() + + select { + case e := <-swapped: + t.Fatalf("the swap finished while a router call was still generating (%v); the old server was killed under it", e) + case <-time.After(200 * time.Millisecond): + } + + close(release) + if e := <-completed; e != nil { + t.Fatalf("the in-flight call did not finish on the old model: %v", e) + } + if e := <-swapped; e != nil { + t.Fatalf("Swap after the drain: %v", e) + } +} + +func TestSwap_RefusesARouterCallThatArrivesMidSwap(t *testing.T) { + fl := &fakeFleet{models: map[string]string{"/m/old.gguf": "old", "/m/new.gguf": "new"}} + p := newSwapPhraser(t, fl, "/m/old.gguf") + + // An open server: the refusal has to come from the gate, not from a stall. + open := make(chan struct{}) + close(open) + c := llm.New(blockingLLM(t, open).URL, 5*time.Second) + c.SetGate(p) + + // Hold the door shut the way quiesce does. + _, held, err := p.acquire() + if err != nil { + t.Fatal(err) + } + defer held() + go p.Swap(context.Background(), SwapSpec{ModelPath: "/m/new.gguf"}) + + deadline := time.Now().Add(2 * time.Second) + for { + _, err := c.Complete(context.Background(), llm.Req{User: "u"}) + if errors.Is(err, ErrSwapping) { + return + } + if time.Now().After(deadline) { + t.Fatalf("a router call during a swap was not refused (last error: %v)", err) + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestSwap_TotalFailureIsNotReportedAsARollback(t *testing.T) { + fl := &fakeFleet{models: map[string]string{"/m/old.gguf": "old"}} + p := newSwapPhraser(t, fl, "/m/old.gguf") + fl.mu.Lock() + delete(fl.models, "/m/old.gguf") + fl.mu.Unlock() + + res, err := p.Swap(context.Background(), SwapSpec{ModelPath: "/m/broken.gguf"}) + if err == nil { + t.Fatal("Swap returned nil when both the load and the rollback failed") + } + if res.RolledBack { + t.Error("a total failure set RolledBack; the page then says she is still answering with the old model, and she is not answering at all") + } + if !res.NoBackend { + t.Error("a total failure did not set NoBackend, so nothing distinguishes it from a rolled-back swap") + } + if path, _, _ := p.LiveModel(); path != "" { + t.Errorf("LiveModel = %q after a total failure; nothing is loaded, and naming a gguf makes the page read as half-working", path) + } +}