From 2da2efc1fdda15ff6bb08ffb3e83664f8493ff21 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 18:33:31 +0000 Subject: [PATCH] =?UTF-8?q?Lot=2016=20:=20pilotage=20du=20moteur=20tenu=20?= =?UTF-8?q?par=20la=20fa=C3=A7ade=20(LOKI=5FENGINE=3Drust)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit La façade sert /api/start, /api/stop, /api/restart, /api/vram/unload et /api/vram/reload quand LOKI_ENGINE=rust (expérimental ; le Go reste le défaut). Elle lance elle-même « loki-rs engine », le superviseur que le Go appelait déjà en conteneur, avec l'environnement qu'il lui passait et de quoi joindre l'API interne du Go ; sa dernière ligne d'erreur devient l'erreur rendue. Le contrôle d'avant démarrage (preflightEngine : preset externe, Strata, BIN, MODEL, tranches) et le bilan de VRAM du déchargement (gpuSettle) sont portés. La façade ne tient ce lot qu'en conteneur : sous systemd ou avec LOKI_ENGINE_SUPERVISOR=go, loki-rs ne passe pas la variable au Go. Restent au Go derrière son API interne (facade_engine.go) : le verrou de l'optimiseur avant tout démarrage (/internal/v1/service/prestart), la réponse en cours quand il tient la conversation (/internal/v1/chat/generating) et l'arrêt de la dictée quand il la tient (/internal/v1/dictate/stop). Parité : parity_engine_test.go tire du Go dix bilans de VRAM, vingt-six installations passées au préflight, les messages du superviseur et dix réponses de route, rejoués par tests/enginectl.rs ; démarrage, arrêt, redémarrage, déchargement chiffré et rechargement vérifiés de bout en bout avec un faux superviseur et un faux nvidia-smi ; TestFacadeMoteur côté Go. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv --- README.md | 1 + ...0-15-rust-migration-lot16-moteur-design.md | 90 +++ internal/loki/facade_chat.go | 6 +- internal/loki/facade_chat_test.go | 75 +++ internal/loki/facade_engine.go | 90 +++ internal/loki/parity_engine_test.go | 292 +++++++++ internal/loki/web_internal.go | 6 + rust/loki-rs/src/api.rs | 3 + rust/loki-rs/src/enginectl.rs | 561 ++++++++++++++++++ rust/loki-rs/src/gguf.rs | 10 + rust/loki-rs/src/lib.rs | 1 + rust/loki-rs/src/main.rs | 37 ++ rust/loki-rs/src/proxy.rs | 9 + rust/loki-rs/src/sysinfo.rs | 2 +- rust/loki-rs/tests/common/fakego.rs | 21 + rust/loki-rs/tests/enginectl.rs | 470 +++++++++++++++ rust/loki-rs/tests/lot2.rs | 1 + rust/loki-rs/tests/parity_engine/engine.json | 551 +++++++++++++++++ rust/loki-rs/tests/proxy.rs | 1 + 19 files changed, 2225 insertions(+), 2 deletions(-) create mode 100644 docs/superpowers/specs/2026-10-15-rust-migration-lot16-moteur-design.md create mode 100644 internal/loki/facade_engine.go create mode 100644 internal/loki/parity_engine_test.go create mode 100644 rust/loki-rs/src/enginectl.rs create mode 100644 rust/loki-rs/tests/enginectl.rs create mode 100644 rust/loki-rs/tests/parity_engine/engine.json diff --git a/README.md b/README.md index adc92da..1cbf23a 100644 --- a/README.md +++ b/README.md @@ -236,6 +236,7 @@ redémarrages (volume `/data`). Variables d'environnement du conteneur : | `LOKI_MEM` | `rust` : la mémoire (éditeur des pages, index, déplacement entre projets, trackers, mode mémoire, interrupteurs du mode agent) est tenue par la façade Rust ; le chiffrement reste au Go (expérimental, migration en cours) | Go | | `LOKI_SERVE` | `rust` : la ligne de commande de llama-server est composée par la façade Rust (`loki-rs engine serve`) au lieu de `loki serve` ; Strata et l'optimiseur restent au Go (expérimental, migration en cours) | Go | | `LOKI_PRESETS` | `rust` : les presets (liste, éditeur et aperçus, suppression, placement auto, presets externes, bascule) sont tenus par la façade Rust ; l'application au moteur reste au Go (expérimental, migration en cours) | Go | +| `LOKI_ENGINE` | `rust` : le pilotage du moteur (démarrage, arrêt, redémarrage, déchargement et rechargement de la VRAM) est tenu par la façade Rust, en conteneur avec son superviseur ; le verrou de l'optimiseur reste au Go (expérimental, migration en cours) | Go | **Accès par nom de domaine.** Sans clé de pilotage, l'API n'accepte que les hôtes locaux (IP, `localhost`, nom sans point, `.local`/`.lan`…) et refuse diff --git a/docs/superpowers/specs/2026-10-15-rust-migration-lot16-moteur-design.md b/docs/superpowers/specs/2026-10-15-rust-migration-lot16-moteur-design.md new file mode 100644 index 0000000..dc24284 --- /dev/null +++ b/docs/superpowers/specs/2026-10-15-rust-migration-lot16-moteur-design.md @@ -0,0 +1,90 @@ +# Lot 16 — pilotage du moteur tenu par la façade + +## Contexte + +Le Go pilotait le moteur depuis l'interface : +- `svcHandler` (`web_api.go`) pour `/api/start`, `/api/stop` et + `/api/restart` ; +- `handleVramUnload` et `handleVramReload` (`web_vram.go`) pour rendre la + carte graphique à la machine, puis la reprendre ; +- `serviceAction` (`sys_service.go`) : le verrou de l'optimiseur, puis, en + conteneur, le superviseur Rust `loki-rs engine ` (lot 3, + `sys_service_rust_linux.go`) ; +- `preflightEngine` : ce sans quoi le moteur ne peut pas démarrer, dit avant + de le lancer. + +Lot 16 : la façade tient ces routes quand `LOKI_ENGINE=rust` +(`enginectl.rs`, expérimental). Sans la variable, rien ne change. + +| Route | Rôle | +|---|---| +| `/api/start`, `/api/stop`, `/api/restart` | pilotage du moteur ; réponse `{ok, out}` comme le Go | +| `POST /api/vram/unload` | arrête le moteur et la dictée, rapporte la VRAM rendue ; refusé pendant une réponse en cours sauf `{force: true}` | +| `POST /api/vram/reload` | relance le moteur après le contrôle d'avant démarrage ; `{already: true}` s'il tourne | + +La façade lance elle-même `loki-rs engine `, avec l'environnement que +le Go lui passait (`LOKI_GO_BIN`, `LOKI_HOME`, `LOKI_SERVICE`) et de quoi +joindre l'API interne du Go pour `loki-rs engine serve` (`LOKI_GO_ADDR`, +`LOKI_INTERNAL_TOKEN`). Sa dernière ligne d'erreur devient l'erreur rendue, +sans le préfixe du binaire. Un preset externe ne lance rien ; un redémarrage +arrête alors le moteur local. + +Le bilan de VRAM suit `gpuSettle` : +- une lecture ratée est sautée, jamais comptée « 0 Mo » ; +- un palier ne conclut qu'une fois la baisse commencée ; +- les jauges du moniteur sont invalidées après coup. + +## Quand la façade ne tient pas ce lot + +La façade ne prend ce lot qu'en conteneur, avec son superviseur. Sous +systemd, ou avec `LOKI_ENGINE_SUPERVISOR=go`, `loki-rs` ne passe pas +`LOKI_ENGINE` au Go, qui garde les routes (décidé au démarrage, raison dans +le journal). + +## Restent au Go + +Le Go expose trois routes internes (`facade_engine.go`) : +- `POST /internal/v1/service/prestart` : avant tout démarrage, refus pendant + une optimisation, puis ménage d'un verrou périmé (essai orphelin arrêté, + application interrompue défaite). Le verrou de l'optimiseur suit le + processus qui le tient, il reste au Go jusqu'au lot de l'optimiseur ; +- `GET /internal/v1/chat/generating` : la réponse en cours, quand le Go tient + la conversation (sinon la façade lit la sienne) ; +- `POST /internal/v1/dictate/stop` : l'arrêt de la dictée, quand le Go la + tient (sinon la façade arrête la sienne). + +`GET /internal/v1/tune/guard` (lot 15) sert le refus 409 du rechargement. + +Les autres appelants de `serviceAction` côté Go (bascule de preset, +optimiseur, mises à jour du moteur) passent toujours par le Go. Les deux voies +lancent le même superviseur, sur les mêmes fichiers PID et journal. + +## Écarts assumés + +- `engineNeedsStop` regarde aussi les états transitoires de systemd. En + conteneur, seul cas où la façade tient ce lot, il se réduit au PID vivant. +- Un superviseur tué par un signal est décrit comme par le Go (« signal: + killed ») pour les signaux courants, par son numéro sinon. + +## Parité + +`internal/loki/parity_engine_test.go` tire du Go : +- dix bilans de VRAM (palier avant la baisse, lectures ratées, VRAM en + hausse…) ; +- la dernière ligne d'erreur du superviseur et le retrait des couleurs ; +- vingt-six installations passées au contrôle d'avant démarrage (preset + externe, Strata, BIN absent ou relatif, MODEL absent, hors des dossiers, + remontant, en tranches incomplètes…) ; +- dix réponses de route qui ne lancent rien (GET refusé avec `Allow: POST`, + optimisation ou réponse en cours, moteur déjà arrêté, contrôle en échec). + +`tests/enginectl.rs` rejoue le tout par la façade. Il vérifie aussi de bout en +bout, avec un faux superviseur et un faux nvidia-smi : +- démarrage, arrêt, redémarrage, refus pendant une optimisation ; +- échec du démarrage, preset externe ; +- déchargement chiffré, arrêt en échec, VRAM en hausse ; +- rechargement. + +Une mutation de chaque règle fait échouer les tests. + +Côté Go, `TestFacadeMoteur` vérifie le relais et les trois routes internes. diff --git a/internal/loki/facade_chat.go b/internal/loki/facade_chat.go index c300872..b651fbb 100644 --- a/internal/loki/facade_chat.go +++ b/internal/loki/facade_chat.go @@ -143,7 +143,9 @@ var facadeChatRoutes = map[string]bool{ // withFacadeChat remet la requête à la façade quand elle tient la // conversation — ou les routes des modèles (facade_models.go), la dictée // (facade_dictate.go), les réglages partagés (facade_settings.go), les fichiers -// de la discussion (facade_files.go), les notifications (facade_push.go). +// de la discussion (facade_files.go), les notifications (facade_push.go), la +// mémoire (facade_mem.go), les presets (facade_presets.go), le pilotage du +// moteur (facade_engine.go). // Posé APRÈS requireWebAuth : seule une requête authentifiée ici part, avec le // jeton interne qui en tient lieu là-bas. func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc { @@ -164,6 +166,8 @@ func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc { held, unavailable = facadeMem, "mémoire indisponible : " case facadePresetsRoutes[path]: held, unavailable = facadePresets, "presets indisponibles : " + case facadeEngineRoutes[path]: + held, unavailable = facadeEngine, "pilotage du moteur indisponible : " default: return h } diff --git a/internal/loki/facade_chat_test.go b/internal/loki/facade_chat_test.go index daddc16..2a7485a 100644 --- a/internal/loki/facade_chat_test.go +++ b/internal/loki/facade_chat_test.go @@ -570,3 +570,78 @@ func TestFacadePresets(t *testing.T) { t.Fatalf("reçu par la façade : %d appels", n) } } + +// Le pilotage du moteur (LOKI_ENGINE=rust, lot 16) part à la façade, et l'API +// interne lui rend ce qui reste ici : verrou de l'optimiseur avant un +// démarrage, génération en cours, arrêt de la dictée. +func TestFacadeMoteur(t *testing.T) { + testHome(t) + f := demarrerFacade(t) + t.Setenv(facadeChatEnv, "") + local := func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "go") } + for p := range facadeEngineRoutes { + w := httptest.NewRecorder() + withFacadeChat(p, local)(w, httptest.NewRequest("POST", p, nil)) + if w.Body.String() != "go" { + t.Fatalf("%s servi ailleurs sans LOKI_ENGINE : %q", p, w.Body.String()) + } + } + t.Setenv(facadeEngineEnv, "rust") + for p := range facadeEngineRoutes { + w := httptest.NewRecorder() + withFacadeChat(p, local)(w, httptest.NewRequest("POST", p, strings.NewReader("{}"))) + if w.Body.String() != f.reply { + t.Fatalf("%s : réponse %q", p, w.Body.String()) + } + } + if n := len(f.recu()); n != len(facadeEngineRoutes) { + t.Fatalf("reçu par la façade : %d appels", n) + } + + // Avant un démarrage : le verrou, puis le ménage. + appels := 0 + old := enginePrestart + t.Cleanup(func() { enginePrestart = old }) + enginePrestart = func() error { + appels++ + if appels == 2 { + return errTuneBusy + } + return nil + } + for i, want := range []string{`{"error":""}`, `{"error":"` + errTuneBusy.Error() + `"}`} { + w := httptest.NewRecorder() + handleInternalServicePrestart(w, httptest.NewRequest("POST", "/internal/v1/service/prestart", nil)) + if got := strings.TrimSpace(w.Body.String()); w.Code != 200 || got != want { + t.Fatalf("prestart %d : %d %s", i, w.Code, got) + } + } + w := httptest.NewRecorder() + handleInternalServicePrestart(w, httptest.NewRequest("GET", "/internal/v1/service/prestart", nil)) + if w.Code != 405 || appels != 2 { + t.Fatalf("prestart en GET : %d (%d appels)", w.Code, appels) + } + + // La génération en cours de la conversation tenue ici. + for _, gen := range []bool{false, true} { + conv.mu.Lock() + conv.Generating = gen + conv.mu.Unlock() + w := httptest.NewRecorder() + handleInternalChatGenerating(w, httptest.NewRequest("GET", "/internal/v1/chat/generating", nil)) + var out struct{ Generating bool } + if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil || out.Generating != gen { + t.Fatalf("generating=%v : %s", gen, w.Body.String()) + } + } + conv.mu.Lock() + conv.Generating = false + conv.mu.Unlock() + + // L'arrêt de la dictée (rien ne tourne : immédiat). + w = httptest.NewRecorder() + handleInternalDictateStop(w, httptest.NewRequest("POST", "/internal/v1/dictate/stop", nil)) + if w.Code != 204 { + t.Fatalf("arrêt de la dictée : %d", w.Code) + } +} diff --git a/internal/loki/facade_engine.go b/internal/loki/facade_engine.go new file mode 100644 index 0000000..cc2732a --- /dev/null +++ b/internal/loki/facade_engine.go @@ -0,0 +1,90 @@ +package loki + +import ( + "net/http" + "os" + "strings" +) + +// facade_engine.go — pilotage du moteur tenu par la façade Rust +// (LOKI_ENGINE=rust, lot 16 de la migration ; expérimental, le Go reste le +// défaut) : démarrage, arrêt, redémarrage (/api/start|stop|restart) et +// déchargement / rechargement de la VRAM (/api/vram/unload|reload). +// +// La façade lance elle-même « loki-rs engine » — le superviseur que +// le Go appelait déjà en conteneur (sys_service_rust_linux.go) — et contrôle +// le moteur avant de le relancer (preflightEngine). Elle ne tient ce lot qu'en +// conteneur avec ce superviseur : sous systemd, ou avec +// LOKI_ENGINE_SUPERVISOR=go, le Go le garde (décidé par loki-rs au démarrage). +// +// Ce qui vit ici reste ici jusqu'au lot qui le portera, derrière l'API +// interne : +// +// - le verrou de l'optimiseur, consulté puis débarrassé d'un essai orphelin +// avant tout démarrage (/internal/v1/service/prestart) ; +// - la génération en cours quand le Go tient la conversation +// (/internal/v1/chat/generating) ; +// - l'arrêt de la dictée quand le Go la tient (/internal/v1/dictate/stop). + +const facadeEngineEnv = "LOKI_ENGINE" + +// facadeEngine : la façade pilote-t-elle le moteur ? +func facadeEngine() bool { + return strings.EqualFold(strings.TrimSpace(os.Getenv(facadeEngineEnv)), "rust") && + facadeAddr() != "" && internalToken() != "" +} + +// facadeEngineRoutes : les routes que la façade sert alors (enginectl.rs). +var facadeEngineRoutes = map[string]bool{ + "/api/start": true, + "/api/stop": true, + "/api/restart": true, + "/api/vram/unload": true, + "/api/vram/reload": true, +} + +// enginePrestart : ce que serviceAction fait avant un démarrage — refus +// pendant une optimisation, puis ménage d'un verrou périmé (essai orphelin +// arrêté, application interrompue défaite). Variable pour les tests. +var enginePrestart = func() error { + if err := tuneGuard(); err != nil { + return err + } + tuneReapStale() + return nil +} + +// handleInternalServicePrestart : POST → 200 {error} ("" : la voie est +// libre). +func handleInternalServicePrestart(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + sendJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "POST attendu"}) + return + } + msg := "" + if err := enginePrestart(); err != nil { + msg = err.Error() + } + sendJSON(w, http.StatusOK, map[string]any{"error": msg}) +} + +// handleInternalChatGenerating : GET → 200 {generating} — la conversation +// tenue ici écrit-elle une réponse ? +func handleInternalChatGenerating(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + sendJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "GET attendu"}) + return + } + sendJSON(w, http.StatusOK, map[string]any{"generating": conv.isGenerating()}) +} + +// handleInternalDictateStop : POST → 204 — arrête la dictée tenue ici, sans +// attendre derrière un démarrage (asrShutdownVite). +func handleInternalDictateStop(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + sendJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "POST attendu"}) + return + } + asrShutdownVite() + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/loki/parity_engine_test.go b/internal/loki/parity_engine_test.go new file mode 100644 index 0000000..3d3bf3d --- /dev/null +++ b/internal/loki/parity_engine_test.go @@ -0,0 +1,292 @@ +package loki + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" +) + +// parity_engine_test.go — journal de référence du pilotage du moteur (lot 16 +// de la migration Rust) : le bilan de VRAM après déchargement (gpuSettle), le +// contrôle d'avant démarrage (preflightEngine) sur une série d'installations, +// la dernière ligne d'erreur du superviseur (lastErrorLine) et les réponses +// des routes qui ne lancent aucun processus (GET refusé, optimisation en +// cours, génération en cours, moteur déjà arrêté, préflight en échec). Rangé +// dans rust/loki-rs/tests/parity_engine/engine.json ; tests/enginectl.rs le +// rejoue par la façade. +// +// LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteMoteur + +const parityEngineFile = "../../rust/loki-rs/tests/parity_engine/engine.json" + +type parityPreflight struct { + Name string `json:"name"` + Config map[string]string `json:"config"` // « {{HOME}} » = LOKI_HOME + Files []string `json:"files,omitempty"` // relatifs à LOKI_HOME ; « / » final = dossier + Dirs []string `json:"dirs,omitempty"` // state/model_dirs, relatifs à LOKI_HOME + Out string `json:"out"` +} + +type parityEngineStep struct { + Name string `json:"name"` + Method string `json:"method"` + Path string `json:"path"` + Body string `json:"body,omitempty"` + Tune bool `json:"tune,omitempty"` // une optimisation tient le moteur + Generating bool `json:"generating,omitempty"` // une réponse est en cours + Config map[string]string `json:"config,omitempty"` + Status int `json:"status"` + Allow string `json:"allow,omitempty"` + Out any `json:"out"` +} + +func TestPariteMoteur(t *testing.T) { + home := testHome(t) + t.Setenv("LOKI_CONTAINER", "1") + t.Setenv("LOKI_MODEL_DIRS", "") + t.Setenv("LOKI_ENGINE_BIN", "") + + var fx struct { + Settle []map[string]any `json:"settle"` + LastError map[string]string `json:"last_error"` + Plain map[string]string `json:"plain"` + Preflight []parityPreflight `json:"preflight"` + Steps []parityEngineStep `json:"steps"` + } + + // --- gpuSettle --- + seq := func(vals []int) func() (int, bool) { + i := 0 + return func() (int, bool) { + if i >= len(vals) { + return vals[len(vals)-1], true + } + v := vals[i] + i++ + if v < 0 { + return 0, false + } + return v, true + } + } + for _, c := range []struct { + before int + seq []int + }{ + {14000, []int{20000, 20000, 12000, 3000, 500, 500}}, + {14000, []int{15000, 8000, 400, 400}}, + {14000, []int{14000, -1, 600, 600}}, + {14000, []int{-1, -1, -1, -1, -1, -1, -1, -1}}, + {14000, []int{14000, 14000, 14000, 14000, 14000, 14000, 14000, 14000}}, + {0, []int{0, 0, 0}}, + {9000, []int{9000, 8999, 8999}}, + {9000, []int{-1, 9000, -1, 9000, 100, -1, 100}}, + {5000, []int{6000, 7000, 7000, 7000, 7000, 7000, 7000, 7000}}, + {5000, []int{4000, 3000, 2000, 1000, 900, 800, 700, 600}}, + } { + fx.Settle = append(fx.Settle, map[string]any{"before": c.before, "seq": c.seq, + "out": gpuSettle(c.before, seq(c.seq), 0)}) + } + + // --- lastErrorLine / plainErr --- + fx.LastError = map[string]string{} + for _, s := range []string{ + "[loki-rs engine] service loki-engine non démarré\n", + "ligne 1\n[loki-rs engine] moteur absent \n\n", + "pas de préfixe", + "", + "[ok] démarré\n[attention] 2 processus du moteur toujours là\n", + } { + fx.LastError[s] = lastErrorLine(s) + } + fx.Plain = map[string]string{} + for _, s := range []string{ + "BIN non défini — installe un moteur : \x1b[1mloki llamacpp install\x1b[0m", + "\x1b[1;36mcouleur\x1b[0m et \x1b[32mvert\x1b[0m", + "rien à retirer", + "\x1b[mvide\x1b[0;1;2m", + } { + fx.Plain[s] = ansiRe.ReplaceAllString(s, "") + } + + // --- preflightEngine --- + writeFiles := func(files []string) { + for _, f := range files { + p := filepath.Join(home, f) + if strings.HasSuffix(f, "/") { + if err := os.MkdirAll(p, 0o755); err != nil { + t.Fatal(err) + } + continue + } + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, []byte("x"), 0o755); err != nil { + t.Fatal(err) + } + } + } + reset := func() { + for _, d := range []string{"bin", "models", "strata", "ailleurs", "disque"} { + _ = os.RemoveAll(filepath.Join(home, d)) + } + cfg := ReadConfig() + for k := range cfg { + _ = SetConfigKey(k, "") + } + _ = putJSON(bkState, "model_dirs", []string{}) + } + setCfg := func(cfg map[string]string) { + for k, v := range cfg { + if err := SetConfigKey(k, strings.ReplaceAll(v, "{{HOME}}", home)); err != nil { + t.Fatal(err) + } + } + } + bin := "bin/llama-server" + ok := []string{bin, "models/m.gguf"} + for _, c := range []parityPreflight{ + {Name: "externe", Config: map[string]string{"EXTERNAL": "1"}}, + {Name: "externe sans BIN ni MODEL", Config: map[string]string{"EXTERNAL": " 1 ", "BIN": ""}}, + {Name: "strata sans config", Config: map[string]string{"ENGINE": "strata"}}, + {Name: "strata config absente", Config: map[string]string{"ENGINE": "Strata", "STRATA_CONFIG": "{{HOME}}/strata/m/config.json"}}, + {Name: "strata sans venv", Config: map[string]string{"ENGINE": "strata", "STRATA_CONFIG": "{{HOME}}/strata/m/config.json"}, + Files: []string{"strata/m/config.json"}}, + {Name: "strata prêt", Config: map[string]string{"ENGINE": "moe", "STRATA_CONFIG": "{{HOME}}/strata/m/config.json"}, + Files: []string{"strata/m/config.json", "strata/m/.venv/bin/python"}}, + {Name: "BIN vide", Config: map[string]string{"MODEL": "m.gguf"}}, + {Name: "BIN espaces", Config: map[string]string{"BIN": " ", "MODEL": "m.gguf"}}, + {Name: "BIN relatif absent", Config: map[string]string{"BIN": "bin/absent", "MODEL": "m.gguf"}}, + {Name: "BIN absolu absent", Config: map[string]string{"BIN": "{{HOME}}/ailleurs/llama-server", "MODEL": "m.gguf"}}, + {Name: "BIN dossier", Config: map[string]string{"BIN": "bin/", "MODEL": "m.gguf"}, Files: []string{"bin/"}}, + {Name: "MODEL vide", Config: map[string]string{"BIN": bin}, Files: []string{bin}}, + {Name: "MODEL absent", Config: map[string]string{"BIN": bin, "MODEL": "absent.gguf"}, Files: []string{bin}}, + {Name: "MODEL relatif", Config: map[string]string{"BIN": "{{HOME}}/" + bin, "MODEL": "m.gguf"}, Files: ok}, + {Name: "MODEL remonte", Config: map[string]string{"BIN": bin, "MODEL": "../m.gguf"}, Files: ok}, + {Name: "MODEL sous-dossier", Config: map[string]string{"BIN": bin, "MODEL": "sub/m.gguf"}, Files: ok}, + {Name: "MODEL absolu hors dossiers", Config: map[string]string{"BIN": bin, "MODEL": "{{HOME}}/ailleurs/x.gguf"}, + Files: []string{bin, "ailleurs/x.gguf"}}, + {Name: "MODEL absolu pas gguf", Config: map[string]string{"BIN": bin, "MODEL": "{{HOME}}/ailleurs/x.bin"}, + Files: []string{bin, "ailleurs/x.bin"}}, + {Name: "MODEL absolu absent", Config: map[string]string{"BIN": bin, "MODEL": "{{HOME}}/ailleurs/y.gguf"}, + Files: []string{bin}}, + {Name: "MODEL dossier déclaré", Config: map[string]string{"BIN": bin, "MODEL": "d.gguf"}, + Files: []string{bin, "disque/d.gguf"}, Dirs: []string{"disque"}}, + {Name: "MODEL chemin dans dossier déclaré", Config: map[string]string{"BIN": bin, "MODEL": "{{HOME}}/disque/d.gguf"}, + Files: []string{bin, "disque/d.gguf"}, Dirs: []string{"disque"}}, + {Name: "tranches manquantes", Config: map[string]string{"BIN": bin, "MODEL": "big-00001-of-00003.gguf"}, + Files: []string{bin, "models/big-00001-of-00003.gguf", "models/big-00003-of-00003.gguf"}}, + {Name: "tranches dossier", Config: map[string]string{"BIN": bin, "MODEL": "big-00001-of-00002.gguf"}, + Files: []string{bin, "models/big-00001-of-00002.gguf", "models/big-00002-of-00002.gguf/"}}, + {Name: "tranches complètes", Config: map[string]string{"BIN": bin, "MODEL": "big-00001-of-00002.gguf"}, + Files: []string{bin, "models/big-00001-of-00002.gguf", "models/big-00002-of-00002.gguf"}}, + {Name: "tranche 2 désignée", Config: map[string]string{"BIN": bin, "MODEL": "big-00002-of-00002.gguf"}, + Files: []string{bin, "models/big-00002-of-00002.gguf"}}, + {Name: "MODEL guillemets", Config: map[string]string{"BIN": bin, "MODEL": " m.gguf "}, Files: ok}, + } { + reset() + writeFiles(c.Files) + setCfg(c.Config) + if len(c.Dirs) > 0 { + var abs []string + for _, d := range c.Dirs { + abs = append(abs, filepath.Join(home, d)) + } + if err := putJSON(bkState, "model_dirs", abs); err != nil { + t.Fatal(err) + } + } + c.Out = strings.ReplaceAll(plainErr(preflightEngine()), home, "{{HOME}}") + fx.Preflight = append(fx.Preflight, c) + } + + // --- routes sans processus lancé --- + reset() + run := func(st parityEngineStep) parityEngineStep { + reset() + setCfg(st.Config) + var lock *tuneLock + if st.Tune { + l, err := tuneLockAcquire("web", false) + if err != nil { + t.Fatal(err) + } + lock = l + } + conv.mu.Lock() + conv.Generating = st.Generating + conv.mu.Unlock() + defer func() { + if lock != nil { + lock.release() + } + conv.mu.Lock() + conv.Generating = false + conv.mu.Unlock() + }() + h := map[string]http.HandlerFunc{ + "/api/vram/unload": handleVramUnload, + "/api/vram/reload": handleVramReload, + }[st.Path] + rec := httptest.NewRecorder() + h(rec, httptest.NewRequest(st.Method, st.Path, strings.NewReader(st.Body))) + st.Status = rec.Code + st.Allow = rec.Header().Get("Allow") + st.Out = parityDecode(t, rec.Body.String(), home) + return st + } + for _, st := range []parityEngineStep{ + {Name: "déchargement en GET", Method: "GET", Path: "/api/vram/unload"}, + {Name: "rechargement en GET", Method: "GET", Path: "/api/vram/reload"}, + {Name: "rechargement en PUT", Method: "PUT", Path: "/api/vram/reload"}, + {Name: "génération en cours", Method: "POST", Path: "/api/vram/unload", Body: `{}`, Generating: true}, + {Name: "génération en cours, corps illisible", Method: "POST", Path: "/api/vram/unload", Body: `nimp`, Generating: true}, + {Name: "moteur déjà arrêté", Method: "POST", Path: "/api/vram/unload", Body: `{"force":true}`, Generating: true}, + {Name: "moteur arrêté, sans corps", Method: "POST", Path: "/api/vram/unload"}, + {Name: "optimisation en cours", Method: "POST", Path: "/api/vram/reload", Tune: true}, + {Name: "préflight sans BIN", Method: "POST", Path: "/api/vram/reload", Config: map[string]string{"MODEL": "m.gguf"}}, + {Name: "préflight sans MODEL", Method: "POST", Path: "/api/vram/reload", + Config: map[string]string{"BIN": "/bin/sh"}}, + } { + fx.Steps = append(fx.Steps, run(st)) + } + + got, err := parityJSON(fx) + if err != nil { + t.Fatal(err) + } + got = []byte(strings.ReplaceAll(string(got), home, "{{HOME}}")) + if os.Getenv("LOKI_PARITY_UPDATE") == "1" { + if err := os.MkdirAll(filepath.Dir(parityEngineFile), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(parityEngineFile, got, 0o644); err != nil { + t.Fatal(err) + } + return + } + want, err := os.ReadFile(parityEngineFile) + if err != nil { + t.Fatalf("journal de référence absent (%v) : LOKI_PARITY_UPDATE=1 pour le créer", err) + } + if string(got) != string(want) { + t.Errorf("le pilotage du moteur ne produit plus le journal rangé dans %s.\n"+ + "Si c'est voulu : LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteMoteur,\n"+ + "puis cargo test --test enginectl dit si le port Rust suit.\n--- obtenu ---\n%s", parityEngineFile, got) + } +} + +// parityDecode : le corps JSON d'une réponse, tel quel s'il n'en est pas un. +func parityDecode(t *testing.T, body, home string) any { + t.Helper() + var v any + if err := json.Unmarshal([]byte(strings.ReplaceAll(body, home, "{{HOME}}")), &v); err != nil { + return body + } + return v +} diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index e2d0e41..2274d93 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -113,6 +113,12 @@ func withInternalAPI(next http.Handler) http.Handler { handleInternalPresetApply(w, r) case "/internal/v1/tune/guard": handleInternalTuneGuard(w, r) + case "/internal/v1/service/prestart": + handleInternalServicePrestart(w, r) + case "/internal/v1/chat/generating": + handleInternalChatGenerating(w, r) + case "/internal/v1/dictate/stop": + handleInternalDictateStop(w, r) default: http.NotFound(w, r) } diff --git a/rust/loki-rs/src/api.rs b/rust/loki-rs/src/api.rs index e44c354..0487d3d 100644 --- a/rust/loki-rs/src/api.rs +++ b/rust/loki-rs/src/api.rs @@ -42,6 +42,8 @@ pub struct Ctx { pub mem: Option>, /// Presets tenus par la façade (`LOKI_PRESETS=rust`). pub presets: bool, + /// Pilotage du moteur tenu par la façade (`LOKI_ENGINE=rust`). + pub engine_ctl: Option>, /// Jeton de l'API interne : une requête qui le porte vient du Go. pub internal_token: String, } @@ -58,6 +60,7 @@ impl Ctx { || self.push.is_some() || self.mem.is_some() || self.presets + || self.engine_ctl.is_some() } /// La requête vient du Go (jeton interne valide) : déjà authentifiée par diff --git a/rust/loki-rs/src/enginectl.rs b/rust/loki-rs/src/enginectl.rs new file mode 100644 index 0000000..b3eb06a --- /dev/null +++ b/rust/loki-rs/src/enginectl.rs @@ -0,0 +1,561 @@ +//! Pilotage du moteur tenu par la façade (lot 16 de la migration, +//! `LOKI_ENGINE=rust` ; expérimental, le Go reste le défaut) : +//! +//! - `/api/start|stop|restart` — port de `svcHandler` (`web_api.go`) et de la +//! voie conteneur de `serviceAction` (`sys_service.go`, +//! `sys_service_rust_linux.go`) : la façade lance elle-même +//! `loki-rs engine `, le superviseur que le Go appelait déjà ; +//! - `/api/vram/unload` et `/api/vram/reload` — port de `handleVramUnload` et +//! `handleVramReload` (`web_vram.go`), bilan de VRAM (`gpuSettle`) compris ; +//! - `preflightEngine` (`sys_service.go`) : ce sans quoi le moteur ne PEUT +//! pas démarrer, dit avant de le lancer. +//! +//! Restent au Go, derrière son API interne, jusqu'au lot qui les portera : le +//! verrou de l'optimiseur (`/internal/v1/service/prestart` avant tout +//! démarrage, `/internal/v1/tune/guard`), la génération en cours quand il +//! tient la conversation (`/internal/v1/chat/generating`) et l'arrêt de la +//! dictée quand il la tient (`/internal/v1/dictate/stop`). +//! +//! La façade ne tient ce lot qu'en conteneur, avec son superviseur : sous +//! systemd ou avec `LOKI_ENGINE_SUPERVISOR=go`, le Go le garde ([`usable`]). + +use std::path::{Path, PathBuf}; +use std::sync::LazyLock; +use std::time::Duration; + +use http_body_util::{BodyExt, Limited}; +use hyper::body::Incoming; +use hyper::header::{ALLOW, HeaderValue}; +use hyper::{Method, Request, Response, StatusCode}; +use regex::Regex; +use serde_json::{Value, json}; + +use crate::api::Ctx; +use crate::auth::json as send_json; +use crate::gguf; +use crate::gojson::{self, Kind}; +use crate::paths::{go_dir, go_is_abs, go_join}; +use crate::proxy::Body; +use crate::serve::{self, Cfg}; +use crate::store::Store; +use crate::sysinfo::{self, GpuStat}; + +/// Ce que la façade sait du superviseur du moteur. +pub struct EngineCtl { + /// Le binaire lancé en `engine ` (`loki-rs` lui-même). + pub bin: PathBuf, + /// Variables passées au superviseur, en plus de l'environnement de la + /// façade : `LOKI_GO_BIN`, `LOKI_HOME`, `LOKI_SERVICE` (comme + /// `rustSvcAction`), et l'adresse et le jeton de l'API interne du Go, que + /// `loki-rs engine serve` (`LOKI_SERVE=rust`) lit. + pub env: Vec<(String, String)>, + /// nvidia-smi (un faux dans les tests). + pub nvidia_smi: String, + /// Pause entre deux lectures du bilan de VRAM (500 ms). + pub settle_pause: Duration, +} + +impl EngineCtl { + pub fn new(bin: PathBuf, env: Vec<(String, String)>) -> Self { + EngineCtl { + bin, + env, + nvidia_smi: "nvidia-smi".into(), + settle_pause: Duration::from_millis(500), + } + } + + /// `rustSvcAction` : `loki-rs engine `. Sa sortie s'affiche telle + /// quelle ; en cas d'échec, sa dernière ligne d'erreur devient l'erreur + /// rendue (c'est elle que l'UI montre), sans le préfixe du binaire. + pub async fn run(&self, action: &str) -> Result<(), String> { + use std::io::Write; + let out = tokio::process::Command::new(&self.bin) + .args(["engine", action]) + .envs(self.env.iter().map(|(k, v)| (k.as_str(), v.as_str()))) + .stdin(std::process::Stdio::null()) + .output() + .await + .map_err(|e| format!("loki-rs engine {action} : {e}"))?; + let _ = std::io::stdout().write_all(&out.stdout); + let _ = std::io::stderr().write_all(&out.stderr); + if out.status.success() { + return Ok(()); + } + let msg = last_error_line(&String::from_utf8_lossy(&out.stderr)); + if !msg.is_empty() { + return Err(msg); + } + Err(format!( + "loki-rs engine {action} : {}", + exit_text(&out.status) + )) + } + + /// `gpuStats` : lecture FRAÎCHE, vide sans mesure. + async fn gpus(&self) -> Vec { + sysinfo::nvidia_smi_query_bin(&self.nvidia_smi) + .await + .map(|o| sysinfo::parse_nvidia_smi(&String::from_utf8_lossy(&o))) + .unwrap_or_default() + } +} + +/// `ProcessState.String()` du Go : « exit status 1 », « signal: killed ». +fn exit_text(st: &std::process::ExitStatus) -> String { + if let Some(c) = st.code() { + return format!("exit status {c}"); + } + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + if let Some(s) = st.signal() { + let name = match s { + 1 => "hangup".to_string(), + 2 => "interrupt".to_string(), + 6 => "aborted".to_string(), + 9 => "killed".to_string(), + 11 => "segmentation fault".to_string(), + 15 => "terminated".to_string(), + n => format!("signal {n}"), + }; + return format!("signal: {name}"); + } + } + st.to_string() +} + +/// La façade peut-elle tenir ce lot ? `Err(raison)` sinon : sous systemd +/// (`systemdAvailable`), le Go pilote son unité ; `LOKI_ENGINE_SUPERVISOR=go` +/// est la porte de sortie du superviseur Rust. +pub fn usable(container_env: Option<&str>, supervisor_env: Option<&str>) -> Result<(), String> { + if supervisor_env.is_some_and(|v| v.trim().eq_ignore_ascii_case("go")) { + return Err("LOKI_ENGINE_SUPERVISOR=go".into()); + } + if container_env != Some("1") && Path::new("/run/systemd/system").is_dir() { + return Err("systemd pilote le moteur".into()); + } + Ok(()) +} + +/// `lastErrorLine`. +pub fn last_error_line(s: &str) -> String { + let t = s.trim(); + let last = t.rsplit('\n').next().unwrap_or("").trim(); + last.strip_prefix("[loki-rs engine]") + .unwrap_or(last) + .trim() + .to_string() +} + +static ANSI: LazyLock = LazyLock::new(|| Regex::new("\x1b\\[[0-9;]*m").expect("motif ANSI")); + +/// `plainErr` : les séquences ANSI des messages écrits pour le terminal +/// retirées. +pub fn plain_err(s: &str) -> String { + ANSI.replace_all(s, "").into_owned() +} + +/// `gpuUsedMB`. +pub fn gpu_used_mb(gpus: &[GpuStat]) -> i64 { + gpus.iter().map(|g| g.used).sum() +} + +/// `gpuSettle` : attend que la VRAM ait fini de redescendre avant de la +/// relire. Une mesure ratée n'est pas « 0 Mo » (sautée) ; un palier ne +/// conclut qu'une fois la baisse commencée ; deux lectures égales de suite +/// après la baisse, et la valeur est rendue — sinon la dernière lecture après +/// huit essais. `sample` rend (Mo occupés, lecture réussie). +pub async fn gpu_settle(before: i64, mut sample: F, pause: Duration) -> i64 +where + F: FnMut() -> Fut, + Fut: std::future::Future, +{ + let (mut last, mut lu) = (before, false); + let (mut baisse, mut palier) = (false, 0); + for _ in 0..8 { + if !pause.is_zero() { + tokio::time::sleep(pause).await; + } + let (cur, ok) = sample().await; + if !ok { + continue; + } + if cur < before { + baisse = true; + } + if lu && cur == last { + palier += 1; + } else { + palier = 0; + } + (last, lu) = (cur, true); + if baisse && palier >= 1 { + return cur; + } + } + last +} + +/// `strataPreflight`. +fn strata_preflight(cfg: &Cfg) -> Result<(), String> { + let path = serve::trim(serve::get(cfg, "STRATA_CONFIG")); + if path.is_empty() { + return Err( + "STRATA_CONFIG non défini : réinstalle le modèle depuis Paramètres → Moteur → Strata" + .into(), + ); + } + if std::fs::metadata(path).is_err() { + return Err(format!( + "configuration de Strata introuvable : {path} — réinstalle le modèle" + )); + } + let py = go_join(&[&go_dir(path), ".venv", "bin", "python"]); + if std::fs::metadata(&py).is_err() { + return Err(format!( + "environnement Python de Strata absent ({py}) : réinstalle le modèle" + )); + } + Ok(()) +} + +/// `preflightEngine` (message déjà débarrassé de ses couleurs, comme +/// `plainErr` le rend à l'UI). Entrées/sorties bloquantes. +/// +/// - `stored_dirs` / `env_model_dirs` : voir +/// [`gguf::resolve_serve_model_path`]. +pub fn preflight_engine( + home: &Path, + cfg: &Cfg, + stored_dirs: &[String], + env_model_dirs: Option<&str>, +) -> Result<(), String> { + // Preset externe : aucun llama-server local à lancer. + if serve::is_external(cfg) { + return Ok(()); + } + // Strata n'a ni BIN ni MODEL : son installeur a tout rangé. + if serve::is_strata(cfg) { + return strata_preflight(cfg); + } + let mut bin = serve::trim(serve::get(cfg, "BIN")).to_string(); + if bin.is_empty() { + return Err( + "BIN non défini — installe un moteur : loki llamacpp install (ou renseigne BIN avec loki edit)" + .into(), + ); + } + if !go_is_abs(&bin) { + bin = go_join(&[&home.to_string_lossy(), &bin]); + } + let eng = serve::run::EnginePaths { + home: home.to_path_buf(), + }; + if std::fs::metadata(eng.prebuilt_resolve_bin(&bin)).is_err() { + return Err(format!( + "moteur introuvable : {bin} — relance loki llamacpp install" + )); + } + let model = serve::trim(serve::get(cfg, "MODEL")); + if model.is_empty() { + return Err(format!( + "MODEL non défini — indique un .gguf avec loki edit (dossier des modèles : {})", + gguf::models_dir(home) + )); + } + let p = gguf::resolve_serve_model_path(model, home, stored_dirs, env_model_dirs) + .map_err(|e| format!("MODEL={model} : {e}"))?; + let p = p.to_string_lossy().into_owned(); + if std::fs::metadata(&p).is_err() { + return Err(format!( + "modèle introuvable : {p} — corrige MODEL avec loki edit" + )); + } + // Modèle en plusieurs fichiers : une tranche manquante ne se voit qu'au + // moment où llama-server réclame un tenseur absent. + let (dir, base) = (go_dir(&p), go_base(&p)); + let missing = gguf::shard_family_missing(Path::new(&dir), &base); + if !missing.is_empty() { + return Err(format!( + "modèle incomplet : il manque {} dans {dir} — ce modèle tient en {} fichiers, télécharge-les tous", + missing.join(", "), + gguf::shard_family(&base).len() + )); + } + Ok(()) +} + +/// `filepath.Base`. +fn go_base(p: &str) -> String { + let t = p.trim_end_matches('/'); + if t.is_empty() { + return if p.is_empty() { ".".into() } else { "/".into() }; + } + t.rsplit('/').next().unwrap_or(t).to_string() +} + +/// Les routes servies par ce module. +pub fn handles(path: &str) -> bool { + matches!( + path, + "/api/start" | "/api/stop" | "/api/restart" | "/api/vram/unload" | "/api/vram/reload" + ) +} + +/// Avant un démarrage : refus pendant une optimisation, puis ménage d'un +/// verrou périmé — au Go (`enginePrestart`). +async fn prestart(store: &Store) -> Result<(), String> { + let (code, v) = store + .internal_post("/internal/v1/service/prestart", &json!({})) + .await + .map_err(|e| format!("verrou de l'optimiseur : {e}"))?; + if !code.is_success() { + return Err(format!("verrou de l'optimiseur : {code}")); + } + match v.get("error").and_then(Value::as_str) { + Some(e) if !e.is_empty() => Err(e.to_string()), + _ => Ok(()), + } +} + +/// `serviceAction` en conteneur : garde-fous du Go avant un démarrage, preset +/// externe (rien à lancer : `loki serve` sortirait aussitôt et le superviseur +/// le prendrait pour un plantage), puis le superviseur. +pub async fn service_action(ctx: &Ctx, ctl: &EngineCtl, action: &str) -> Result<(), String> { + if action == "start" || action == "restart" { + prestart(&ctx.store).await?; + } + match action { + "start" | "restart" => { + if serve::is_external(&ctx.store.config().await) { + if action == "restart" { + let _ = ctl.run("stop").await; + } + println!("[info] preset externe actif : pas de moteur local à démarrer"); + return Ok(()); + } + ctl.run(action).await + } + "stop" | "status" => ctl.run(action).await, + other => Err(format!("action inconnue: {other}")), + } +} + +/// `serviceIsActive` en conteneur : le PID du fichier vit. +fn service_is_active(ctx: &Ctx) -> bool { + pid_alive(crate::engine::read_pid(&ctx.engine.pid())) +} + +#[cfg(target_os = "linux")] +fn pid_alive(pid: i32) -> bool { + crate::engine::proctree::linux::pid_alive(pid) +} + +#[cfg(not(target_os = "linux"))] +fn pid_alive(_: i32) -> bool { + false +} + +/// La conversation écrit-elle une réponse ? Celle de la façade si elle la +/// tient, sinon celle du Go. +async fn generating(ctx: &Ctx) -> bool { + if let Some(chat) = &ctx.chat { + return chat.sessions.conv().is_generating(); + } + ctx.store + .internal_json(Method::GET, "/internal/v1/chat/generating", None) + .await + .ok() + .and_then(|v| v.get("generating").and_then(Value::as_bool)) + .unwrap_or(false) +} + +/// `asrShutdownVite` : la dictée de la façade si elle la tient, sinon celle +/// du Go. +async fn dictate_stop(ctx: &Ctx) { + if let Some(d) = &ctx.dictate { + d.shutdown_fast().await; + return; + } + if let Err(e) = ctx + .store + .internal_post("/internal/v1/dictate/stop", &json!({})) + .await + { + eprintln!("[dictée] arrêt demandé au Go : {e}"); + } +} + +/// Le verrou de l'optimiseur (au Go) : `Some(réponse 409)` s'il tient le +/// moteur. +async fn tune_deny(store: &Store) -> Option> { + let v = store + .internal_json(Method::GET, "/internal/v1/tune/guard", None) + .await + .ok()?; + if v.get("busy").and_then(Value::as_bool) == Some(true) { + return Some(send_json( + StatusCode::CONFLICT, + &json!({ "ok": false, "tuning": true, "error": v.get("error").and_then(Value::as_str).unwrap_or("") }), + )); + } + None +} + +/// `postOnly` : ces routes arrêtent ou relancent des processus ; un simple GET +/// (une balise `` sur une page tierce) ne doit pas y suffire. +fn post_only(method: &Method) -> Option> { + if method == Method::POST { + return None; + } + let mut r = send_json( + StatusCode::METHOD_NOT_ALLOWED, + &json!({ "ok": false, "error": format!("méthode {method} refusée : POST attendu") }), + ); + r.headers_mut() + .insert(ALLOW, HeaderValue::from_static("POST")); + Some(r) +} + +const UNLOAD_REQ_TYPE: &str = r#"struct { Force bool "json:\"force\"" }"#; + +/// `from_go` : remise par le Go (tunnel), déjà authentifiée par lui. +pub async fn serve( + ctx: &Ctx, + ctl: &EngineCtl, + req: Request, + from_go: bool, +) -> Response { + let (parts, body) = req.into_parts(); + if !from_go && let Err(refus) = ctx.guard(&parts.headers).await { + return *refus; + } + match parts.uri.path() { + p @ ("/api/start" | "/api/stop" | "/api/restart") => { + let action = &p["/api/".len()..]; + let res = service_action(ctx, ctl, action).await; + let msg = match &res { + Ok(()) => "ok".to_string(), + Err(e) => e.clone(), + }; + send_json(StatusCode::OK, &json!({ "ok": res.is_ok(), "out": msg })) + } + "/api/vram/unload" => { + if let Some(r) = post_only(&parts.method) { + return r; + } + let raw = Limited::new(body, 1 << 20) + .collect() + .await + .map(|b| b.to_bytes().to_vec()) + .unwrap_or_default(); + let force = gojson::decode_struct(&raw, UNLOAD_REQ_TYPE, &[("force", Kind::Bool)]) + .bool("force"); + if !force && generating(ctx).await { + return send_json( + StatusCode::CONFLICT, + &json!({ "ok": false, "generating": true, + "error": "une réponse est en cours d'écriture — décharger maintenant l'interromprait" }), + ); + } + unload(ctx, ctl).await + } + "/api/vram/reload" => { + if let Some(r) = post_only(&parts.method) { + return r; + } + if let Some(r) = tune_deny(&ctx.store).await { + return r; + } + if service_is_active(ctx) { + return send_json(StatusCode::OK, &json!({ "ok": true, "already": true })); + } + let cfg = ctx.store.config().await; + let d = crate::models::model_dirs(ctx).await; + let pre = tokio::task::spawn_blocking(move || { + preflight_engine(&d.home, &cfg, &d.stored, d.env.as_deref()) + }) + .await + .unwrap_or_else(|e| Err(e.to_string())); + if let Err(e) = pre { + return send_json( + StatusCode::OK, + &json!({ "ok": false, "error": plain_err(&e) }), + ); + } + if let Err(e) = service_action(ctx, ctl, "start").await { + return send_json( + StatusCode::OK, + &json!({ "ok": false, "error": plain_err(&e) }), + ); + } + send_json(StatusCode::OK, &json!({ "ok": true })) + } + _ => send_json(StatusCode::NOT_FOUND, &json!({ "error": "route inconnue" })), + } +} + +/// `handleVramUnload` après les refus : moteur arrêté, dictée coupée, bilan +/// de la VRAM rendue. +async fn unload(ctx: &Ctx, ctl: &EngineCtl) -> Response { + let gpus = ctl.gpus().await; + let before = gpu_used_mb(&gpus); + // `engineNeedsStop` : en conteneur, actif suffit (les états transitoires + // n'existent que sous systemd, où ce lot reste au Go). + let was_active = service_is_active(ctx); + let mut errs = Vec::new(); + if was_active && let Err(e) = service_action(ctx, ctl, "stop").await { + errs.push(format!("arrêt du moteur : {}", plain_err(&e))); + } + dictate_stop(ctx).await; + let mut after = before; + if !gpus.is_empty() { + after = gpu_settle( + before, + || async { + let g = ctl.gpus().await; + (gpu_used_mb(&g), !g.is_empty()) + }, + ctl.settle_pause, + ) + .await; + } + // Les jauges ne doivent pas montrer, deux secondes encore, une carte + // pleine qu'on vient d'annoncer libérée. + sysinfo::GPU_MONITOR.invalidate().await; + let freed = (before - after).max(0); + send_json( + StatusCode::OK, + &json!({ + "ok": errs.is_empty(), + "error": errs.join(" ; "), + "was_active": was_active, + "active": service_is_active(ctx), + "gpu": !gpus.is_empty(), + "before_mb": before, + "used_mb": after, + "freed_mb": freed, + }), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn base_comme_go() { + assert_eq!(go_base("/a/b.gguf"), "b.gguf"); + assert_eq!(go_base("/a/b/"), "b"); + assert_eq!(go_base("/"), "/"); + assert_eq!(go_base(""), "."); + } + + #[test] + fn porte_de_sortie() { + assert!(usable(Some("1"), Some(" Go ")).is_err()); + assert!(usable(Some("1"), None).is_ok()); + assert!(usable(Some("1"), Some("rust")).is_ok()); + } +} diff --git a/rust/loki-rs/src/gguf.rs b/rust/loki-rs/src/gguf.rs index f548cc8..bc26786 100644 --- a/rust/loki-rs/src/gguf.rs +++ b/rust/loki-rs/src/gguf.rs @@ -665,6 +665,16 @@ pub fn shard_family(name: &str) -> Vec { } } +/// `shardFamilyMissing` : tranches ABSENTES de `dir` (un dossier du même nom +/// ne compte pas). Un modèle dont il manque une tranche démarre puis meurt sur +/// une erreur de tenseur introuvable. +pub fn shard_family_missing(dir: &Path, name: &str) -> Vec { + shard_family(name) + .into_iter() + .filter(|n| !std::fs::metadata(dir.join(n)).is_ok_and(|m| !m.is_dir())) + .collect() +} + // --- dossiers de modèles (backend_model_dirs.go) --------------------------- /// `modelsDir()` : `LOKI_HOME/models`. diff --git a/rust/loki-rs/src/lib.rs b/rust/loki-rs/src/lib.rs index 91fa5e4..f1c81c1 100644 --- a/rust/loki-rs/src/lib.rs +++ b/rust/loki-rs/src/lib.rs @@ -19,6 +19,7 @@ pub mod convfiles; pub mod dictate; pub mod downloads; pub mod engine; +pub mod enginectl; pub mod engstatus; pub mod gguf; pub mod gojson; diff --git a/rust/loki-rs/src/main.rs b/rust/loki-rs/src/main.rs index da75b67..506136e 100644 --- a/rust/loki-rs/src/main.rs +++ b/rust/loki-rs/src/main.rs @@ -148,6 +148,20 @@ fn serve() -> ExitCode { // LOKI_PRESETS=rust : les presets (liste, éditeur, aperçus, externes, // bascule ; l'application au moteur reste au Go). let presets_rust = held_by_facade("LOKI_PRESETS"); + // LOKI_ENGINE=rust : le pilotage du moteur (démarrage, arrêt, + // déchargement de la VRAM), en conteneur avec le superviseur Rust + // seulement — sous systemd, le Go pilote son unité. + let engine_rust = held_by_facade("LOKI_ENGINE") + && match loki_rs::enginectl::usable( + std::env::var("LOKI_CONTAINER").ok().as_deref(), + std::env::var("LOKI_ENGINE_SUPERVISOR").ok().as_deref(), + ) { + Ok(()) => true, + Err(why) => { + eprintln!("[loki-rs] pilotage du moteur tenu par le Go ({why})"); + false + } + }; let mut env = vec![ (auth::INTERNAL_ENV.to_string(), token.clone()), // Adresse de l'API interne du Go, héritée par ses enfants : @@ -163,6 +177,7 @@ fn serve() -> ExitCode { ("LOKI_PUSH", push_rust), ("LOKI_MEM", mem_rust), ("LOKI_PRESETS", presets_rust), + ("LOKI_ENGINE", engine_rust), ]; for (var, on) in held { if on { @@ -172,6 +187,7 @@ fn serve() -> ExitCode { if held.iter().any(|(_, on)| *on) { env.push(("LOKI_FACADE_ADDR".into(), format!("127.0.0.1:{web_port}"))); } + let go_bin_path = go_bin.clone(); let sup = child::Supervisor::spawn(go_bin, go_port, env); wait_for(backend, Duration::from_secs(30)).await; let engine_paths = engine::Paths::from_env(); @@ -230,6 +246,26 @@ fn serve() -> ExitCode { }; eprintln!("[loki-rs] façade : http://{addr} → {backend}"); let mem_home = engine_paths.home.clone(); + // Le superviseur reçoit ce que le Go lui passait (rustSvcAction), et de + // quoi joindre l'API interne du Go pour « loki-rs engine serve ». + let engine_ctl = engine_rust.then(|| { + std::sync::Arc::new(loki_rs::enginectl::EngineCtl::new( + exe(), + vec![ + ( + "LOKI_GO_BIN".into(), + go_bin_path.to_string_lossy().into_owned(), + ), + ( + "LOKI_HOME".into(), + engine_paths.home.to_string_lossy().into_owned(), + ), + ("LOKI_SERVICE".into(), engine_paths.service.clone()), + ("LOKI_GO_ADDR".into(), backend.to_string()), + (auth::INTERNAL_ENV.into(), token.clone()), + ], + )) + }); let cfg = proxy::Config { backend, internal_token: token, @@ -243,6 +279,7 @@ fn serve() -> ExitCode { push: push_rust.then(|| std::sync::Arc::new(loki_rs::push::Push::new())), mem: mem_rust.then(|| std::sync::Arc::new(loki_rs::memory::Memory::new(mem_home))), presets: presets_rust, + engine_ctl, }; let res = proxy::serve(listener, cfg, shutdown_signal()).await; eprintln!("[loki-rs] arrêt demandé — arrêt du serveur Go"); diff --git a/rust/loki-rs/src/proxy.rs b/rust/loki-rs/src/proxy.rs index dde5fee..ee26fb6 100644 --- a/rust/loki-rs/src/proxy.rs +++ b/rust/loki-rs/src/proxy.rs @@ -79,6 +79,9 @@ pub struct Config { pub mem: Option>, /// Presets tenus par la façade (`LOKI_PRESETS=rust`). pub presets: bool, + /// Pilotage du moteur tenu par la façade (`LOKI_ENGINE=rust`) : + /// démarrage, arrêt, déchargement de la VRAM ; `None` : au Go. + pub engine_ctl: Option>, } struct Proxy { @@ -113,6 +116,7 @@ pub async fn serve( push: cfg.push, mem: cfg.mem, presets: cfg.presets, + engine_ctl: cfg.engine_ctl, internal_token: cfg.internal_token.clone(), }, }); @@ -200,6 +204,11 @@ impl Proxy { { return chatapi::serve(&self.ctx, chat, req, from_go).await; } + if let Some(e) = &self.ctx.engine_ctl + && crate::enginectl::handles(path) + { + return crate::enginectl::serve(&self.ctx, e, req, from_go).await; + } if self.ctx.presets && crate::presets_api::handles(path) { return crate::presets_api::serve(&self.ctx, req, from_go).await; } diff --git a/rust/loki-rs/src/sysinfo.rs b/rust/loki-rs/src/sysinfo.rs index 8fcee3e..3f69b94 100644 --- a/rust/loki-rs/src/sysinfo.rs +++ b/rust/loki-rs/src/sysinfo.rs @@ -107,7 +107,7 @@ pub async fn nvidia_smi_query() -> io::Result> { } /// [`nvidia_smi_query`] avec un exécutable choisi (tests). -async fn nvidia_smi_query_bin(bin: &str) -> io::Result> { +pub(crate) async fn nvidia_smi_query_bin(bin: &str) -> io::Result> { let mut cmd = tokio::process::Command::new(bin); cmd.args([ "--query-gpu=name,memory.used,memory.total,utilization.gpu,temperature.gpu", diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index cc13f81..0632d7c 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -78,6 +78,12 @@ pub struct FakeGo { pub tune_busy: Mutex>, /// /internal/v1/preset/apply reçus : (chemin, mode). pub preset_applies: Mutex>, + /// /internal/v1/chat/generating : le Go écrit une réponse. + pub generating: Mutex, + /// /internal/v1/service/prestart reçus. + pub prestarts: Mutex, + /// /internal/v1/dictate/stop reçus. + pub dictate_stops: Mutex, /// Comportement de /internal/v1/crypt/* (préfixe « SCELLE: » par défaut). pub crypt_mode: Mutex, /// Réponse de /internal/v1/conv/workspace (None : 404). @@ -336,6 +342,19 @@ async fn handle(st: Arc, req: Request) -> Response ), None => resp(200, r#"{"busy":false}"#), }, + "/internal/v1/service/prestart" => { + *st.prestarts.lock().unwrap() += 1; + let e = st.tune_busy.lock().unwrap().clone().unwrap_or_default(); + resp(200, serde_json::json!({ "error": e }).to_string()) + } + "/internal/v1/chat/generating" => resp( + 200, + serde_json::json!({ "generating": *st.generating.lock().unwrap() }).to_string(), + ), + "/internal/v1/dictate/stop" => { + *st.dictate_stops.lock().unwrap() += 1; + resp(204, "") + } "/internal/v1/preset/apply" => { let body = req.into_body().collect().await.unwrap().to_bytes(); let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default(); @@ -530,6 +549,7 @@ pub struct Opts { pub push: Option>, pub mem: Option>, pub presets: bool, + pub engine_ctl: Option>, } /// Comme [`start`], avec la conversation tenue par la façade @@ -626,6 +646,7 @@ pub async fn start_with( push: opts.push, mem: opts.mem, presets: opts.presets, + engine_ctl: opts.engine_ctl, }; tokio::spawn(loki_rs::proxy::serve(fl, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/enginectl.rs b/rust/loki-rs/tests/enginectl.rs new file mode 100644 index 0000000..52bdbb0 --- /dev/null +++ b/rust/loki-rs/tests/enginectl.rs @@ -0,0 +1,470 @@ +//! Lot 16 : pilotage du moteur tenu par la façade (`LOKI_ENGINE=rust`). +//! +//! Le journal tiré du Go (`internal/loki/parity_engine_test.go` → +//! `tests/parity_engine/engine.json`) est rejoué : bilan de VRAM +//! (`gpuSettle`), contrôle d'avant démarrage (`preflightEngine`) sur chaque +//! installation, dernière ligne d'erreur du superviseur, puis les réponses des +//! routes qui ne lancent aucun processus. Le reste — démarrage, arrêt, +//! redémarrage, déchargement avec bilan chiffré, rechargement — est vérifié de +//! bout en bout avec un faux superviseur (`loki-rs engine `) et un +//! faux nvidia-smi. + +mod common; + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use bytes::Bytes; +use common::fakego; +use http_body_util::{BodyExt, Full}; +use hyper::header::HeaderMap; +use hyper::{Request, StatusCode}; +use loki_rs::enginectl::{self as ec, EngineCtl}; +use serde_json::{Value, json}; + +fn fixture() -> Value { + serde_json::from_str(include_str!("parity_engine/engine.json")).expect("journal") +} + +fn s(v: &Value) -> &str { + v.as_str().unwrap() +} + +/// Écrit un script exécutable sans le garder ouvert (ETXTBSY). +fn script(dir: &Path, name: &str, body: &str) -> PathBuf { + let p = dir.join(name); + let ok = std::process::Command::new("sh") + .args(["-c", "printf %s \"$1\" > \"$2\" && chmod 755 \"$2\""]) + .arg("sh") + .arg(body) + .arg(&p) + .status() + .unwrap() + .success(); + assert!(ok); + p +} + +async fn call( + facade: std::net::SocketAddr, + method: &str, + path: &str, + body: &str, +) -> (StatusCode, HeaderMap, Value) { + use hyper_util::client::legacy::Client; + use hyper_util::rt::TokioExecutor; + let req = Request::builder() + .method(method) + .uri(format!("http://{facade}{path}")) + .body(Full::new(Bytes::from(body.to_string()))) + .unwrap(); + let client: Client<_, Full> = Client::builder(TokioExecutor::new()).build_http(); + let r = client.request(req).await.unwrap(); + let (parts, b) = r.into_parts(); + let b = b.collect().await.unwrap().to_bytes(); + let v = serde_json::from_slice(&b) + .unwrap_or_else(|_| Value::String(String::from_utf8_lossy(&b).into_owned())); + (parts.status, parts.headers, v) +} + +#[tokio::test] +async fn parite_du_bilan_de_vram() { + for c in fixture()["settle"].as_array().unwrap() { + let before = c["before"].as_i64().unwrap(); + let seq: Vec = c["seq"] + .as_array() + .unwrap() + .iter() + .map(|x| x.as_i64().unwrap()) + .collect(); + let i = std::sync::Mutex::new(0usize); + let got = ec::gpu_settle( + before, + || { + let mut n = i.lock().unwrap(); + let v = if *n >= seq.len() { + (seq[seq.len() - 1], true) + } else { + let v = seq[*n]; + *n += 1; + if v < 0 { (0, false) } else { (v, true) } + }; + async move { v } + }, + Duration::ZERO, + ) + .await; + assert_eq!(got, c["out"].as_i64().unwrap(), "{c}"); + } +} + +#[test] +fn parite_des_messages() { + let fx = fixture(); + for (k, v) in fx["last_error"].as_object().unwrap() { + assert_eq!(ec::last_error_line(k), s(v), "{k:?}"); + } + for (k, v) in fx["plain"].as_object().unwrap() { + assert_eq!(ec::plain_err(k), s(v), "{k:?}"); + } +} + +#[test] +fn parite_du_preflight() { + for c in fixture()["preflight"].as_array().unwrap() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path(); + let h = home.to_string_lossy().into_owned(); + for f in c["files"].as_array().into_iter().flatten() { + let f = s(f); + let p = home.join(f.trim_end_matches('/')); + if f.ends_with('/') { + std::fs::create_dir_all(&p).unwrap(); + } else { + std::fs::create_dir_all(p.parent().unwrap()).unwrap(); + std::fs::write(&p, "x").unwrap(); + } + } + let cfg: HashMap = c["config"] + .as_object() + .unwrap() + .iter() + .map(|(k, v)| (k.clone(), s(v).replace("{{HOME}}", &h))) + .collect(); + let dirs: Vec = c["dirs"] + .as_array() + .into_iter() + .flatten() + .map(|d| home.join(s(d)).to_string_lossy().into_owned()) + .collect(); + let got = match ec::preflight_engine(home, &cfg, &dirs, None) { + Ok(()) => String::new(), + Err(e) => e.replace(&h, "{{HOME}}"), + }; + assert_eq!(got, s(&c["out"]), "{}", c["name"]); + } +} + +/// Un faux superviseur : note chaque appel, démarre un `sleep` dont le PID +/// va dans le fichier du service, l'arrête sur « stop » ; « start » échoue si +/// `$LOKI_HOME/echec` existe, comme le vrai : dernière ligne d'erreur. +fn fake_supervisor(dir: &Path) -> PathBuf { + script( + dir, + "loki-rs", + r#"#!/bin/sh +echo "$1 $2 $LOKI_HOME $LOKI_SERVICE $LOKI_GO_ADDR $LOKI_INTERNAL_TOKEN" >> "$LOKI_HOME/trace" +pid="$LOKI_HOME/$LOKI_SERVICE.pid" +case "$2" in +start) + if [ -e "$LOKI_HOME/echec" ]; then + echo "[ERREUR] le processus s'est arrêté" + echo "[loki-rs engine] service $LOKI_SERVICE non démarré" >&2 + exit 1 + fi + sleep 60 >/dev/null 2>&1 < /dev/null & + echo $! > "$pid" + ;; +stop) + [ -s "$pid" ] && kill "$(cat "$pid")" 2>/dev/null + rm -f "$pid" + ;; +esac +exit 0 +"#, + ) +} + +struct Banc { + facade: std::net::SocketAddr, + go: Arc, + home: PathBuf, + _stop: tokio::sync::oneshot::Sender<()>, + _tmp: tempfile::TempDir, +} + +impl Banc { + fn trace(&self) -> Vec { + std::fs::read_to_string(self.home.join("trace")) + .unwrap_or_default() + .lines() + .map(|l| l.split(' ').take(2).collect::>().join(" ")) + .collect() + } + fn pid(&self) -> i32 { + loki_rs::engine::read_pid(&self.home.join("loki-engine.pid")) + } +} + +async fn banc(nvidia_smi: Option<&str>) -> Banc { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().to_path_buf(); + let bin = fake_supervisor(&home); + let mut ctl = EngineCtl::new( + bin, + vec![ + ("LOKI_HOME".into(), home.to_string_lossy().into_owned()), + ("LOKI_SERVICE".into(), "loki-engine".into()), + ("LOKI_GO_ADDR".into(), "127.0.0.1:1".into()), + ("LOKI_INTERNAL_TOKEN".into(), "jeton".into()), + ], + ); + ctl.nvidia_smi = match nvidia_smi { + Some(body) => script(&home, "nvidia-smi", body) + .to_string_lossy() + .into_owned(), + None => home + .join("absent-nvidia-smi") + .to_string_lossy() + .into_owned(), + }; + ctl.settle_pause = Duration::ZERO; + let opts = fakego::Opts { + engine_ctl: Some(Arc::new(ctl)), + ..Default::default() + }; + let (facade, go, stop, _) = fakego::start_with(home.clone(), opts).await; + Banc { + facade, + go, + home, + _stop: stop, + _tmp: tmp, + } +} + +#[tokio::test] +async fn parite_des_routes() { + for st in fixture()["steps"].as_array().unwrap() { + let b = banc(None).await; + let h = b.home.to_string_lossy().into_owned(); + if st["tune"].as_bool() == Some(true) { + *b.go.tune_busy.lock().unwrap() = Some( + "optimisation en cours (loki tune) : le moteur est réservé aux essais — attends la fin ou annule-la" + .into(), + ); + } + *b.go.generating.lock().unwrap() = st["generating"].as_bool() == Some(true); + for (k, v) in st["config"].as_object().into_iter().flatten() { + b.go.set("config", k, s(v)); + } + let (code, headers, out) = call( + b.facade, + s(&st["method"]), + s(&st["path"]), + st["body"].as_str().unwrap_or(""), + ) + .await; + let name = &st["name"]; + assert_eq!( + code.as_u16() as i64, + st["status"].as_i64().unwrap(), + "{name}" + ); + assert_eq!( + headers.get("allow").map_or("", |v| v.to_str().unwrap()), + st["allow"].as_str().unwrap_or(""), + "{name}" + ); + let out: Value = serde_json::from_str(&out.to_string().replace(&h, "{{HOME}}")).unwrap(); + assert_eq!(out, st["out"], "{name}"); + // Aucune de ces réponses ne lance quoi que ce soit. + assert!(b.trace().is_empty(), "{name} : {:?}", b.trace()); + } +} + +#[tokio::test] +async fn demarrage_arret_redemarrage() { + let b = banc(None).await; + let (c, _, v) = call(b.facade, "POST", "/api/start", "").await; + assert_eq!( + (c.as_u16(), v.clone()), + (200, json!({"ok": true, "out": "ok"})) + ); + assert_eq!(*b.go.prestarts.lock().unwrap(), 1); + assert!(b.pid() > 0); + // Environnement passé au superviseur. + let line = std::fs::read_to_string(b.home.join("trace")).unwrap(); + assert_eq!( + line.trim(), + format!( + "engine start {} loki-engine 127.0.0.1:1 jeton", + b.home.display() + ) + ); + + // Optimisation en cours : refus, superviseur jamais appelé. + *b.go.tune_busy.lock().unwrap() = Some("optimisation en cours".into()); + let (_, _, v) = call(b.facade, "POST", "/api/restart", "").await; + assert_eq!(v, json!({"ok": false, "out": "optimisation en cours"})); + assert_eq!(b.trace(), ["engine start"]); + // L'arrêt, lui, passe toujours (et sans consulter le verrou). + let (_, _, v) = call(b.facade, "GET", "/api/stop", "").await; + assert_eq!(v, json!({"ok": true, "out": "ok"})); + assert_eq!(*b.go.prestarts.lock().unwrap(), 2); + assert_eq!(b.pid(), 0); + *b.go.tune_busy.lock().unwrap() = None; + + // Échec du démarrage : la dernière ligne d'erreur, sans préfixe. + std::fs::write(b.home.join("echec"), "").unwrap(); + let (_, _, v) = call(b.facade, "POST", "/api/start", "").await; + assert_eq!( + v, + json!({"ok": false, "out": "service loki-engine non démarré"}) + ); + std::fs::remove_file(b.home.join("echec")).unwrap(); + + let (_, _, v) = call(b.facade, "POST", "/api/restart", "").await; + assert_eq!(v, json!({"ok": true, "out": "ok"})); + assert_eq!( + b.trace(), + [ + "engine start", + "engine stop", + "engine start", + "engine restart" + ] + ); + + // Preset externe : rien à lancer ; un redémarrage arrête le moteur local. + b.go.set("config", "EXTERNAL", "1"); + let (_, _, v) = call(b.facade, "POST", "/api/restart", "").await; + assert_eq!(v, json!({"ok": true, "out": "ok"})); + let (_, _, v) = call(b.facade, "POST", "/api/start", "").await; + assert_eq!(v, json!({"ok": true, "out": "ok"})); + assert_eq!(b.trace().last().unwrap(), "engine stop"); + assert_eq!(b.trace().len(), 5); + assert_eq!(b.pid(), 0); +} + +#[tokio::test] +async fn dechargement_avec_bilan() { + // nvidia-smi : 20 Go occupés avant l'arrêt, puis la mémoire part. + let smi = r#"#!/bin/sh +n=$(cat "$0.n" 2>/dev/null || echo 0) +echo $((n+1)) > "$0.n" +case $n in +0) echo "RTX, 20000, 24000, 50, 60" ;; +1) echo "RTX, 20000, 24000, 0, 60" ;; +2) echo "RTX, 9000, 24000, 0, 55" ;; +*) echo "RTX, 700, 24000, 0, 50" ;; +esac +"#; + let b = banc(Some(smi)).await; + let (_, _, v) = call(b.facade, "POST", "/api/start", "").await; + assert_eq!(v["ok"], true); + let pid = b.pid(); + assert!(pid > 0); + + // Une réponse en cours (conversation au Go) : refus, sauf forcé. + *b.go.generating.lock().unwrap() = true; + let (c, _, v) = call(b.facade, "POST", "/api/vram/unload", "{}").await; + assert_eq!(c, 409); + assert_eq!(v["generating"], true); + let (c, _, v) = call(b.facade, "POST", "/api/vram/unload", r#"{"force":true}"#).await; + assert_eq!(c, 200); + assert_eq!( + v, + json!({ + "ok": true, "error": "", "was_active": true, "active": false, "gpu": true, + "before_mb": 20000, "used_mb": 700, "freed_mb": 19300, + }) + ); + assert_eq!(b.trace(), ["engine start", "engine stop"]); + assert_eq!(*b.go.dictate_stops.lock().unwrap(), 1); + assert_eq!(b.pid(), 0); + + // Rechargement : préflight, puis démarrage ; déjà actif ensuite. + let bin = b.home.join("llama-server"); + std::fs::write(&bin, "").unwrap(); + std::fs::create_dir_all(b.home.join("models")).unwrap(); + std::fs::write(b.home.join("models/m.gguf"), "x").unwrap(); + b.go.set("config", "BIN", &bin.to_string_lossy()); + b.go.set("config", "MODEL", "m.gguf"); + let (_, _, v) = call(b.facade, "POST", "/api/vram/reload", "").await; + assert_eq!(v, json!({"ok": true})); + let (_, _, v) = call(b.facade, "POST", "/api/vram/reload", "").await; + assert_eq!(v, json!({"ok": true, "already": true})); + assert_eq!(b.trace(), ["engine start", "engine stop", "engine start"]); + // Démarrage refusé par le verrou après le préflight. + let (_, _, _) = call(b.facade, "POST", "/api/stop", "").await; + *b.go.tune_busy.lock().unwrap() = Some("optimisation en cours".into()); + let (c, _, v) = call(b.facade, "POST", "/api/vram/reload", "").await; + assert_eq!(c, 409); + assert_eq!(v["tuning"], true); + let _ = std::fs::remove_file(b.home.join("loki-engine.pid")); +} + +#[tokio::test] +async fn dechargement_arret_en_echec() { + // Un moteur actif que le superviseur ne sait pas arrêter. + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().to_path_buf(); + let bin = script( + &home, + "loki-rs", + "#!/bin/sh\necho \"[loki-rs engine] \x1b[1marrêt impossible\x1b[0m\" >&2\nexit 1\n", + ); + let mut sleeper = std::process::Command::new("sleep") + .arg("30") + .spawn() + .unwrap(); + std::fs::write(home.join("loki-engine.pid"), sleeper.id().to_string()).unwrap(); + let mut ctl = EngineCtl::new(bin, vec![]); + ctl.nvidia_smi = home.join("absent").to_string_lossy().into_owned(); + let opts = fakego::Opts { + engine_ctl: Some(Arc::new(ctl)), + ..Default::default() + }; + let (facade, go, _stop, _) = fakego::start_with(home.clone(), opts).await; + let (_, _, v) = call(facade, "POST", "/api/vram/unload", "").await; + assert_eq!( + v, + json!({ + "ok": false, "error": "arrêt du moteur : arrêt impossible", "was_active": true, + "active": true, "gpu": false, "before_mb": 0, "used_mb": 0, "freed_mb": 0, + }) + ); + assert_eq!(*go.dictate_stops.lock().unwrap(), 1); + let _ = sleeper.kill(); + let _ = sleeper.wait(); +} + +#[tokio::test] +async fn sans_la_variable_le_moteur_reste_au_go() { + let tmp = tempfile::tempdir().unwrap(); + let (facade, _go, _stop) = fakego::start(tmp.path().to_path_buf()).await; + for p in [ + "/api/start", + "/api/stop", + "/api/restart", + "/api/vram/unload", + "/api/vram/reload", + ] { + let (_, _, b) = fakego::get(facade, p, &[]).await; + assert_eq!(&b[..], format!("go:{p}").as_bytes()); + } +} + +#[tokio::test] +async fn dechargement_vram_en_hausse() { + // Une autre application prend la carte pendant le déchargement : jamais + // de « Mo libérés » négatifs. + let smi = r#"#!/bin/sh +n=$(cat "$0.n" 2>/dev/null || echo 0) +echo $((n+1)) > "$0.n" +if [ "$n" = 0 ]; then echo "A, 5000, 24000, 0, 40"; else echo "A, 6000, 24000, 0, 40"; fi +"#; + let b = banc(Some(smi)).await; + let (_, _, v) = call(b.facade, "POST", "/api/vram/unload", "").await; + assert_eq!( + v, + json!({ + "ok": true, "error": "", "was_active": false, "active": false, "gpu": true, + "before_mb": 5000, "used_mb": 6000, "freed_mb": 0, + }) + ); + assert!(b.trace().is_empty()); +} diff --git a/rust/loki-rs/tests/lot2.rs b/rust/loki-rs/tests/lot2.rs index 5c8b14b..fcdffd7 100644 --- a/rust/loki-rs/tests/lot2.rs +++ b/rust/loki-rs/tests/lot2.rs @@ -101,6 +101,7 @@ async fn banc(cle: Option<&str>) -> Banc { push: None, mem: None, presets: false, + engine_ctl: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/parity_engine/engine.json b/rust/loki-rs/tests/parity_engine/engine.json new file mode 100644 index 0000000..f561ccf --- /dev/null +++ b/rust/loki-rs/tests/parity_engine/engine.json @@ -0,0 +1,551 @@ +{ + "settle": [ + { + "before": 14000, + "out": 500, + "seq": [ + 20000, + 20000, + 12000, + 3000, + 500, + 500 + ] + }, + { + "before": 14000, + "out": 400, + "seq": [ + 15000, + 8000, + 400, + 400 + ] + }, + { + "before": 14000, + "out": 600, + "seq": [ + 14000, + -1, + 600, + 600 + ] + }, + { + "before": 14000, + "out": 14000, + "seq": [ + -1, + -1, + -1, + -1, + -1, + -1, + -1, + -1 + ] + }, + { + "before": 14000, + "out": 14000, + "seq": [ + 14000, + 14000, + 14000, + 14000, + 14000, + 14000, + 14000, + 14000 + ] + }, + { + "before": 0, + "out": 0, + "seq": [ + 0, + 0, + 0 + ] + }, + { + "before": 9000, + "out": 8999, + "seq": [ + 9000, + 8999, + 8999 + ] + }, + { + "before": 9000, + "out": 100, + "seq": [ + -1, + 9000, + -1, + 9000, + 100, + -1, + 100 + ] + }, + { + "before": 5000, + "out": 7000, + "seq": [ + 6000, + 7000, + 7000, + 7000, + 7000, + 7000, + 7000, + 7000 + ] + }, + { + "before": 5000, + "out": 600, + "seq": [ + 4000, + 3000, + 2000, + 1000, + 900, + 800, + 700, + 600 + ] + } + ], + "last_error": { + "": "", + "[loki-rs engine] service loki-engine non démarré\n": "service loki-engine non démarré", + "[ok] démarré\n[attention] 2 processus du moteur toujours là\n": "[attention] 2 processus du moteur toujours là", + "ligne 1\n[loki-rs engine] moteur absent \n\n": "moteur absent", + "pas de préfixe": "pas de préfixe" + }, + "plain": { + "\u001b[1;36mcouleur\u001b[0m et \u001b[32mvert\u001b[0m": "couleur et vert", + "\u001b[mvide\u001b[0;1;2m": "vide", + "BIN non défini — installe un moteur : \u001b[1mloki llamacpp install\u001b[0m": "BIN non défini — installe un moteur : loki llamacpp install", + "rien à retirer": "rien à retirer" + }, + "preflight": [ + { + "name": "externe", + "config": { + "EXTERNAL": "1" + }, + "out": "" + }, + { + "name": "externe sans BIN ni MODEL", + "config": { + "BIN": "", + "EXTERNAL": " 1 " + }, + "out": "" + }, + { + "name": "strata sans config", + "config": { + "ENGINE": "strata" + }, + "out": "STRATA_CONFIG non défini : réinstalle le modèle depuis Paramètres → Moteur → Strata" + }, + { + "name": "strata config absente", + "config": { + "ENGINE": "Strata", + "STRATA_CONFIG": "{{HOME}}/strata/m/config.json" + }, + "out": "configuration de Strata introuvable : {{HOME}}/strata/m/config.json — réinstalle le modèle" + }, + { + "name": "strata sans venv", + "config": { + "ENGINE": "strata", + "STRATA_CONFIG": "{{HOME}}/strata/m/config.json" + }, + "files": [ + "strata/m/config.json" + ], + "out": "environnement Python de Strata absent ({{HOME}}/strata/m/.venv/bin/python) : réinstalle le modèle" + }, + { + "name": "strata prêt", + "config": { + "ENGINE": "moe", + "STRATA_CONFIG": "{{HOME}}/strata/m/config.json" + }, + "files": [ + "strata/m/config.json", + "strata/m/.venv/bin/python" + ], + "out": "" + }, + { + "name": "BIN vide", + "config": { + "MODEL": "m.gguf" + }, + "out": "BIN non défini — installe un moteur : loki llamacpp install (ou renseigne BIN avec loki edit)" + }, + { + "name": "BIN espaces", + "config": { + "BIN": " ", + "MODEL": "m.gguf" + }, + "out": "BIN non défini — installe un moteur : loki llamacpp install (ou renseigne BIN avec loki edit)" + }, + { + "name": "BIN relatif absent", + "config": { + "BIN": "bin/absent", + "MODEL": "m.gguf" + }, + "out": "moteur introuvable : {{HOME}}/bin/absent — relance loki llamacpp install" + }, + { + "name": "BIN absolu absent", + "config": { + "BIN": "{{HOME}}/ailleurs/llama-server", + "MODEL": "m.gguf" + }, + "out": "moteur introuvable : {{HOME}}/ailleurs/llama-server — relance loki llamacpp install" + }, + { + "name": "BIN dossier", + "config": { + "BIN": "bin/", + "MODEL": "m.gguf" + }, + "files": [ + "bin/" + ], + "out": "modèle introuvable : {{HOME}}/models/m.gguf — corrige MODEL avec loki edit" + }, + { + "name": "MODEL vide", + "config": { + "BIN": "bin/llama-server" + }, + "files": [ + "bin/llama-server" + ], + "out": "MODEL non défini — indique un .gguf avec loki edit (dossier des modèles : {{HOME}}/models)" + }, + { + "name": "MODEL absent", + "config": { + "BIN": "bin/llama-server", + "MODEL": "absent.gguf" + }, + "files": [ + "bin/llama-server" + ], + "out": "modèle introuvable : {{HOME}}/models/absent.gguf — corrige MODEL avec loki edit" + }, + { + "name": "MODEL relatif", + "config": { + "BIN": "{{HOME}}/bin/llama-server", + "MODEL": "m.gguf" + }, + "files": [ + "bin/llama-server", + "models/m.gguf" + ], + "out": "" + }, + { + "name": "MODEL remonte", + "config": { + "BIN": "bin/llama-server", + "MODEL": "../m.gguf" + }, + "files": [ + "bin/llama-server", + "models/m.gguf" + ], + "out": "" + }, + { + "name": "MODEL sous-dossier", + "config": { + "BIN": "bin/llama-server", + "MODEL": "sub/m.gguf" + }, + "files": [ + "bin/llama-server", + "models/m.gguf" + ], + "out": "" + }, + { + "name": "MODEL absolu hors dossiers", + "config": { + "BIN": "bin/llama-server", + "MODEL": "{{HOME}}/ailleurs/x.gguf" + }, + "files": [ + "bin/llama-server", + "ailleurs/x.gguf" + ], + "out": "" + }, + { + "name": "MODEL absolu pas gguf", + "config": { + "BIN": "bin/llama-server", + "MODEL": "{{HOME}}/ailleurs/x.bin" + }, + "files": [ + "bin/llama-server", + "ailleurs/x.bin" + ], + "out": "MODEL={{HOME}}/ailleurs/x.bin : le modèle doit être un fichier .gguf" + }, + { + "name": "MODEL absolu absent", + "config": { + "BIN": "bin/llama-server", + "MODEL": "{{HOME}}/ailleurs/y.gguf" + }, + "files": [ + "bin/llama-server" + ], + "out": "MODEL={{HOME}}/ailleurs/y.gguf : dossier non autorisé : {{HOME}}/ailleurs — ajoute-le dans « Dossiers de modèles »" + }, + { + "name": "MODEL dossier déclaré", + "config": { + "BIN": "bin/llama-server", + "MODEL": "d.gguf" + }, + "files": [ + "bin/llama-server", + "disque/d.gguf" + ], + "dirs": [ + "disque" + ], + "out": "" + }, + { + "name": "MODEL chemin dans dossier déclaré", + "config": { + "BIN": "bin/llama-server", + "MODEL": "{{HOME}}/disque/d.gguf" + }, + "files": [ + "bin/llama-server", + "disque/d.gguf" + ], + "dirs": [ + "disque" + ], + "out": "" + }, + { + "name": "tranches manquantes", + "config": { + "BIN": "bin/llama-server", + "MODEL": "big-00001-of-00003.gguf" + }, + "files": [ + "bin/llama-server", + "models/big-00001-of-00003.gguf", + "models/big-00003-of-00003.gguf" + ], + "out": "modèle incomplet : il manque big-00002-of-00003.gguf dans {{HOME}}/models — ce modèle tient en 3 fichiers, télécharge-les tous" + }, + { + "name": "tranches dossier", + "config": { + "BIN": "bin/llama-server", + "MODEL": "big-00001-of-00002.gguf" + }, + "files": [ + "bin/llama-server", + "models/big-00001-of-00002.gguf", + "models/big-00002-of-00002.gguf/" + ], + "out": "modèle incomplet : il manque big-00002-of-00002.gguf dans {{HOME}}/models — ce modèle tient en 2 fichiers, télécharge-les tous" + }, + { + "name": "tranches complètes", + "config": { + "BIN": "bin/llama-server", + "MODEL": "big-00001-of-00002.gguf" + }, + "files": [ + "bin/llama-server", + "models/big-00001-of-00002.gguf", + "models/big-00002-of-00002.gguf" + ], + "out": "" + }, + { + "name": "tranche 2 désignée", + "config": { + "BIN": "bin/llama-server", + "MODEL": "big-00002-of-00002.gguf" + }, + "files": [ + "bin/llama-server", + "models/big-00002-of-00002.gguf" + ], + "out": "modèle incomplet : il manque big-00001-of-00002.gguf dans {{HOME}}/models — ce modèle tient en 2 fichiers, télécharge-les tous" + }, + { + "name": "MODEL guillemets", + "config": { + "BIN": "bin/llama-server", + "MODEL": " m.gguf " + }, + "files": [ + "bin/llama-server", + "models/m.gguf" + ], + "out": "" + } + ], + "steps": [ + { + "name": "déchargement en GET", + "method": "GET", + "path": "/api/vram/unload", + "status": 405, + "allow": "POST", + "out": { + "error": "méthode GET refusée : POST attendu", + "ok": false + } + }, + { + "name": "rechargement en GET", + "method": "GET", + "path": "/api/vram/reload", + "status": 405, + "allow": "POST", + "out": { + "error": "méthode GET refusée : POST attendu", + "ok": false + } + }, + { + "name": "rechargement en PUT", + "method": "PUT", + "path": "/api/vram/reload", + "status": 405, + "allow": "POST", + "out": { + "error": "méthode PUT refusée : POST attendu", + "ok": false + } + }, + { + "name": "génération en cours", + "method": "POST", + "path": "/api/vram/unload", + "body": "{}", + "generating": true, + "status": 409, + "out": { + "error": "une réponse est en cours d'écriture — décharger maintenant l'interromprait", + "generating": true, + "ok": false + } + }, + { + "name": "génération en cours, corps illisible", + "method": "POST", + "path": "/api/vram/unload", + "body": "nimp", + "generating": true, + "status": 409, + "out": { + "error": "une réponse est en cours d'écriture — décharger maintenant l'interromprait", + "generating": true, + "ok": false + } + }, + { + "name": "moteur déjà arrêté", + "method": "POST", + "path": "/api/vram/unload", + "body": "{\"force\":true}", + "generating": true, + "status": 200, + "out": { + "active": false, + "before_mb": 0, + "error": "", + "freed_mb": 0, + "gpu": false, + "ok": true, + "used_mb": 0, + "was_active": false + } + }, + { + "name": "moteur arrêté, sans corps", + "method": "POST", + "path": "/api/vram/unload", + "status": 200, + "out": { + "active": false, + "before_mb": 0, + "error": "", + "freed_mb": 0, + "gpu": false, + "ok": true, + "used_mb": 0, + "was_active": false + } + }, + { + "name": "optimisation en cours", + "method": "POST", + "path": "/api/vram/reload", + "tune": true, + "status": 409, + "out": { + "error": "optimisation en cours (loki tune) : le moteur est réservé aux essais — attends la fin ou annule-la", + "ok": false, + "tuning": true + } + }, + { + "name": "préflight sans BIN", + "method": "POST", + "path": "/api/vram/reload", + "config": { + "MODEL": "m.gguf" + }, + "status": 200, + "out": { + "error": "BIN non défini — installe un moteur : loki llamacpp install (ou renseigne BIN avec loki edit)", + "ok": false + } + }, + { + "name": "préflight sans MODEL", + "method": "POST", + "path": "/api/vram/reload", + "config": { + "BIN": "/bin/sh" + }, + "status": 200, + "out": { + "error": "MODEL non défini — indique un .gguf avec loki edit (dossier des modèles : {{HOME}}/models)", + "ok": false + } + } + ] +} diff --git a/rust/loki-rs/tests/proxy.rs b/rust/loki-rs/tests/proxy.rs index da89148..6339327 100644 --- a/rust/loki-rs/tests/proxy.rs +++ b/rust/loki-rs/tests/proxy.rs @@ -107,6 +107,7 @@ async fn facade(backend: SocketAddr) -> (SocketAddr, oneshot::Sender<()>) { push: None, mem: None, presets: false, + engine_ctl: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await;