diff --git a/internal/loki/chat_conversation.go b/internal/loki/chat_conversation.go index 4929e55..70faff8 100644 --- a/internal/loki/chat_conversation.go +++ b/internal/loki/chat_conversation.go @@ -88,6 +88,8 @@ type Conversation struct { // errBenchBusy pour un bench, errTuneBusy pour l'optimiseur, qui prend le // même verrou (backend_tune_job.go). leaseWhy error + // extLease : le verrou est tenu par la façade Rust (web_internal_lease.go). + extLease *extLeaseState // File d'attente des messages envoyés PENDANT une génération (AJEAN // 0.14.0). Ils sont injectés dans le tour en cours à la prochaine frontière @@ -200,6 +202,7 @@ func LoadConversation() { // Une génération n'a pas pu survivre à l'arrêt du process : on repart propre. conv.Generating = false conv.cancel = nil + conv.extLeaseClearLocked() } // loadFrom remplace l'état en mémoire par une autre discussion (b vide = fil @@ -235,6 +238,7 @@ func (c *Conversation) loadFrom(id string, b []byte) { if !c.benching { c.Generating = false c.cancel = nil + c.extLeaseClearLocked() // le tour de la façade est rompu, comme un tour Go } c.epoch++ c.cond.Broadcast() @@ -1199,7 +1203,8 @@ func (c *Conversation) Reset() { c.benching = false // Stop a annulé le bench : le verrou est rendu ici c.leaseWhy = nil c.cancel = nil - c.queued = nil // les messages en attente visaient le fil qu'on vient de vider + c.extLeaseClearLocked() // Reset débloque aussi un tour de la façade + c.queued = nil // les messages en attente visaient le fil qu'on vient de vider c.cond.Broadcast() c.mu.Unlock() c.persist() diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index 3391d6b..1a7a869 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -75,6 +75,14 @@ func withInternalAPI(next http.Handler) http.Handler { handleInternalCrypt(w, r, true) case "/internal/v1/crypt/open": handleInternalCrypt(w, r, false) + case "/internal/v1/engine/lease": + handleInternalLease(w, r) + case "/internal/v1/engine/lease/renew": + handleInternalLeaseRenew(w, r) + case "/internal/v1/engine/lease/release": + handleInternalLeaseRelease(w, r) + case "/internal/v1/engine/stop": + handleInternalEngineStop(w, r) default: http.NotFound(w, r) } diff --git a/internal/loki/web_internal_lease.go b/internal/loki/web_internal_lease.go new file mode 100644 index 0000000..ee1eb1c --- /dev/null +++ b/internal/loki/web_internal_lease.go @@ -0,0 +1,196 @@ +package loki + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "time" +) + +// API interne du bail du moteur (lot 6 de la migration Rust). +// +// Le verrou de génération (conv.Generating) est partagé : tour de chat, tâche +// planifiée (RunAutonomous), bench et optimiseur (measureLease), et quatre +// sondes isGenerating (optimiseur, VRAM, projets, liste des discussions). Une +// conversation tenue par la façade Rust doit prendre CE verrou-là, sans quoi +// une tâche planifiée partirait en plein tour vers le même moteur. +// +// POST /internal/v1/engine/lease {"ttl_ms":N} → 200 {"lease":id,"ttl_ms":N} +// 409 {"error":…,"task_id":…,"task_name":…} +// POST /internal/v1/engine/lease/renew {"lease":id} → 200 {"stop":bool} · 410 bail perdu +// POST /internal/v1/engine/lease/release {"lease":id} → 204 (idempotent) +// POST /internal/v1/engine/stop → 204 : arrête le détenteur Go +// (tâche, bench, optimiseur) +// +// Le bail expire s'il n'est pas renouvelé : une façade tuée en plein tour ne +// laisse pas le moteur réservé pour toujours. Reset et les bascules de +// discussion côté Go le rompent (renew → 410), comme ils libèrent un tour Go. +// « stop » : le bouton stop est passé par le Go (conv.Stop) — la façade arrête +// son tour. + +const ( + extLeaseDefaultTTL = 30 * time.Second + extLeaseMinTTL = 5 * time.Second + extLeaseMaxTTL = 5 * time.Minute + internalLeaseMax = 4 << 10 +) + +// extLeaseState : le bail tenu par la façade. Protégé par conv.mu. +type extLeaseState struct { + id string + stop bool + ttl time.Duration + timer *time.Timer +} + +// extLeaseClearLocked oublie le bail de la façade. c.mu détenu. +func (c *Conversation) extLeaseClearLocked() { + if c.extLease != nil { + c.extLease.timer.Stop() + c.extLease = nil + } +} + +// extLeaseAcquire prend le verrou de génération pour la façade. +func (c *Conversation) extLeaseAcquire(ttl time.Duration) (string, error) { + if err := tuneGuard(); err != nil { + return "", err + } + c.mu.Lock() + defer c.mu.Unlock() + if c.Generating { + return "", ErrBusy + } + b, err := randBytes(16) + if err != nil { + return "", err + } + st := &extLeaseState{id: fmt.Sprintf("%x", b), ttl: ttl} + st.timer = time.AfterFunc(ttl, func() { c.extLeaseExpire(st) }) + c.Generating = true + c.cancel = func() { + c.mu.Lock() + st.stop = true + c.mu.Unlock() + } + c.extLease = st + return st.id, nil +} + +// extLeaseExpire rend le verrou d'un bail que la façade n'a pas renouvelé. +func (c *Conversation) extLeaseExpire(st *extLeaseState) { + c.mu.Lock() + defer c.mu.Unlock() + if c.extLease != st { + return + } + c.extLease = nil + c.Generating, c.cancel = false, nil + fmt.Fprintf(os.Stderr, "[lease] bail de la façade expiré sans renouvellement (%s) : verrou du moteur rendu\n", st.ttl) +} + +// extLeaseRenew prolonge le bail ; ok=false s'il n'est plus le bail courant. +func (c *Conversation) extLeaseRenew(id string) (stop, ok bool) { + c.mu.Lock() + defer c.mu.Unlock() + st := c.extLease + if st == nil || st.id != id { + return false, false + } + st.timer.Reset(st.ttl) + return st.stop, true +} + +// extLeaseRelease rend le verrou, si ce bail le tient encore. +func (c *Conversation) extLeaseRelease(id string) { + c.mu.Lock() + defer c.mu.Unlock() + if c.extLease == nil || c.extLease.id != id { + return + } + c.extLeaseClearLocked() + c.Generating, c.cancel = false, nil +} + +func readLeaseBody(w http.ResponseWriter, r *http.Request, v any) bool { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + sendJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "POST attendu"}) + return false + } + raw, err := io.ReadAll(io.LimitReader(r.Body, internalLeaseMax+1)) + if err != nil || len(raw) > internalLeaseMax { + sendJSON(w, http.StatusBadRequest, map[string]any{"error": "corps illisible"}) + return false + } + if len(raw) > 0 && json.Unmarshal(raw, v) != nil { + sendJSON(w, http.StatusBadRequest, map[string]any{"error": "JSON illisible"}) + return false + } + return true +} + +func handleInternalLease(w http.ResponseWriter, r *http.Request) { + var in struct { + TTLms int64 `json:"ttl_ms"` + } + if !readLeaseBody(w, r, &in) { + return + } + ttl := extLeaseDefaultTTL + if in.TTLms > 0 { + ttl = min(max(time.Duration(in.TTLms)*time.Millisecond, extLeaseMinTTL), extLeaseMaxTTL) + } + id, err := conv.extLeaseAcquire(ttl) + if err != nil { + if err == ErrBusy { + err = conv.busyReason() // dit QUI occupe le moteur + } + out := map[string]any{"error": err.Error()} + conv.mu.Lock() + if conv.runningTaskID != "" { + out["task_id"], out["task_name"] = conv.runningTaskID, conv.runningTaskName + } + conv.mu.Unlock() + sendJSON(w, http.StatusConflict, out) + return + } + sendJSON(w, http.StatusOK, map[string]any{"lease": id, "ttl_ms": ttl.Milliseconds()}) +} + +func handleInternalLeaseRenew(w http.ResponseWriter, r *http.Request) { + var in struct { + Lease string `json:"lease"` + } + if !readLeaseBody(w, r, &in) { + return + } + stop, ok := conv.extLeaseRenew(in.Lease) + if !ok { + sendJSON(w, http.StatusGone, map[string]any{"error": "bail perdu"}) + return + } + sendJSON(w, http.StatusOK, map[string]any{"stop": stop}) +} + +func handleInternalLeaseRelease(w http.ResponseWriter, r *http.Request) { + var in struct { + Lease string `json:"lease"` + } + if !readLeaseBody(w, r, &in) { + return + } + conv.extLeaseRelease(in.Lease) + w.WriteHeader(http.StatusNoContent) +} + +func handleInternalEngineStop(w http.ResponseWriter, r *http.Request) { + var in struct{} + if !readLeaseBody(w, r, &in) { + return + } + conv.Stop() + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/loki/web_internal_lease_test.go b/internal/loki/web_internal_lease_test.go new file mode 100644 index 0000000..1d0c442 --- /dev/null +++ b/internal/loki/web_internal_lease_test.go @@ -0,0 +1,153 @@ +package loki + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" +) + +func leaseJSON(t *testing.T, body string) map[string]any { + t.Helper() + var m map[string]any + if err := json.Unmarshal([]byte(body), &m); err != nil { + t.Fatalf("réponse illisible %q : %v", body, err) + } + return m +} + +// libérerVerrou rend le verrou de génération quoi qu'un test ait laissé. +func libérerVerrou(t *testing.T) { + t.Cleanup(func() { + conv.mu.Lock() + conv.extLeaseClearLocked() + conv.Generating, conv.benching, conv.cancel, conv.leaseWhy = false, false, nil, nil + conv.mu.Unlock() + }) +} + +// Le bail de la façade EST le verrou de génération : tant qu'il tient, une +// mesure (bench, optimiseur) ou un second bail est refusé. +func TestBailInterneVerrouPartage(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + libérerVerrou(t) + + w := appelOAIInterne(t, "POST", "/internal/v1/engine/lease", `{"ttl_ms":60000}`) + if w.Code != 200 { + t.Fatalf("bail : %d %s", w.Code, w.Body.String()) + } + m := leaseJSON(t, w.Body.String()) + id, _ := m["lease"].(string) + if len(id) != 32 || m["ttl_ms"] != float64(60000) { + t.Fatalf("bail : %v", m) + } + if !conv.isGenerating() { + t.Fatal("le bail ne tient pas le verrou de génération") + } + if _, err := conv.measureLease(func() {}, errBenchBusy); err != ErrBusy { + t.Fatalf("bench accepté pendant le bail : %v", err) + } + if w := appelOAIInterne(t, "POST", "/internal/v1/engine/lease", ""); w.Code != 409 || + !strings.Contains(w.Body.String(), ErrBusy.Error()) { + t.Fatalf("second bail : %d %s", w.Code, w.Body.String()) + } + + // Renouvelé ; le stop passé par le Go remonte à la façade. + w = appelOAIInterne(t, "POST", "/internal/v1/engine/lease/renew", `{"lease":"`+id+`"}`) + if w.Code != 200 || leaseJSON(t, w.Body.String())["stop"] != false { + t.Fatalf("renouvellement : %d %s", w.Code, w.Body.String()) + } + conv.Stop() + w = appelOAIInterne(t, "POST", "/internal/v1/engine/lease/renew", `{"lease":"`+id+`"}`) + if w.Code != 200 || leaseJSON(t, w.Body.String())["stop"] != true { + t.Fatalf("stop non transmis : %d %s", w.Code, w.Body.String()) + } + if w := appelOAIInterne(t, "POST", "/internal/v1/engine/lease/renew", `{"lease":"autre"}`); w.Code != 410 { + t.Fatalf("bail inconnu renouvelé : %d", w.Code) + } + + // Rendu : le verrou est libre, le bail est mort, rendre deux fois est sans effet. + for i := 0; i < 2; i++ { + if w := appelOAIInterne(t, "POST", "/internal/v1/engine/lease/release", `{"lease":"`+id+`"}`); w.Code != 204 { + t.Fatalf("libération : %d", w.Code) + } + } + if conv.isGenerating() { + t.Fatal("verrou toujours tenu après la libération") + } + if w := appelOAIInterne(t, "POST", "/internal/v1/engine/lease/renew", `{"lease":"`+id+`"}`); w.Code != 410 { + t.Fatalf("bail rendu renouvelé : %d", w.Code) + } +} + +// Un bail rendu après coup ne libère pas le verrou d'un autre détenteur. +func TestBailInterneNeLibereQueLeSien(t *testing.T) { + testHome(t) + libérerVerrou(t) + id, err := conv.extLeaseAcquire(time.Minute) + if err != nil { + t.Fatal(err) + } + conv.Reset() // rompt le bail, comme un tour Go + if _, ok := conv.extLeaseRenew(id); ok { + t.Fatal("bail encore valide après Reset") + } + release, err := conv.measureLease(func() {}, errBenchBusy) + if err != nil { + t.Fatalf("verrou non rendu par Reset : %v", err) + } + defer release() + conv.extLeaseRelease(id) + if !conv.isGenerating() { + t.Fatal("un bail périmé a libéré le verrou du bench") + } +} + +// Une façade tuée en plein tour ne garde pas le moteur : sans renouvellement, +// le bail expire et rend le verrou. +func TestBailInterneExpire(t *testing.T) { + testHome(t) + libérerVerrou(t) + id, err := conv.extLeaseAcquire(50 * time.Millisecond) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(5 * time.Second) + for conv.isGenerating() { + if time.Now().After(deadline) { + t.Fatal("bail jamais expiré") + } + time.Sleep(10 * time.Millisecond) + } + if _, ok := conv.extLeaseRenew(id); ok { + t.Fatal("bail expiré encore renouvelable") + } +} + +// Moteur tenu par le Go : le refus dit qui le tient, et /engine/stop l'arrête. +func TestBailInterneRefusEtStop(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + libérerVerrou(t) + ctx, cancel := context.WithCancel(context.Background()) + release, err := conv.measureLease(cancel, errBenchBusy) + if err != nil { + t.Fatal(err) + } + defer release() + w := appelOAIInterne(t, "POST", "/internal/v1/engine/lease", "") + if w.Code != 409 || leaseJSON(t, w.Body.String())["error"] != errBenchBusy.Error() { + t.Fatalf("refus : %d %s", w.Code, w.Body.String()) + } + if w := appelOAIInterne(t, "POST", "/internal/v1/engine/stop", ""); w.Code != 204 { + t.Fatalf("stop : %d", w.Code) + } + if ctx.Err() == nil { + t.Fatal("le bench n'a pas été arrêté") + } + if w := appelOAIInterne(t, "GET", "/internal/v1/engine/stop", ""); w.Code != 405 { + t.Fatalf("GET accepté : %d", w.Code) + } +}