diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index 6a239ad..f321725 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -63,6 +63,12 @@ func withInternalAPI(next http.Handler) http.Handler { handleInternalStore(w, r) case "/internal/v1/store/all": handleInternalStoreAll(w, r) + case "/internal/v1/oai/route": + handleInternalOAIRoute(w, r) + case "/internal/v1/oai/begin": + handleInternalOAIBegin(w, r) + case "/internal/v1/oai/end": + handleInternalOAIEnd(w, r) default: http.NotFound(w, r) } diff --git a/internal/loki/web_internal_oai.go b/internal/loki/web_internal_oai.go new file mode 100644 index 0000000..71da86b --- /dev/null +++ b/internal/loki/web_internal_oai.go @@ -0,0 +1,134 @@ +package loki + +import ( + "encoding/json" + "net/http" + "strings" + "sync" + "time" +) + +// API interne de la surface /v1 servie par la façade Rust (lot 5). +// +// Rust relaie désormais /v1 directement au moteur, mais les DÉCISIONS restent +// ici, parce qu'elles dépendent de l'état de la conversation que ce process +// tient en mémoire : +// +// - route : ce qu'il faut pour router (port du moteur, Strata, clé des +// complétions, second slot en service) ; +// - begin : la requête entre « en vol » (engineRequestBeginSide : le slot de +// la discussion n'est pas effacé pendant ce temps, un préchauffage est +// annulé si besoin), puis les refus benchmark / optimisation, puis la trace +// de perte de cache (perfNoteForeign) — dans l'ordre d'oaiHandler ; +// - end : la requête est terminée (fin du flux, ou client parti). +// +// Un ticket jamais rendu (façade tuée en plein flux) expire au bout de +// oaiTicketMaxAge : sans ça, le compteur de requêtes en vol resterait bloqué +// et le slot ne serait plus jamais isolé. + +const oaiTicketMaxAge = 6 * time.Hour + +type oaiTicket struct { + end func() + when time.Time +} + +var oaiTickets = struct { + mu sync.Mutex + next uint64 + m map[uint64]oaiTicket +}{m: map[uint64]oaiTicket{}} + +func handleInternalOAIRoute(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + method, path := q.Get("method"), q.Get("path") + key, err := effectiveAPIKeyErr() + out := map[string]any{ + "port": LLMPort(), + "strata": strataActive(), + "key": key, + "key_ok": err == nil, + "max_body": sideSlotProxyMaxBody, + } + // La sonde du second slot interroge le moteur (avec cache) : seulement + // pour une complétion, comme sideSlotProxyRewrite. + out["side_live"] = method == http.MethodPost && strings.HasPrefix(path, "/v1/") && + sideSlotLive(resolveChatEndpoint()) + sendJSON(w, http.StatusOK, out) +} + +func handleInternalOAIBegin(w http.ResponseWriter, r *http.Request) { + var in struct { + Read bool `json:"read"` // GET/HEAD : ne calcule rien dans un slot + Side bool `json:"side"` // corps routé vers le second slot + PostV1 bool `json:"post_v1"` // complétion d'un client externe + } + if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&in) != nil { + sendJSON(w, http.StatusBadRequest, map[string]any{"error": "POST JSON attendu"}) + return + } + var end func() + if in.Read { + end, _ = engineRequestBeginSide(func(*prewarmRun) bool { return true }, true) + } else { + end, _ = engineRequestBeginSide(nil, in.Side) + } + if in.PostV1 { + refuse := func(retry, msg, code string) { + end() + sendJSON(w, http.StatusOK, map[string]any{"refuse": map[string]any{ + "status": http.StatusServiceUnavailable, "message": msg, + "type": "server_error", "code": code, "retry_after": retry, + }}) + } + if benchRunning() { + refuse("30", "benchmark en cours sur ce moteur — réessaie dans un moment", "benchmark_running") + return + } + if tuneGuard() != nil { + refuse("60", "optimisation du moteur en cours — réessaie dans quelques minutes", "tuning_running") + return + } + perfNoteForeign() + } + now := time.Now() + oaiTickets.mu.Lock() + for id, t := range oaiTickets.m { + if now.Sub(t.when) > oaiTicketMaxAge { + t.end() + delete(oaiTickets.m, id) + } + } + oaiTickets.next++ + id := oaiTickets.next + oaiTickets.m[id] = oaiTicket{end: end, when: now} + oaiTickets.mu.Unlock() + sendJSON(w, http.StatusOK, map[string]any{"ticket": id}) +} + +func handleInternalOAIEnd(w http.ResponseWriter, r *http.Request) { + var in struct { + Ticket uint64 `json:"ticket"` + } + if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&in) != nil { + sendJSON(w, http.StatusBadRequest, map[string]any{"error": "POST JSON attendu"}) + return + } + oaiTickets.mu.Lock() + t, ok := oaiTickets.m[in.Ticket] + delete(oaiTickets.m, in.Ticket) + oaiTickets.mu.Unlock() + if !ok { + http.NotFound(w, r) + return + } + t.end() + w.WriteHeader(http.StatusNoContent) +} + +// oaiInflight : nombre de tickets ouverts (tests). +func oaiInflight() int { + oaiTickets.mu.Lock() + defer oaiTickets.mu.Unlock() + return len(oaiTickets.m) +} diff --git a/internal/loki/web_internal_oai_test.go b/internal/loki/web_internal_oai_test.go new file mode 100644 index 0000000..3c5190f --- /dev/null +++ b/internal/loki/web_internal_oai_test.go @@ -0,0 +1,105 @@ +package loki + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func appelOAIInterne(t *testing.T, method, path, body string) *httptest.ResponseRecorder { + t.Helper() + r := httptest.NewRequest(method, path, strings.NewReader(body)) + r.Header.Set(internalTokenHeader, jetonTest) + w := httptest.NewRecorder() + withInternalAPI(http.NotFoundHandler()).ServeHTTP(w, r) + return w +} + +func enVol() int { + engineGate.mu.Lock() + defer engineGate.mu.Unlock() + return engineGate.inflight +} + +// Une requête /v1 relayée par Rust compte « en vol » de begin à end, comme +// le defer engineProxyBegin(r, side)() d'oaiHandler. +func TestOAIInterneTicket(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + avant := enVol() + w := appelOAIInterne(t, "POST", "/internal/v1/oai/begin", `{"read":false,"side":false,"post_v1":true}`) + var got struct { + Ticket uint64 `json:"ticket"` + Refuse map[string]any `json:"refuse"` + } + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || got.Ticket == 0 || got.Refuse != nil { + t.Fatalf("begin : %d %s", w.Code, w.Body.String()) + } + if enVol() != avant+1 || oaiInflight() != 1 { + t.Fatalf("en vol : %d (avant %d), tickets %d", enVol(), avant, oaiInflight()) + } + body, _ := json.Marshal(map[string]any{"ticket": got.Ticket}) + if w := appelOAIInterne(t, "POST", "/internal/v1/oai/end", string(body)); w.Code != 204 { + t.Fatalf("end : %d", w.Code) + } + if enVol() != avant || oaiInflight() != 0 { + t.Fatalf("après end : en vol %d, tickets %d", enVol(), oaiInflight()) + } + if w := appelOAIInterne(t, "POST", "/internal/v1/oai/end", string(body)); w.Code != 404 { + t.Fatalf("second end : %d", w.Code) + } +} + +// Pendant un benchmark, une complétion externe est refusée — et ne reste pas +// comptée en vol. +func TestOAIInterneRefusBenchmark(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + benchJob.mu.Lock() + benchJob.running = true + benchJob.mu.Unlock() + t.Cleanup(func() { + benchJob.mu.Lock() + benchJob.running = false + benchJob.mu.Unlock() + }) + avant := enVol() + w := appelOAIInterne(t, "POST", "/internal/v1/oai/begin", `{"post_v1":true}`) + var got struct { + Refuse struct { + Status int `json:"status"` + Code string `json:"code"` + RetryAfter string `json:"retry_after"` + } `json:"refuse"` + } + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.Refuse.Status != 503 || got.Refuse.Code != "benchmark_running" || got.Refuse.RetryAfter != "30" { + t.Fatalf("refus : %s", w.Body.String()) + } + if enVol() != avant || oaiInflight() != 0 { + t.Fatalf("refus compté en vol : %d", enVol()) + } +} + +func TestOAIInterneRoute(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + if err := putStr(bkConfig, "PORT", "8099"); err != nil { + t.Fatal(err) + } + if err := putStr(bkState, "api_key", "cle-oai"); err != nil { + t.Fatal(err) + } + w := appelOAIInterne(t, "GET", "/internal/v1/oai/route?method=GET&path=/v1/models", "") + var got map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got["port"] != float64(8099) || got["key"] != "cle-oai" || got["key_ok"] != true || got["side_live"] != false { + t.Fatalf("route : %s", w.Body.String()) + } +}