diff --git a/internal/loki/chat_export.go b/internal/loki/chat_export.go index 306fbf7..36456af 100644 --- a/internal/loki/chat_export.go +++ b/internal/loki/chat_export.go @@ -378,11 +378,12 @@ func exportFilename(ext string) string { // exportBody rend la conversation selon les options. Renvoie le corps, son type // MIME et l'extension de fichier. func exportBody(o exportOpts) ([]byte, string, string, error) { + src := exportSource() // la façade peut tenir la conversation (facade_chat.go) if o.Format == "json" { - b, err := conv.ExportJSON(o) + b, err := src.ExportJSON(o) return b, "application/json; charset=utf-8", "json", err } - return []byte(conv.ExportMarkdown(o)), "text/markdown; charset=utf-8", "md", nil + return []byte(src.ExportMarkdown(o)), "text/markdown; charset=utf-8", "md", nil } // cmdExport écrit la conversation dans un fichier (ou sur la sortie standard). diff --git a/internal/loki/code_criteria.go b/internal/loki/code_criteria.go index ae19082..eb7830c 100644 --- a/internal/loki/code_criteria.go +++ b/internal/loki/code_criteria.go @@ -214,14 +214,15 @@ func toolCriteria(args map[string]any, allowPass bool) string { // conversation. Best-effort : hors génération il n'y a pas d'epoch en cours, // on publie sur l'epoch courant. func critNotify(list []Criterion) { - if conv == nil { + if list == nil { + list = []Criterion{} + } + delta := map[string]any{"criteria": list, "ts": time.Now().UnixMilli()} + if facadeDelta(delta) || conv == nil { return } conv.mu.Lock() epoch := conv.epoch conv.mu.Unlock() - if list == nil { - list = []Criterion{} - } - conv.appendDelta(epoch, map[string]any{"criteria": list, "ts": time.Now().UnixMilli()}) + conv.appendDelta(epoch, delta) } diff --git a/internal/loki/code_mode.go b/internal/loki/code_mode.go index 2d4ae37..4a51181 100644 --- a/internal/loki/code_mode.go +++ b/internal/loki/code_mode.go @@ -132,10 +132,12 @@ func handleChatMode(w http.ResponseWriter, r *http.Request) { } // Pousse la bascule à tous les appareils abonnés (le sélecteur doit // suivre, comme le fil lui-même). - conv.mu.Lock() - epoch := conv.epoch - conv.mu.Unlock() - conv.appendDelta(epoch, map[string]any{"mode": body.Mode}) + if !facadeDelta(map[string]any{"mode": body.Mode}) { + conv.mu.Lock() + epoch := conv.epoch + conv.mu.Unlock() + conv.appendDelta(epoch, map[string]any{"mode": body.Mode}) + } sendJSON(w, 200, map[string]any{"ok": true, "mode": body.Mode}) return } diff --git a/internal/loki/facade_chat.go b/internal/loki/facade_chat.go new file mode 100644 index 0000000..af9f266 --- /dev/null +++ b/internal/loki/facade_chat.go @@ -0,0 +1,203 @@ +package loki + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httputil" + "net/url" + "os" + "strings" + "sync" + "time" +) + +// facade_chat.go — conversation tenue par la façade Rust (LOKI_CHAT=rust, lot 6). +// +// La façade sert elle-même les routes du chat sur le port public. Deux chemins +// arrivent pourtant ICI sans passer par elle : le tunnel du relais, qui sert ce +// mux nu (/api/e2e/*), et sa redistribution /api/e2e/req dans le mux local. Ils +// liraient la conversation du Go — périmée — et pourraient même l'écrire en +// base par-dessus celle de la façade. Quand la façade tient la conversation : +// +// - les routes du chat lui sont remises, authentifiées par le jeton interne +// (la requête l'a déjà été ici) ; +// - /api/e2e/chat lit son flux et le rechiffre événement par événement ; +// - les événements d'affichage nés ici (bascule du mode Code, critères) vont +// dans SON journal ; +// - l'export lit la discussion enregistrée ; +// - le déverrouillage de la mémoire lui fait recharger la discussion. + +const ( + facadeChatEnv = "LOKI_CHAT" + facadeAddrEnv = "LOKI_FACADE_ADDR" +) + +// facadeChat : la façade tient-elle la conversation ? Il faut aussi son +// adresse et le jeton interne, sans lesquels rien ne peut lui être remis. +func facadeChat() bool { + return strings.EqualFold(strings.TrimSpace(os.Getenv(facadeChatEnv)), "rust") && + facadeAddr() != "" && internalToken() != "" +} + +func facadeAddr() string { return strings.TrimSpace(os.Getenv(facadeAddrEnv)) } + +// facadeChatRoutes : les routes que la façade sert quand elle tient la +// conversation (chatapi.rs). Le reste (mode Code, critères, pièces jointes, +// fichiers, export) reste ici. +var facadeChatRoutes = map[string]bool{ + "/api/chat": true, + "/api/chat/send": true, + "/api/chat/stop": true, + "/api/chat/reset": true, + "/api/chat/state": true, + "/api/chat/compact": true, + "/api/chat/tool-result": true, + "/api/conversations": true, + "/api/conversations/new": true, + "/api/conversations/switch": true, + "/api/conversations/rename": true, + "/api/conversations/delete": true, +} + +// withFacadeChat remet la requête à la façade quand elle tient la +// conversation. 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 { + if !facadeChatRoutes[path] { + return h + } + return func(w http.ResponseWriter, r *http.Request) { + if !facadeChat() { + h(w, r) + return + } + facadeRelay(w, r) + } +} + +// facadeRelay : relais transparent (flux compris) vers la façade. +func facadeRelay(w http.ResponseWriter, r *http.Request) { + rp := &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { + pr.SetURL(&url.URL{Scheme: "http", Host: facadeAddr()}) + pr.Out.Header.Set(internalTokenHeader, internalToken()) + }, + FlushInterval: -1, // SSE : chaque trame part aussitôt + ErrorHandler: func(w http.ResponseWriter, _ *http.Request, err error) { + sendJSON(w, http.StatusBadGateway, map[string]any{"ok": false, "error": "conversation indisponible : " + err.Error()}) + }, + } + rp.ServeHTTP(w, r) +} + +// facadeClient : appels courts vers l'API interne de la façade. +var facadeClient = &http.Client{Timeout: 5 * time.Second} + +// facadePost appelle l'API interne de la façade. +func facadePost(path string, body any) error { + b, err := json.Marshal(body) + if err != nil { + return err + } + req, err := http.NewRequest(http.MethodPost, "http://"+facadeAddr()+path, bytes.NewReader(b)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set(internalTokenHeader, internalToken()) + resp, err := facadeClient.Do(req) + if err != nil { + return err + } + resp.Body.Close() + if resp.StatusCode >= 300 { + return fmt.Errorf("façade : %s", resp.Status) + } + return nil +} + +// facadeDelta : un événement d'affichage né ici, journalisé par la façade +// (tous les appareils le reçoivent). Rend false si la façade ne tient pas la +// conversation : l'appelant journalise alors comme avant. +func facadeDelta(delta map[string]any) bool { + if !facadeChat() { + return false + } + if err := facadePost("/internal/facade/v1/delta", delta); err != nil { + fmt.Fprintf(os.Stderr, "[chat] événement non transmis à la façade : %v\n", err) + } + return true +} + +// facadeReload : la mémoire vient d'être déverrouillée ; la façade recharge la +// discussion si elle l'avait chargée vide. +func facadeReload() { + if !facadeChat() { + return + } + go func() { + if err := facadePost("/internal/facade/v1/reload", map[string]any{}); err != nil { + fmt.Fprintf(os.Stderr, "[chat] rechargement de la discussion par la façade : %v\n", err) + } + }() +} + +// facadeChatStream : /api/e2e/chat quand la façade tient la conversation. Son +// flux est lu ligne par ligne et chaque événement repart par out (scellé, un +// sceau par événement). Les battements de la façade sont ignorés : ce flux a +// les siens (sseHeartbeat). +func facadeChatStream(ctx context.Context, body []byte, out *sseStream) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://"+facadeAddr()+"/api/chat", bytes.NewReader(body)) + if err != nil { + return + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set(internalTokenHeader, internalToken()) + resp, err := http.DefaultClient.Do(req) // sans délai : un flux dure ce qu'il dure + if err != nil { + if ctx.Err() == nil { + out.emit(map[string]any{"error": "conversation indisponible : " + err.Error()}) + } + return + } + defer resp.Body.Close() + sc := bufio.NewScanner(resp.Body) + sc.Buffer(make([]byte, 0, 64<<10), 64<<20) + for sc.Scan() { + data, ok := strings.CutPrefix(sc.Text(), "data: ") + if !ok { + continue + } + var ev struct { + Choices []struct { + Delta map[string]any `json:"delta"` + } `json:"choices"` + } + dec := json.NewDecoder(strings.NewReader(data)) + dec.UseNumber() // les nombres repartent tels quels + if dec.Decode(&ev) != nil || len(ev.Choices) == 0 { + continue + } + if !out.emit(ev.Choices[0].Delta) { + return + } + } +} + +// exportSource : la conversation à exporter. Tenue par la façade, celle du Go +// est périmée : on lit la discussion enregistrée (à jour à chaque fin de tour). +func exportSource() *Conversation { + if !facadeChat() { + return conv + } + c := &Conversation{} + c.cond = sync.NewCond(&c.mu) + if b := storeBytes(bkChat, convKey(convEnsureActive())); len(b) > 0 { + _ = json.Unmarshal(b, c) + } + return c +} diff --git a/internal/loki/facade_chat_test.go b/internal/loki/facade_chat_test.go new file mode 100644 index 0000000..a27aa1b --- /dev/null +++ b/internal/loki/facade_chat_test.go @@ -0,0 +1,141 @@ +package loki + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +// fausseFacade : enregistre ce qui lui arrive ; /api/chat rend un flux. +type fausseFacade struct { + mu sync.Mutex + vu []string // "METHODE chemin jeton-ok corps" + srv *httptest.Server + t *testing.T + sse string + reply string +} + +func demarrerFacade(t *testing.T) *fausseFacade { + t.Helper() + f := &fausseFacade{t: t, reply: `{"ok":true,"de":"facade"}`} + f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + tok := r.Header.Get(internalTokenHeader) == jetonTest + f.mu.Lock() + f.vu = append(f.vu, r.Method+" "+r.URL.Path+" "+map[bool]string{true: "jeton", false: "sans-jeton"}[tok]+" "+string(b)) + f.mu.Unlock() + if r.URL.Path == "/api/chat" { + w.Header().Set("Content-Type", "text/event-stream") + io.WriteString(w, f.sse) + return + } + if strings.HasPrefix(r.URL.Path, "/internal/facade/") { + w.WriteHeader(http.StatusNoContent) + return + } + io.WriteString(w, f.reply) + })) + t.Cleanup(f.srv.Close) + t.Setenv(internalTokenEnv, jetonTest) + t.Setenv(facadeChatEnv, "rust") + t.Setenv(facadeAddrEnv, strings.TrimPrefix(f.srv.URL, "http://")) + return f +} + +func (f *fausseFacade) recu() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.vu...) +} + +// Les routes du chat arrivées ici (tunnel, redistribution E2E) partent à la +// façade avec le jeton interne ; les autres restent ici ; sans la variable, +// rien ne change. +func TestFacadeRelaieLesRoutesDuChat(t *testing.T) { + testHome(t) + f := demarrerFacade(t) + local := func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "go") } + + w := httptest.NewRecorder() + withFacadeChat("/api/chat/send", local)(w, httptest.NewRequest("POST", "/api/chat/send", strings.NewReader(`{"message":"salut"}`))) + if w.Body.String() != f.reply { + t.Fatalf("réponse : %q", w.Body.String()) + } + if got := f.recu(); len(got) != 1 || got[0] != `POST /api/chat/send jeton {"message":"salut"}` { + t.Fatalf("reçu par la façade : %v", got) + } + // Mode Code : reste au Go. + w = httptest.NewRecorder() + withFacadeChat("/api/chat/mode", local)(w, httptest.NewRequest("POST", "/api/chat/mode", nil)) + if w.Body.String() != "go" { + t.Fatalf("mode relayé : %q", w.Body.String()) + } + // Sans LOKI_CHAT=rust : tout reste ici. + t.Setenv(facadeChatEnv, "") + w = httptest.NewRecorder() + withFacadeChat("/api/chat/send", local)(w, httptest.NewRequest("POST", "/api/chat/send", nil)) + if w.Body.String() != "go" || len(f.recu()) != 1 { + t.Fatalf("relayé sans la variable : %q", w.Body.String()) + } +} + +// /api/e2e/chat suit le journal de la façade : un événement par trame, les +// battements de la façade ignorés, les nombres repartent tels quels. +func TestFacadeFluxE2E(t *testing.T) { + testHome(t) + f := demarrerFacade(t) + f.sse = "data: {\"choices\":[{\"delta\":{\"pad\":\"··\"}}]}\n\n" + + ": ping\n\n" + + "data: {\"choices\":[{\"delta\":{\"seq\":12345678901,\"user\":\"salut\"}}]}\n\n" + + "data: {\"choices\":[{\"delta\":{\"caught_up\":true,\"id\":\"c1\"}}]}\n\n" + w := httptest.NewRecorder() + var trames []string + out := &sseStream{w: w, flusher: w, mu: &sync.Mutex{}, frame: func(obj map[string]any) ([]byte, bool) { + b, _ := json.Marshal(obj) + trames = append(trames, string(b)) + return []byte("data: " + string(b) + "\n\n"), true + }} + facadeChatStream(t.Context(), []byte(`{"from":0,"tail":20}`), out) + want := []string{`{"pad":"··"}`, `{"seq":12345678901,"user":"salut"}`, `{"caught_up":true,"id":"c1"}`} + if strings.Join(trames, "|") != strings.Join(want, "|") { + t.Fatalf("trames : %v", trames) + } + if got := f.recu(); len(got) != 1 || got[0] != `POST /api/chat jeton {"from":0,"tail":20}` { + t.Fatalf("abonnement : %v", got) + } +} + +// Les événements d'affichage nés ici vont au journal de la façade, pas à +// celui du Go. +func TestFacadeEvenementsEtExport(t *testing.T) { + testHome(t) + f := demarrerFacade(t) + resetConvForTest() + critNotify([]Criterion{{ID: 1, Text: "les tests passent", Status: "pending"}}) + got := f.recu() + if len(got) != 1 || !strings.HasPrefix(got[0], "POST /internal/facade/v1/delta jeton ") || !strings.Contains(got[0], "les tests passent") { + t.Fatalf("critères : %v", got) + } + conv.mu.Lock() + n := len(conv.Log) + conv.mu.Unlock() + if n != 0 { + t.Fatal("événement journalisé aussi dans la conversation du Go") + } + + // Export : la discussion enregistrée, pas la conversation du Go. + id := convEnsureActive() + b, _ := json.Marshal(map[string]any{"messages": []Message{{Role: "user", Content: "fil de la façade"}}, "log": []any{}, "seq": 0, "ctx_used": 0}) + if err := putStoreBytes(bkChat, convKey(id), b); err != nil { + t.Fatal(err) + } + body, _, _, err := exportBody(exportOpts{Format: "json"}) + if err != nil || !strings.Contains(string(body), "fil de la façade") { + t.Fatalf("export : %v %s", err, body) + } +} diff --git a/internal/loki/relay_e2e.go b/internal/loki/relay_e2e.go index 75dabee..267b3cc 100644 --- a/internal/loki/relay_e2e.go +++ b/internal/loki/relay_e2e.go @@ -174,7 +174,13 @@ func handleE2EChat(w http.ResponseWriter, r *http.Request) { sealedEv := append(nonce, gcm.Seal(nil, nonce, b, nil)...) return []byte("data: " + base64.StdEncoding.EncodeToString(sealedEv) + "\n\n"), true } - runChatStream(r.Context(), body, &sseStream{w: w, flusher: flusher, mu: mu, frame: frame}) + out := &sseStream{w: w, flusher: flusher, mu: mu, frame: frame} + // Conversation tenue par la façade : c'est SON journal qu'on suit. + if facadeChat() { + facadeChatStream(r.Context(), plain, out) + return + } + runChatStream(r.Context(), body, out) } // handleE2EReq : proxy de CONTRÔLE chiffré de bout en bout. Même enveloppe que le diff --git a/internal/loki/web_api.go b/internal/loki/web_api.go index fafdf6e..7ec0545 100644 --- a/internal/loki/web_api.go +++ b/internal/loki/web_api.go @@ -945,6 +945,7 @@ func handleMemUnlock(w http.ResponseWriter, r *http.Request) { // qu'une version précédente chiffrait par erreur (voir plainStoreKey). healPlainStoreKeys() reloadEncryptedStores() // recharge la discussion chiffrée en RAM + facadeReload() // … ou celle de la façade, si c'est elle qui la tient // Migration douce : chiffre ce qui serait resté en clair (discussions, // blocs archivés, trackers). Idempotent — n'encode que ce qui ne l'est pas. if memEncActive() && memUnlocked() { diff --git a/internal/loki/web_internal_tool.go b/internal/loki/web_internal_tool.go index e48a752..edaae25 100644 --- a/internal/loki/web_internal_tool.go +++ b/internal/loki/web_internal_tool.go @@ -58,29 +58,10 @@ func handleInternalToolExec(w http.ResponseWriter, r *http.Request) { sendJSON(w, http.StatusOK, struct { Label string `json:"label"` toolExecResult - Image string `json:"image,omitempty"` - Shot string `json:"shot,omitempty"` - Relay *Message `json:"relay,omitempty"` - Criteria []Criterion `json:"criteria,omitempty"` - }{label, res, image, shot, relay, criteriaAfter(in.Name, in.Args, res)}) -} - -// criteriaAfter : la liste des critères après un appel qui l'a changée -// (critNotify la pousse au journal du Go ; une conversation tenue par la -// façade la reçoit ici). nil sinon. -func criteriaAfter(name string, args map[string]any, res toolExecResult) []Criterion { - if name != "criteria" || strings.HasPrefix(res.Result, "[erreur]") { - return nil - } - switch a, _ := args["action"].(string); a { - case "add", "set", "clear": - list := critList(convActiveID()) - if list == nil { - list = []Criterion{} - } - return list - } - return nil + Image string `json:"image,omitempty"` + Shot string `json:"shot,omitempty"` + Relay *Message `json:"relay,omitempty"` + }{label, res, image, shot, relay}) } // toolStepExtras : ce que runChatLoop tire d'un résultat d'outil après diff --git a/internal/loki/web_internal_tool_test.go b/internal/loki/web_internal_tool_test.go index 2dbc9a7..a7efc80 100644 --- a/internal/loki/web_internal_tool_test.go +++ b/internal/loki/web_internal_tool_test.go @@ -84,19 +84,3 @@ func TestOutilInterneSuitesDuResultat(t *testing.T) { t.Fatalf("aperçu après une erreur : %q", shot) } } - -// Les critères changés par l'outil reviennent avec son résultat : une -// conversation tenue par la façade les pousse à son journal. -func TestOutilInterneCriteres(t *testing.T) { - testHome(t) - t.Setenv(internalTokenEnv, jetonTest) - got := appelOutil(t, `{"name":"criteria","args":{"action":"add","texts":["les tests passent"]},"caps":{"Agent":true,"Code":true}}`) - list, _ := got["criteria"].([]any) - if len(list) != 1 { - t.Fatalf("critères après add : %v", got) - } - got = appelOutil(t, `{"name":"criteria","args":{"action":"list"},"caps":{"Agent":true,"Code":true}}`) - if _, ok := got["criteria"]; ok { - t.Fatalf("une lecture ne pousse rien : %v", got) - } -} diff --git a/internal/loki/web_server.go b/internal/loki/web_server.go index e137bea..37640a3 100644 --- a/internal/loki/web_server.go +++ b/internal/loki/web_server.go @@ -217,7 +217,9 @@ func newWebMux() *http.ServeMux { // clé, celle des complétions (web_auth.go). mountOAI(mux) // api enregistre une route /api/* protégée par la clé de pilotage (web_auth.go). - api := func(path string, h http.HandlerFunc) { mux.HandleFunc(path, requireWebAuth(h)) } + // withFacadeChat : routes du chat remises à la façade quand elle tient la + // conversation (LOKI_CHAT=rust, facade_chat.go) ; sans effet sinon. + api := func(path string, h http.HandlerFunc) { mux.HandleFunc(path, requireWebAuth(withFacadeChat(path, h))) } api("/api/ping", handlePing) api("/api/status", handleStatus) api("/api/transcribe", handleTranscribe) // dictée vocale (Parakeet via sherpa-onnx, local)