diff --git a/README.md b/README.md index ff933db..7529515 100644 --- a/README.md +++ b/README.md @@ -227,7 +227,7 @@ redémarrages (volume `/data`). Variables d'environnement du conteneur : | `LOKI_AUTO_UPDATE` | `0` : pas de mise à jour des binaires au démarrage | `1` | | `LOKI_FACADE` | `0` : sans façade Rust, le Go sert directement le port public (porte de sortie) | `1` | | `LOKI_GO_PORT` | port local (127.0.0.1) du serveur Go derrière la façade | `8091` | -| `LOKI_CHAT` | qui tient la conversation : la façade Rust par défaut, outils fichiers, shell et git compris, avec l'export, le mode Code et les critères de la discussion ; le Go garde les autres outils et le reste du mode Code. La façade sonde aussi le gabarit du moteur après ses complétions (sonde de gabarit, `NUDGE_IN_TOOL`), range la télémétrie de ses complétions et ses requêtes en vol auprès du Go, honore les motifs de relance personnalisés (`retry_patterns`) et les slots du moteur (`SIDE_SLOT`, `SLOT_PERSIST`, isolation après la vérification). `go` : la rend au Go (porte de sortie). Avec `PROJ_SNAPSHOT`, `KEEP_TURN_IMAGES` ou `PREWARM`, le Go la garde de toute façon (décidé au démarrage) | Rust | +| `LOKI_CHAT` | qui tient la conversation : la façade Rust par défaut, outils fichiers, shell et git compris, avec l'export, le mode Code et les critères de la discussion ; le Go garde les autres outils et le reste du mode Code. La façade sonde aussi le gabarit du moteur après ses complétions (sonde de gabarit, `NUDGE_IN_TOOL`), range la télémétrie de ses complétions et ses requêtes en vol auprès du Go, honore les motifs de relance personnalisés (`retry_patterns`) et les slots du moteur (`SIDE_SLOT`, `SLOT_PERSIST`, isolation après la vérification) et les images gardées (`KEEP_TURN_IMAGES`). `go` : la rend au Go (porte de sortie). Avec `PROJ_SNAPSHOT` ou `PREWARM`, le Go la garde de toute façon (décidé au démarrage) | Rust | | `LOKI_MODELS` | `rust` : les routes des modèles — liste, dossiers, suppression, téléchargements, catalogue Hugging Face — sont servies par la façade Rust (expérimental, migration en cours) | Go | | `LOKI_DICTATE` | `rust` : la dictée vocale (`/api/transcribe`, `/api/dictate/*`) et son serveur de transcription sont tenus par la façade Rust (expérimental, migration en cours) | Go | | `LOKI_SETTINGS` | `rust` : les réglages partagés — préférences d'affichage, prompt système, intensité du raisonnement, adresse publique — sont servis par la façade Rust (expérimental, migration en cours) | Go | diff --git a/docs/superpowers/specs/2026-10-16-rust-migration-lot32-images-gardees-design.md b/docs/superpowers/specs/2026-10-16-rust-migration-lot32-images-gardees-design.md new file mode 100644 index 0000000..254406a --- /dev/null +++ b/docs/superpowers/specs/2026-10-16-rust-migration-lot32-images-gardees-design.md @@ -0,0 +1,97 @@ +# Lot 32 — Images gardées (KEEP_TURN_IMAGES) dans le tour de la façade + +## Contexte + +Sans la clé, une image montrée par un outil (capture, `see_image`) est +éphémère. Le modèle la voit pendant le tour, puis elle disparaît de +l'historique. Le tour suivant renvoie donc un fil où elle manque : le +préfixe en cache diverge à elle, et toute la boucle d'outils qui suivait est +recalculée. + +`KEEP_TURN_IMAGES` (`chat_keep_images.go`) garde l'image telle qu'envoyée, +sous conditions : + +- vision réellement active ; +- image rangée par référence et relue à l'octet près ; +- coût **mesuré** par le moteur, jamais estimé ; +- toutes les images gardées tiennent dans 10 % de la fenêtre. + +Les images gardées partent avant tout résumé. La clé gardait la conversation +au Go. + +## Le lot + +### Go (`facade_keepimg.go`) + +- `/internal/v1/turn/keepimg` rend deux choses : + - si la vision est réellement active (`visionEnabled`, `engineSeesImages`) ; + - le budget des images gardées (`keepImgBudget`). +- `/internal/v1/turn/keepimg/store` range une image relayée par référence + (`refRelayImage`) : + - fichier chiffré si la mémoire l'est ; + - date rafraîchie ; + - relu à l'octet près. + + Il rend le message qui la porte ainsi et son poids en octets, ou un refus. +- `KEEP_TURN_IMAGES` ne garde plus la conversation. + +### Rust + +- `chat/keepimg.rs` porte l'état du tour (`keepImages`) : + - `relay` : image gardée à l'essai, dans `messages` et `extra` ; + - `sending` ; + - `observe` : mesure de la requête qui l'a montrée la première, partage au + prorata des octets, seuil de 16 jetons, budget ; + - `rewritten` ; + - `publishable` ; + - `finish` : une image à l'essai jamais mesurée redevient éphémère. + + Une image refusée arrête le tour pour les suivantes. +- La boucle du tour : + - `TurnHost::keep_images` donne le budget, mais seulement pour le fil + principal d'une discussion, avec la clé et la vision active ; + - `TurnHost::keep_image` range l'image par le Go ; + - l'historique réécrit en cours de tour est publié sans les relais non + gardés. +- Les images gardées partent d'abord, avant tout résumé : + - au filet réactif (prompt trop long) : requête rejouée sans elles ; + - en cours de tour, au seuil : la compaction ne suit que si elle reste + nécessaire ; + - sur une complétion coupée par une fenêtre pleine : rejouée sans résumé + si leur retrait suffit. +- Début et fin de tour (`keptImagesTurnStart`) : les images gardées sont + retirées si le contexte le demande, ou si elles ne doivent plus partir + (clé retirée, vision absente). + +La compaction et la réduction forcée retiraient déjà les images gardées +d'abord (`chat/compact.rs`). + +## Tests + +- `chat/keepimg` : + - clé ; + - gardée au coût mesuré ; + - budget ; + - sans mesure (pas d'usage, complétion coupée, pas de base) ; + - requête relancée ; + - rangement refusé ; + - partage entre deux images ; + - publication, réécriture, fin de tour, budget plein avant rangement. +- La boucle du tour : + - requête identique avec et sans la clé, image gardée par référence au + coût mesuré ; + - refus : budget, pas d'usage, rangement refusé ; + - image à l'essai en fin de tour ; + - retrait en cours de tour, au filet réactif et sur une fenêtre pleine, à + chaque fois sans résumé. +- Le vrai hôte (`tests/lot6.rs`) : + - image rangée par le Go, vue en data-URL par le moteur, gardée avec son + coût ; + - clé retirée : l'image part au début du tour suivant, légende et marque + gardées. +- Le Go (`facade_keepimg_test.go`) : + - état sans projecteur puis vu par le moteur, budget ; + - rangement relu à l'octet près ; + - refus : texte seul, URL distante, format inconnu, base64 illisible, deux + images ; + - corps illisible, GET refusé. diff --git a/internal/loki/facade_chat.go b/internal/loki/facade_chat.go index 832a853..0e2cdeb 100644 --- a/internal/loki/facade_chat.go +++ b/internal/loki/facade_chat.go @@ -48,17 +48,16 @@ func facadeChat() bool { // facadeChatBlocker : le réglage qui garde la conversation ici, "" s'il n'y // en a pas. Ces clés opt-in demandent une machinerie restée au Go (bloc projet -// figé, images gardées, préchauffage). NUDGE_IN_TOOL n'en est plus (lot 23) : -// la façade a sa sonde de gabarit ; ni les motifs de relance personnalisés -// (retry_patterns, lot 30) : la façade les lit et les compile ; ni SIDE_SLOT et -// SLOT_PERSIST (lot 31) : la façade demande son slot avant chaque requête -// (facade_slots.go). +// figé, préchauffage). NUDGE_IN_TOOL n'en est plus (lot 23) : la façade a sa +// sonde de gabarit ; ni les motifs de relance personnalisés (retry_patterns, +// lot 30) : la façade les lit et les compile ; ni SIDE_SLOT et SLOT_PERSIST +// (lot 31) : la façade demande son slot avant chaque requête +// (facade_slots.go) ; ni KEEP_TURN_IMAGES (lot 32) : la façade tient l'état +// du tour, le Go range les images (facade_keepimg.go). func facadeChatBlocker(cfg map[string]string) string { switch { case projSnapEnabled(cfg): return "PROJ_SNAPSHOT" - case keepTurnImagesOn(cfg): - return "KEEP_TURN_IMAGES" case prewarmMode(cfg) != "": return "PREWARM" } diff --git a/internal/loki/facade_chat_test.go b/internal/loki/facade_chat_test.go index 1bbdd39..e9df015 100644 --- a/internal/loki/facade_chat_test.go +++ b/internal/loki/facade_chat_test.go @@ -277,7 +277,7 @@ func TestFacadeReglageQuiGardeLaConversation(t *testing.T) { t.Fatalf("détenteur : %d %s", w.Code, w.Body.String()) } // Posé après la décision : sans effet avant le redémarrage. - if err := SetConfigKey("KEEP_TURN_IMAGES", "on"); err != nil { + if err := SetConfigKey("PREWARM", "on"); err != nil { t.Fatal(err) } if !facadeChat() { @@ -285,10 +285,10 @@ func TestFacadeReglageQuiGardeLaConversation(t *testing.T) { } oublierDecisionFacade(t) if facadeChat() { - t.Fatal("KEEP_TURN_IMAGES posé : la conversation doit rester au Go") + t.Fatal("PREWARM posé : la conversation doit rester au Go") } w := appelOAIInterne(t, "GET", "/internal/v1/chat/holder", "") - if !strings.Contains(w.Body.String(), `"holder":"go"`) || !strings.Contains(w.Body.String(), "KEEP_TURN_IMAGES") { + if !strings.Contains(w.Body.String(), `"holder":"go"`) || !strings.Contains(w.Body.String(), "PREWARM") { t.Fatalf("détenteur : %s", w.Body.String()) } local := func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "go") } @@ -299,7 +299,7 @@ func TestFacadeReglageQuiGardeLaConversation(t *testing.T) { } for k, want := range map[string]string{ "PROJ_SNAPSHOT": "PROJ_SNAPSHOT", "NUDGE_IN_TOOL": "", "SIDE_SLOT": "", - "SLOT_PERSIST": "", "PREWARM": "PREWARM", + "SLOT_PERSIST": "", "KEEP_TURN_IMAGES": "", "PREWARM": "PREWARM", } { if got := facadeChatBlocker(map[string]string{k: "on"}); got != want { t.Errorf("%s : %q", k, got) diff --git a/internal/loki/facade_keepimg.go b/internal/loki/facade_keepimg.go new file mode 100644 index 0000000..de26039 --- /dev/null +++ b/internal/loki/facade_keepimg.go @@ -0,0 +1,47 @@ +package loki + +import ( + "net/http" +) + +// facade_keepimg.go — KEEP_TURN_IMAGES pour le tour tenu par la façade (lot 32 +// de la migration). L'état du tour (images à l'essai, mesure, budget) vit dans +// la façade ; ce qui touche aux fichiers et à la vision reste ici : +// +// - /internal/v1/turn/keepimg : la vision est-elle réellement active +// (visionEnabled, engineSeesImages) et le budget des images gardées +// (keepImgBudget) ; +// - /internal/v1/turn/keepimg/store : une image relayée rangée par +// référence (refRelayImage : fichier chiffré si la mémoire l'est, date +// rafraîchie, relu à l'octet près). + +// handleInternalTurnKeepImg : POST {} → {"vision":bool,"budget":n}. +func handleInternalTurnKeepImg(w http.ResponseWriter, r *http.Request) { + var in struct{} + if !readInternalJSON(w, r, &in) { + return + } + sendJSON(w, http.StatusOK, map[string]any{ + "vision": visionEnabled() && engineSeesImages(), + "budget": keepImgBudget(), + }) +} + +// handleInternalTurnKeepImgStore : POST {"msg":Message} → +// {"ok":true,"msg":Message,"weight":octets} ou {"ok":false} (format inconnu, +// disque plein, mémoire verrouillée, relecture différente : l'image reste +// éphémère). +func handleInternalTurnKeepImgStore(w http.ResponseWriter, r *http.Request) { + var in struct { + Msg Message `json:"msg"` + } + if !readInternalJSON(w, r, &in) { + return + } + m, weight, ok := refRelayImage(in.Msg) + if !ok { + sendJSON(w, http.StatusOK, map[string]any{"ok": false}) + return + } + sendJSON(w, http.StatusOK, map[string]any{"ok": true, "msg": m, "weight": weight}) +} diff --git a/internal/loki/facade_keepimg_test.go b/internal/loki/facade_keepimg_test.go new file mode 100644 index 0000000..48d142f --- /dev/null +++ b/internal/loki/facade_keepimg_test.go @@ -0,0 +1,87 @@ +package loki + +import ( + "encoding/json" + "os" + "strings" + "testing" +) + +// L'état que la façade demande au début d'un tour : vision réellement active +// (projecteur déclaré ET vu par le moteur) et budget des images gardées. +func TestFacadeKeepImgEtat(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + if err := SetConfigKey("CTX", "2000"); err != nil { + t.Fatal(err) + } + etat := func() (bool, float64) { + w := appelOAIInterne(t, "POST", "/internal/v1/turn/keepimg", `{}`) + var out struct { + Vision bool `json:"vision"` + Budget float64 `json:"budget"` + } + if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil || w.Code != 200 { + t.Fatalf("état : %d %s", w.Code, w.Body.String()) + } + return out.Vision, out.Budget + } + if v, b := etat(); v || b != 200 { + t.Fatalf("sans projecteur : vision=%v budget=%v", v, b) + } + visionServer(t, "") + if v, _ := etat(); !v { + t.Fatal("projecteur vu par le moteur : vision attendue") + } + if w := appelOAIInterne(t, "GET", "/internal/v1/turn/keepimg", ""); w.Code != 405 { + t.Fatalf("GET : %d", w.Code) + } +} + +// Une image relayée rangée par référence, relue à l'octet près ; sinon elle +// reste éphémère. +func TestFacadeKeepImgRangement(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + store := func(content string) map[string]any { + w := appelOAIInterne(t, "POST", "/internal/v1/turn/keepimg/store", `{"msg":{"role":"user","content":`+content+`}}`) + var out map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil || w.Code != 200 { + t.Fatalf("rangement : %d %s", w.Code, w.Body.String()) + } + return out + } + out := store(`[{"type":"text","text":"Voici la capture demandée."},{"type":"image_url","image_url":{"url":"data:image/png;base64,iVBORw0KGgo="}}]`) + if out["ok"] != true || out["weight"] != float64(8) { + t.Fatalf("rangée : %v", out) + } + msg := out["msg"].(map[string]any) + parts := msg["content"].([]any) + if parts[0].(map[string]any)["text"] != "Voici la capture demandée." { + t.Fatalf("légende : %v", parts) + } + ref := parts[1].(map[string]any)["image_url"].(map[string]any)["url"].(string) + if !strings.HasPrefix(ref, imgRefScheme) { + t.Fatalf("pas par référence : %s", ref) + } + if back, ok := chatImageDataURL(ref); !ok || back != "data:image/png;base64,iVBORw0KGgo=" { + t.Fatalf("relecture : %v %.40s", ok, back) + } + if ents, _ := os.ReadDir(chatImgDir()); len(ents) != 1 { + t.Fatalf("%d fichier(s) rangé(s)", len(ents)) + } + for _, c := range []string{ + `"texte seul"`, + `[{"type":"image_url","image_url":{"url":"https://exemple.org/a.png"}}]`, + `[{"type":"image_url","image_url":{"url":"data:image/x-inconnu;base64,AAAA"}}]`, + `[{"type":"image_url","image_url":{"url":"data:image/png;base64,pas du base64"}}]`, + `[{"type":"image_url","image_url":{"url":"data:image/png;base64,iVBORw0KGgo="}},{"type":"image_url","image_url":{"url":"data:image/png;base64,iVBORw0KGgo="}}]`, + } { + if out := store(c); out["ok"] != false { + t.Errorf("%s : %v", c, out) + } + } + if w := appelOAIInterne(t, "POST", "/internal/v1/turn/keepimg/store", `{`); w.Code != 400 { + t.Fatalf("corps illisible : %d", w.Code) + } +} diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index d7c76f7..6a36e7d 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -75,6 +75,10 @@ func withInternalAPI(next http.Handler) http.Handler { handleInternalTurnPerf(w, r) case "/internal/v1/turn/slot": handleInternalTurnSlot(w, r) + case "/internal/v1/turn/keepimg": + handleInternalTurnKeepImg(w, r) + case "/internal/v1/turn/keepimg/store": + handleInternalTurnKeepImgStore(w, r) case "/internal/v1/tool/exec": handleInternalToolExec(w, r) case "/internal/v1/tool/context": diff --git a/rust/loki-rs/src/chat/compact.rs b/rust/loki-rs/src/chat/compact.rs index b21f67c..215ca70 100644 --- a/rust/loki-rs/src/chat/compact.rs +++ b/rust/loki-rs/src/chat/compact.rs @@ -416,7 +416,7 @@ pub(crate) fn is_loki_injected(m: &Message) -> bool { /// `dropKeptImages` : copie où chaque image gardée devient sa légende suivie /// d'[`IMAGE_LOST_MARKER`], et coût libéré. `None` : aucune image gardée. -fn drop_kept_images(msgs: &[Message]) -> Option<(Vec, i64)> { +pub fn drop_kept_images(msgs: &[Message]) -> Option<(Vec, i64)> { if !msgs.iter().any(|m| m.img_tokens > 0) { return None; } @@ -441,7 +441,7 @@ fn drop_kept_images(msgs: &[Message]) -> Option<(Vec, i64)> { Some((out, freed)) } -fn log_kept_images_dropped(phase: &str, freed: i64) { +pub fn log_kept_images_dropped(phase: &str, freed: i64) { eprintln!("[images] {phase} : images gardées retirées ({freed} jetons)"); } diff --git a/rust/loki-rs/src/chat/generate.rs b/rust/loki-rs/src/chat/generate.rs index 5ed7626..840812d 100644 --- a/rust/loki-rs/src/chat/generate.rs +++ b/rust/loki-rs/src/chat/generate.rs @@ -22,15 +22,17 @@ //! corrections du builder) suit le tour, avant la compaction de fin //! ([`super::verify`]) ; toute compaction rend l'état des critères au résumé. //! -//! Écarts assumés (restent au Go) : PROJ_SNAPSHOT et KEEP_TURN_IMAGES -//! (refusés par `/turn/view`), préchauffage. La télémétrie des complétions +//! Écarts assumés (restent au Go) : PROJ_SNAPSHOT (refusé par +//! `/turn/view`), préchauffage. Les images gardées (KEEP_TURN_IMAGES, lot 32) +//! partent en début et en fin de tour quand il le faut +//! ([`ChatService::kept_images_turn_start`]). La télémétrie des complétions //! (tours, vérifications, résumés) rejoint l'anneau du Go //! (`/internal/v1/turn/perf`) ; les motifs `retry_patterns` personnalisés //! sont lus au démarrage ([`retry_patterns`]). Chaque requête au moteur local //! part sur le slot que rend le Go (SIDE_SLOT, rechargement SLOT_PERSIST, //! lot 31). -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Mutex}; @@ -42,14 +44,18 @@ use tokio::sync::watch; use super::compact::{ CompactCaps, CompactConfig, CompactOpts, Compactor, TurnView, compact_needed, compact_opts_for, - ctx_pending, estimate_tokens, log_compact, log_ctx_estimate, strip_context_updates, + ctx_pending, drop_kept_images, estimate_tokens, log_compact, log_ctx_estimate, + log_kept_images_dropped, strip_context_updates, }; use super::engine::EngineHttp; use super::history::{ drop_stray_nudges, history_from_view, inject_prefix, remind_read_mem_pages, tool_used_delta, }; -use super::host::{ChatShared, EngineSummarizer, GoTurnHost, StoreRecall, TurnCompact}; +use super::host::{ + ChatShared, EngineSummarizer, GoTurnHost, StoreRecall, TurnCompact, keep_images_state, +}; use super::journal::Delta; +use super::keepimg::{has_kept_images, keep_turn_images_on}; use super::lease::{EngineLease, LEASE_TTL, LeaseError, LeaseSignal, engine_stop}; use super::sessions::{GoSessionHooks, Sessions}; use super::turn::{Cancel, TurnConfig, run_turn}; @@ -410,6 +416,55 @@ impl ChatService { } } + /// `keptImagesTurnStart` : début (ou fin) d'un tour. Les images gardées + /// (KEEP_TURN_IMAGES) sont retirées d'abord — avant toute compaction, qui + /// ne résume alors que si c'est encore nécessaire — quand le contexte la + /// réclame, ou quand elles ne doivent plus partir : clé retirée, vision + /// absente (un moteur sans projecteur refuserait la requête). Rend + /// l'historique et le contexte à jour. + async fn kept_images_turn_start( + &self, + epoch: i64, + cfg: &HashMap, + ccfg: &CompactConfig, + msgs: Vec, + ctx_used: i64, + peak: i64, + ) -> (Vec, i64) { + if !has_kept_images(&msgs) { + return (msgs, ctx_used); + } + if keep_turn_images_on(cfg) + && matches!(keep_images_state(&self.shared.store).await, Some((true, _))) + && !compact_needed(ccfg, &msgs, ctx_used, peak) + { + return (msgs, ctx_used); + } + let Some((out, freed)) = drop_kept_images(&msgs) else { + return (msgs, ctx_used); + }; + log_kept_images_dropped("début-tour", freed); + let ctx_used = if ctx_used > 0 { + (ctx_used - freed).max(1) + } else { + ctx_used + }; + let conv = self.sessions.conv(); + let cur = conv + .with_state_if(epoch, |st| { + st.messages = out.clone(); + if st.ctx_used > 0 { + st.ctx_used = (st.ctx_used - freed).max(1); + } + st.ctx_used + }) + .unwrap_or(0); + if cur > 0 { + conv.append_delta(epoch, obj("ctx_used", cur)); + } + (out, ctx_used) + } + /// `compactAndPublish` : bannière, journal, installation du résultat, /// jauge de contexte. Rend l'historique compacté, `None` s'il n'a pas /// changé. @@ -718,10 +773,15 @@ impl ChatService { let peak = std::mem::take(&mut st.gen_peak); (msgs, used, peak) }); - let Some((mut msgs, ctx_used, peak)) = snap else { + let Some((msgs, ctx_used, peak)) = snap else { lease.release().await; return; }; + // KEEP_TURN_IMAGES : images gardées retirées d'abord si le contexte le + // demande ou si elles ne doivent plus partir. + let (mut msgs, ctx_used) = self + .kept_images_turn_start(epoch, &cfg, &ccfg, msgs, ctx_used, peak) + .await; // Compaction de début de tour. let mut start_est = ctx_used; @@ -972,7 +1032,13 @@ impl ChatService { } // Compaction de fin de tour : la jauge retombe tout de suite, le tour - // suivant démarre avec de la marge. Pas après un stop. + // suivant démarre avec de la marge. Pas après un stop. Les images + // gardées partent d'abord, comme en début de tour. + if !cancel.is_cancelled() { + (msgs, ctx_used) = self + .kept_images_turn_start(epoch, &cfg, &ccfg, msgs, ctx_used, peak) + .await; + } if !cancel.is_cancelled() && compact_needed(&ccfg, &msgs, ctx_used, peak) && !self diff --git a/rust/loki-rs/src/chat/host.rs b/rust/loki-rs/src/chat/host.rs index a3af40f..385d1c8 100644 --- a/rust/loki-rs/src/chat/host.rs +++ b/rust/loki-rs/src/chat/host.rs @@ -420,6 +420,22 @@ pub async fn engine_begin(store: &Store, main_conv: &str, side: bool) -> Option< } } +/// KEEP_TURN_IMAGES (lot 32) : la vision est-elle réellement active +/// (projecteur déclaré et vu par le moteur) et le budget des images gardées. +/// Go injoignable : `None` (images éphémères). +pub async fn keep_images_state(store: &Store) -> Option<(bool, i64)> { + match store + .internal_post("/internal/v1/turn/keepimg", &json!({})) + .await + { + Ok((code, v)) if code.is_success() => Some(( + v["vision"].as_bool().unwrap_or(false), + v["budget"].as_i64().unwrap_or(0), + )), + _ => None, + } +} + /// Rend le ticket « en vol » au Go quand la réponse du moteur est lue ou /// abandonnée (`engineRequestBeginSide` … `end`). pub struct GateTicket { @@ -704,6 +720,35 @@ impl TurnHost for GoTurnHost { self.shared.echo_refusals.refuse(model); } + async fn keep_images(&self) -> Option { + // Fil principal d'une discussion seulement : la vérification ne + // persiste pas son historique. + if !super::keepimg::keep_turn_images_on(&self.cfg) + || self.perf_kind != super::perf::KIND_MAIN + || self.conv_id.is_empty() + { + return None; + } + match keep_images_state(&self.shared.store).await { + Some((true, budget)) => Some(budget), + _ => None, + } + } + + async fn keep_image(&self, m: &Message) -> Option<(Message, i64)> { + let (code, v) = self + .shared + .store + .internal_post("/internal/v1/turn/keepimg/store", &json!({ "msg": m })) + .await + .ok()?; + if !code.is_success() || v["ok"] != true { + return None; + } + let msg: Message = serde_json::from_value(v["msg"].clone()).ok()?; + Some((msg, v["weight"].as_i64().unwrap_or(0))) + } + async fn engine_slot(&self, ep: &ChatEndpoint) -> Option { if ep.external { self.side_slot.store(false, Ordering::SeqCst); diff --git a/rust/loki-rs/src/chat/keepimg.rs b/rust/loki-rs/src/chat/keepimg.rs new file mode 100644 index 0000000..9deea5a --- /dev/null +++ b/rust/loki-rs/src/chat/keepimg.rs @@ -0,0 +1,272 @@ +//! KEEP_TURN_IMAGES pour le tour tenu par la façade (lot 32) — port de +//! l'état du tour de `chat_keep_images.go` (`keepImages`). +//! +//! Sans la clé, une image montrée par un outil (capture, `see_image`) est +//! ÉPHÉMÈRE : vue pendant le tour, absente de l'historique. Avec la clé, elle +//! est gardée telle qu'envoyée, sous conditions : +//! - fil principal d'une discussion, vision réellement active (demandé au Go +//! à chaque tour) ; +//! - rangée par référence par le Go et relue à l'octet près ; +//! - coût MESURÉ par le moteur (écart de `usage.prompt_tokens` entre deux +//! requêtes, moins l'estimation du texte ajouté), jamais estimé ; +//! - toutes les images gardées tiennent dans 10 % de la fenêtre ; la +//! première qui ne tient pas reste éphémère, et toutes celles qui la +//! suivent dans le tour aussi (le préfixe divergera de toute façon à elle). +//! +//! Le retrait prioritaire par une compaction ou une réduction est dans +//! [`super::compact`] ; celui du début de tour, de la boucle, ici et dans +//! [`super::turn`]. + +use std::collections::{HashMap, HashSet}; + +use super::compact::msg_tokens; +use super::types::{Message, StatsEvent}; +use super::wire::content_parts; + +/// En dessous, la mesure ne vaut rien (écart noyé dans l'erreur +/// d'estimation du texte) : l'image reste éphémère. +pub const KEEP_IMG_MIN_TOKENS: i64 = 16; + +/// `keepTurnImagesOn`. +pub fn keep_turn_images_on(cfg: &HashMap) -> bool { + matches!( + cfg.get("KEEP_TURN_IMAGES") + .map(|v| v.trim().to_lowercase()) + .as_deref(), + Some("on" | "1" | "true" | "yes" | "oui") + ) +} + +/// `keptImageTokens` : coût total des images gardées. +pub fn kept_image_tokens(msgs: &[Message]) -> i64 { + msgs.iter().map(|m| m.img_tokens).sum() +} + +/// `hasKeptImages`. +pub fn has_kept_images(msgs: &[Message]) -> bool { + msgs.iter().any(|m| m.img_tokens > 0) +} + +fn has_image_part(m: &Message) -> bool { + content_parts(m.content.as_ref()) + .iter() + .any(|p| p.get("type").and_then(|t| t.as_str()) == Some("image_url")) +} + +#[derive(Debug, Clone, Copy, PartialEq)] +struct Pend { + msg_idx: usize, + extra_idx: usize, + weight: i64, +} + +/// L'état de KEEP_TURN_IMAGES pendant UN tour ; absent (`None` dans la +/// boucle) sans la clé, hors fil principal ou sans vision. +#[derive(Debug, Default)] +pub struct KeepImages { + /// `keepImgBudget` : part de la fenêtre que les images gardées occupent. + budget: i64, + /// Une image du tour n'a pas été gardée : les suivantes non plus. + stopped: bool, + /// Images de l'étape courante, à l'essai jusqu'à la mesure de la requête + /// qui les montre la première. + pend: Vec, + /// Numéro de cette requête, contexte que la complétion précédente laisse + /// (`ctxAfter`), premier message ajouté depuis. + pend_req: i64, + pend_base: i64, + pend_from: usize, + /// Requêtes envoyées ; messages de la dernière. + req_n: i64, + sent_len: usize, + /// `ctxAfter` de la dernière complétion complète (0 : inconnu). + last_after: i64, +} + +impl KeepImages { + pub fn new(budget: i64) -> Self { + KeepImages { + budget, + ..Default::default() + } + } + + /// Faut-il ranger l'image (ni tour arrêté, ni budget déjà plein) ? + pub fn wants(&self, messages: &[Message]) -> bool { + !self.stopped && kept_image_tokens(messages) < self.budget + } + + /// `relay` : ajoute le message qui relaie une image d'outil. `stored` : + /// l'image rangée par référence et son poids (octets), `None` si elle ne + /// l'a pas été — elle reste alors éphémère, et le tour s'arrête là pour + /// les suivantes. `from` : début des messages de l'étape. + pub fn relay( + &mut self, + messages: &mut Vec, + extra: &mut Vec, + mut m: Message, + from: usize, + stored: Option<(Message, i64)>, + ) { + m.img_relay = true; + if self.stopped { + messages.push(m); + return; + } + let Some((mut r, weight)) = stored.filter(|(_, w)| *w > 0) else { + self.stopped = true; + messages.push(m); + return; + }; + r.img_relay = true; + if self.pend.is_empty() { + self.pend_req = self.req_n + 1; + self.pend_base = self.last_after; + self.pend_from = from; + } + messages.push(r.clone()); + extra.push(r); + self.pend.push(Pend { + msg_idx: messages.len() - 1, + extra_idx: extra.len() - 1, + weight, + }); + } + + /// `sending` : une requête part avec `n` messages. + pub fn sending(&mut self, n: usize) { + self.req_n += 1; + self.sent_len = n; + } + + /// `observe` : une complétion est lue. Mesure le coût des images à + /// l'essai si c'est la requête qui les a montrées la première, et les + /// garde si le budget le permet ; sinon elles redeviennent éphémères. + pub fn observe( + &mut self, + complete: bool, + st: &StatsEvent, + messages: &mut [Message], + extra: &mut Vec, + ) { + let after = if complete && st.prompt_tokens_total > 0 { + st.ctx_after() + } else { + 0 + }; + if !self.pend.is_empty() { + if !self.measure(after, st.prompt_tokens_total, messages, extra) { + self.reject(extra); + } + self.pend.clear(); + } + self.last_after = after; + } + + /// `measure` : répartit le coût mesuré entre les images à l'essai, au + /// prorata de leur poids, et les garde si tout tient dans le budget. + fn measure( + &self, + after: i64, + prompt: i64, + messages: &mut [Message], + extra: &mut [Message], + ) -> bool { + if after <= 0 + || self.pend_base <= 0 + || self.pend_req != self.req_n + || self.pend_from > self.sent_len + || self.sent_len > messages.len() + { + return false; + } + // Ce que la requête a ajouté depuis la précédente, hors images. + let text: i64 = messages[self.pend_from..self.sent_len] + .iter() + .filter(|m| !m.img_relay) + .map(msg_tokens) + .sum(); + let img = prompt - self.pend_base - text; + let weight: i64 = self.pend.iter().map(|p| p.weight).sum(); + let n = self.pend.len() as i64; + if weight <= 0 || img < KEEP_IMG_MIN_TOKENS * n { + return false; + } + if kept_image_tokens(messages) + img > self.budget { + return false; + } + let mut shares = Vec::with_capacity(self.pend.len()); + let mut left = img; + for (i, p) in self.pend.iter().enumerate() { + let share = if i == self.pend.len() - 1 { + left + } else { + img * p.weight / weight + }; + if share < KEEP_IMG_MIN_TOKENS { + return false; + } + left -= share; + shares.push(share); + } + if self.pend.iter().any(|p| { + !messages.get(p.msg_idx).is_some_and(|m| m.img_relay) + || !extra.get(p.extra_idx).is_some_and(|m| m.img_relay) + }) { + return false; + } + for (p, share) in self.pend.iter().zip(shares) { + messages[p.msg_idx].img_tokens = share; + extra[p.extra_idx].img_tokens = share; + } + true + } + + /// `reject` : retire d'`extra` les images à l'essai — éphémères, comme + /// sans la clé — et arrête le tour pour les suivantes. + fn reject(&mut self, extra: &mut Vec) { + if self.pend.is_empty() { + return; + } + self.stopped = true; + let drop: HashSet = self.pend.iter().map(|p| p.extra_idx).collect(); + let mut i = 0; + extra.retain(|_| { + let keep = !drop.contains(&i); + i += 1; + keep + }); + self.pend.clear(); + } + + /// `rewritten` : l'historique vient d'être réécrit en cours de tour et + /// `extra` repart de zéro. Les images à l'essai ne se mesurent plus : + /// éphémères, et le tour s'arrête là pour les suivantes. + pub fn rewritten(&mut self) { + if !self.pend.is_empty() { + self.stopped = true; + } + self.pend.clear(); + } + + /// `finish` : fin du tour. Une image encore à l'essai n'a pas été + /// mesurée : éphémère. + pub fn finish(&mut self, extra: &mut Vec) { + self.reject(extra); + } +} + +/// `publishable` : l'historique publié par une réécriture en cours de tour, +/// sans les relais dont l'image n'est pas gardée. Sans l'état : tel quel. +pub fn publishable(kt: Option<&KeepImages>, msgs: &[Message]) -> Vec { + if kt.is_none() { + return msgs.to_vec(); + } + msgs.iter() + .filter(|m| !(m.img_relay && m.img_tokens == 0 && has_image_part(m))) + .cloned() + .collect() +} + +#[cfg(test)] +mod tests; diff --git a/rust/loki-rs/src/chat/keepimg/tests.rs b/rust/loki-rs/src/chat/keepimg/tests.rs new file mode 100644 index 0000000..9a17d2c --- /dev/null +++ b/rust/loki-rs/src/chat/keepimg/tests.rs @@ -0,0 +1,298 @@ +//! Tests de l'état KEEP_TURN_IMAGES d'un tour (cas de +//! `chat_keep_images_test.go`). + +use super::*; +use crate::chat::compact::IMAGE_LOST_MARKER; +use serde_json::{Value, json}; + +fn um(s: &str) -> Message { + Message { + role: "user".into(), + content: Some(json!(s)), + ..Default::default() + } +} + +fn tm(s: &str) -> Message { + Message { + role: "tool".into(), + tool_call_id: "v1".into(), + content: Some(json!(s)), + ..Default::default() + } +} + +fn img(url: &str) -> Message { + Message { + role: "user".into(), + content: Some(json!([ + {"type": "text", "text": "Voici la capture demandée."}, + {"type": "image_url", "image_url": {"url": url}} + ])), + ..Default::default() + } +} + +fn stats(prompt: i64, gen_tokens: i64) -> StatsEvent { + StatsEvent { + prompt_tokens_total: prompt, + gen_tokens, + ..Default::default() + } +} + +fn kept(tokens: i64) -> Message { + let mut m = img("loki-img:aa.png"); + m.img_relay = true; + m.img_tokens = tokens; + m +} + +/// Une étape `see_image` : première complétion (prompt 1000, 20 générés), +/// résultat d'outil, relais rangé ; rend (état, messages, extra, base du +/// texte ajouté). +fn step(k: &mut KeepImages, stored: bool) -> (Vec, Vec, i64) { + let mut messages = vec![um("regarde img.png")]; + let mut extra = Vec::new(); + k.sending(messages.len()); + k.observe(true, &stats(1000, 20), &mut messages, &mut extra); + let from = messages.len(); + let t = tm("[ok] image chargée : img.png"); + let text = msg_tokens(&t); + messages.push(t.clone()); + extra.push(t); + let s = stored.then(|| (img("loki-img:aa.png"), 1000)); + k.relay( + &mut messages, + &mut extra, + img("data:image/png;base64,AA=="), + from, + s, + ); + k.sending(messages.len()); + (messages, extra, text) +} + +#[test] +fn cle() { + for (v, want) in [ + ("on", true), + (" OUI ", true), + ("1", true), + ("true", true), + ("yes", true), + ("off", false), + ("", false), + ("2", false), + ] { + let cfg = HashMap::from([("KEEP_TURN_IMAGES".to_string(), v.to_string())]); + assert_eq!(keep_turn_images_on(&cfg), want, "{v:?}"); + } + assert!(!keep_turn_images_on(&HashMap::new())); +} + +/// Coût mesuré par le moteur : écart de prompt moins le texte ajouté ; gardée +/// par référence, dans `messages` comme dans `extra`. +#[test] +fn gardee_et_mesuree() { + let mut k = KeepImages::new(3200); + let (mut messages, mut extra, text) = step(&mut k, true); + assert_eq!(extra.len(), 2); + k.observe( + true, + &stats(1020 + text + 300, 5), + &mut messages, + &mut extra, + ); + assert_eq!(extra.len(), 2); + let r = &extra[1]; + assert!(r.img_relay && r.img_tokens == 300, "{r:?}"); + assert_eq!( + r.content.as_ref().unwrap()[1]["image_url"]["url"], + "loki-img:aa.png" + ); + assert_eq!(messages[2].img_tokens, 300); + assert_eq!(kept_image_tokens(&messages), 300); +} + +/// Au-delà du budget, l'image reste éphémère : retirée d'`extra`, gardée +/// dans `messages` (le modèle l'a vue), et le tour s'arrête là. +#[test] +fn budget() { + let mut k = KeepImages::new(200); + let (mut messages, mut extra, text) = step(&mut k, true); + k.observe( + true, + &stats(1020 + text + 300, 5), + &mut messages, + &mut extra, + ); + assert_eq!(extra.len(), 1); + assert_eq!(messages.len(), 3); + assert!(messages[2].img_relay && messages[2].img_tokens == 0); + assert!(k.stopped); + // Une image suivante n'est même pas rangée. + assert!(!k.wants(&messages)); +} + +/// Moteur sans usage, ou complétion coupée : rien à mesurer. +#[test] +fn sans_mesure() { + for (complete, prompt) in [(true, 0), (false, 2000)] { + let mut k = KeepImages::new(3200); + let (mut messages, mut extra, _) = step(&mut k, true); + k.observe(complete, &stats(prompt, 5), &mut messages, &mut extra); + assert_eq!(extra.len(), 1, "complète={complete} prompt={prompt}"); + assert_eq!(kept_image_tokens(&messages), 0); + } + // Première complétion sans usage : pas de base. + let mut k = KeepImages::new(3200); + let mut messages = vec![um("q")]; + let mut extra = Vec::new(); + k.sending(1); + k.observe(true, &stats(0, 0), &mut messages, &mut extra); + let t = tm("[ok]"); + messages.push(t.clone()); + extra.push(t); + k.relay( + &mut messages, + &mut extra, + img("data:x"), + 1, + Some((img("loki-img:b.png"), 9)), + ); + k.sending(messages.len()); + k.observe(true, &stats(5000, 5), &mut messages, &mut extra); + assert_eq!(extra.len(), 1); +} + +/// Requête relancée entre-temps (reprise réseau) : la mesure ne vaut plus. +#[test] +fn requete_relancee() { + let mut k = KeepImages::new(3200); + let (mut messages, mut extra, text) = step(&mut k, true); + k.sending(messages.len()); + k.observe( + true, + &stats(1020 + text + 300, 5), + &mut messages, + &mut extra, + ); + assert_eq!(extra.len(), 1); +} + +/// Rangement refusé : éphémère, marqué relais, et les suivantes aussi. +#[test] +fn rangement_refuse() { + let mut k = KeepImages::new(3200); + let (mut messages, extra, _) = step(&mut k, false); + assert_eq!(extra.len(), 1); + assert!(messages[2].img_relay); + assert!(k.stopped); + let mut extra = extra; + k.relay( + &mut messages, + &mut extra, + img("data:y"), + 1, + Some((img("loki-img:c.png"), 5)), + ); + assert_eq!(extra.len(), 1, "tour arrêté : plus rien de gardé"); + assert!(messages[3].img_relay); +} + +/// Deux images dans l'étape : le coût réparti au prorata des octets, la +/// dernière prend le reste ; une part sous le seuil, tout reste éphémère. +#[test] +fn partage_entre_images() { + let mut k = KeepImages::new(3200); + let mut messages = vec![um("q")]; + let mut extra = Vec::new(); + k.sending(1); + k.observe(true, &stats(1000, 20), &mut messages, &mut extra); + k.relay( + &mut messages, + &mut extra, + img("data:a"), + 1, + Some((img("loki-img:a.png"), 3000)), + ); + k.relay( + &mut messages, + &mut extra, + img("data:b"), + 1, + Some((img("loki-img:b.png"), 1000)), + ); + k.sending(messages.len()); + k.observe(true, &stats(1020 + 401, 5), &mut messages, &mut extra); + assert_eq!( + (extra[0].img_tokens, extra[1].img_tokens), + (300, 101), + "{extra:?}" + ); + let mut k = KeepImages::new(3200); + let mut messages = vec![um("q")]; + let mut extra = Vec::new(); + k.sending(1); + k.observe(true, &stats(1000, 20), &mut messages, &mut extra); + k.relay( + &mut messages, + &mut extra, + img("data:a"), + 1, + Some((img("loki-img:a.png"), 3000)), + ); + k.relay( + &mut messages, + &mut extra, + img("data:b"), + 1, + Some((img("loki-img:b.png"), 10)), + ); + k.sending(messages.len()); + k.observe(true, &stats(1020 + 400, 5), &mut messages, &mut extra); + assert!(extra.is_empty(), "part de 1 jeton : {extra:?}"); +} + +/// Réécriture en cours de tour : une image non gardée n'entre pas dans +/// l'historique publié ; gardée ou déjà retirée, si. Les images à l'essai ne +/// sont plus suivies. +#[test] +fn publication() { + let mut eph = img("data:image/png;base64,AA=="); + eph.img_relay = true; + let mut stub = um(&format!("Image :{IMAGE_LOST_MARKER}")); + stub.img_relay = true; + let msgs = vec![um("q"), tm("a"), eph, tm("b"), kept(50), stub]; + let k = KeepImages::new(3200); + let got = publishable(Some(&k), &msgs); + assert_eq!(got.len(), 5); + assert_eq!(got[2].content, Some(Value::from("b"))); + assert_eq!(publishable(None, &msgs).len(), 6, "sans la clé : tel quel"); + let mut k = KeepImages::new(3200); + let (_, _, _) = step(&mut k, true); + k.rewritten(); + assert!(k.stopped && k.pend.is_empty()); + let mut k = KeepImages::new(3200); + k.rewritten(); + assert!(!k.stopped, "rien à l'essai : le tour continue"); +} + +/// Fin du tour avec une image à l'essai : éphémère. +#[test] +fn fin_de_tour() { + let mut k = KeepImages::new(3200); + let (_, mut extra, _) = step(&mut k, true); + k.finish(&mut extra); + assert_eq!(extra.len(), 1); + assert_eq!(extra[0].role, "tool"); +} + +/// Budget déjà plein : rien à ranger. +#[test] +fn budget_plein_avant_rangement() { + let k = KeepImages::new(200); + assert!(!k.wants(&[um("q"), kept(200)])); + assert!(k.wants(&[um("q"), kept(199)])); +} diff --git a/rust/loki-rs/src/chat/mod.rs b/rust/loki-rs/src/chat/mod.rs index b5c67fa..fbd6997 100644 --- a/rust/loki-rs/src/chat/mod.rs +++ b/rust/loki-rs/src/chat/mod.rs @@ -28,6 +28,7 @@ pub mod generate; pub mod history; pub mod host; pub mod journal; +pub mod keepimg; pub mod lease; pub mod nudges; pub mod perf; diff --git a/rust/loki-rs/src/chat/turn.rs b/rust/loki-rs/src/chat/turn.rs index 3856d31..040f0cd 100644 --- a/rust/loki-rs/src/chat/turn.rs +++ b/rust/loki-rs/src/chat/turn.rs @@ -34,9 +34,13 @@ //! SLOT_PERSIST et le comptage des requêtes en vol passent par le Go (l'hôte, //! autour de [`TurnHost::engine_slot`] et de [`TurnHost::send`]). //! +//! Images gardées (KEEP_TURN_IMAGES, lot 32) : une image relayée par un outil +//! est gardée à l'essai jusqu'à la mesure de son coût par le moteur +//! ([`super::keepimg`]) ; les images gardées partent avant tout résumé. +//! //! Non portés (clés opt-in, ou machinerie du moteur local restée au Go) : -//! KEEP_TURN_IMAGES, PROJ_SNAPSHOT, préchauffage. Avec l'une de ces clés, le -//! Go garde la conversation (`facadeChatBlocker`, décidé au démarrage). +//! PROJ_SNAPSHOT, préchauffage. Avec l'une de ces clés, le Go garde la +//! conversation (`facadeChatBlocker`, décidé au démarrage). use std::collections::HashMap; use std::future::Future; @@ -51,13 +55,15 @@ use tokio::sync::watch; use super::compact::cap_tool_result; use super::compact::{ - CompactConfig, compact_needed, compact_would_trigger, context_overflow, estimate_tokens, - log_compact, log_ctx, log_ctx_estimate, overflow_tokens, shrink_to_fit, + CompactConfig, compact_needed, compact_would_trigger, context_overflow, drop_kept_images, + estimate_tokens, log_compact, log_ctx, log_ctx_estimate, log_kept_images_dropped, + overflow_tokens, shrink_to_fit, }; use super::engine::{ LineError, Lines, NET_RETRIES, NetKind, net_delay, retryable, retryable_status, stream_cut_error, }; +use super::keepimg::{KeepImages, has_kept_images, publishable}; use super::nudges::{ MAX_NUDGES, THINK_NUDGE_FIRST, THINK_NUDGE_STUCK, append_nudge, budget_nudge, dedupable_tool, log_budget, repeated_call_result, steer_system, tool_result_preview, @@ -193,6 +199,17 @@ pub trait TurnHost: Send + Sync { fn preflight(&self) -> Option { None } + /// KEEP_TURN_IMAGES : le budget des images gardées si la clé vaut pour + /// ce tour (fil principal, vision active) ; `None` : images éphémères. + fn keep_images(&self) -> impl Future> + Send { + async { None } + } + /// Range par référence l'image d'un relais (`refRelayImage`) : le + /// message qui la porte ainsi et son poids en octets ; `None` : elle + /// reste éphémère. + fn keep_image(&self, _m: &Message) -> impl Future> + Send { + async { None } + } /// Slot du moteur où envoyer la requête qui part (`engineSlotFor`) ; /// `None` : aucun `id_slot`. Appelé avant chaque requête, moteur local ou /// non : l'hôte y recharge aussi l'état SLOT_PERSIST de la discussion. @@ -428,13 +445,46 @@ struct Loop { pool_retries: u32, } -/// `runChatLoop` : le tour, de la première requête à la réponse finale. +/// `runChatTools` : le tour, de la première requête à la réponse finale. +/// KEEP_TURN_IMAGES : en fin de tour, une image gardée à l'essai mais jamais +/// mesurée redevient éphémère, quelle que soit la sortie de la boucle. pub async fn run_turn( + host: &H, + tc: &TurnConfig, + messages: Vec, + cancel: &Cancel, + cb: &mut (dyn FnMut(StreamEvent) -> bool + Send), +) -> TurnResult { + let mut kt = host.keep_images().await.map(KeepImages::new); + let mut r = run_turn_loop(host, tc, messages, cancel, cb, &mut kt).await; + if let Some(k) = kt.as_mut() { + k.finish(&mut r.extra); + } + r +} + +/// `publishHistory` : publie l'historique réécrit en cours de tour +/// (compaction, réduction, retrait des images gardées). KEEP_TURN_IMAGES : +/// une image non gardée n'y entre pas ; sans la clé, tel quel. +fn publish_history( + cb: &mut (dyn FnMut(StreamEvent) -> bool + Send), + kt: &mut Option, + m: &[Message], +) { + if let Some(k) = kt.as_mut() { + k.rewritten(); + } + cb(StreamEvent::NewHistory(publishable(kt.as_ref(), m))); +} + +/// `runChatLoop` : la boucle de [`run_turn`]. +async fn run_turn_loop( host: &H, tc: &TurnConfig, mut messages: Vec, cancel: &Cancel, cb: &mut (dyn FnMut(StreamEvent) -> bool + Send), + kt: &mut Option, ) -> TurnResult { let mut extra: Vec = Vec::new(); let mut s = Loop::default(); @@ -451,9 +501,6 @@ pub async fn run_turn( if tc.code { budget *= super::nudges::CODE_BUDGET_FACTOR; } - let new_history = |cb: &mut (dyn FnMut(StreamEvent) -> bool + Send), m: &[Message]| { - cb(StreamEvent::NewHistory(m.to_vec())); - }; let fail = |cb: &mut (dyn FnMut(StreamEvent) -> bool + Send), extra: Vec, err: String| { cb(StreamEvent::Err(err.clone())); @@ -524,6 +571,10 @@ pub async fn run_turn( ) }; let t_req = Instant::now(); + // KEEP_TURN_IMAGES : rien sans la clé. + if let Some(k) = kt.as_mut() { + k.sending(messages.len()); + } let reply = match cancel.or_stop(host.send(ep, body)).await { None => return TurnResult { extra, err: None }, Some(Ok(r)) => r, @@ -603,6 +654,18 @@ pub async fn run_turn( if s.echo_off_turn { messages = strip_reasoning(&messages).into_owned(); } + // KEEP_TURN_IMAGES : les images gardées partent d'abord, et la + // requête est rejouée sans elles ; le résumé (avec perte) ne vient + // qu'ensuite, s'il le faut encore. + if context_overflow(ccfg, &msg, &messages) + && let Some((c, freed)) = drop_kept_images(&messages) + { + log_kept_images_dropped("réactif", freed); + messages = c; + extra.clear(); + publish_history(cb, kt, &messages); + continue; + } let overflow = context_overflow(ccfg, &msg, &messages); if ccfg.enabled && !s.compacted_retry && overflow { let Some(c) = cancel @@ -616,7 +679,7 @@ pub async fn run_turn( log_compact(ccfg, "réactif", 0, &messages, &c, true); messages = c; extra.clear(); - new_history(cb, &messages); + publish_history(cb, kt, &messages); continue; } } @@ -628,7 +691,7 @@ pub async fn run_turn( log_compact(ccfg, "réduction", actual, &messages, &c, true); messages = c; extra.clear(); - new_history(cb, &messages); + publish_history(cb, kt, &messages); continue; } } @@ -761,6 +824,11 @@ pub async fn run_turn( } s.last_est = 0; } + // KEEP_TURN_IMAGES : coût des images de l'étape précédente, mesuré sur + // cette complétion ; gardées ou rendues éphémères. + if let Some(k) = kt.as_mut() { + k.observe(complete, &out.stats, &mut messages, &mut extra); + } if out.finish_reason == "length" { log_ctx(&format!( "finish=length prompt={} généré={} raisonnement={} fenêtre={}", @@ -980,7 +1048,20 @@ pub async fn run_turn( // Plafond pour le MODÈLE seulement : l'UI a reçu le résultat. push_result(&mut messages, &mut extra, cap_tool_result(&ex.result)); if let Some(img) = ex.relay { - messages.push(img); // éphémère : jamais dans l'historique + match kt.as_mut() { + // Éphémère : jamais dans l'historique. + None => messages.push(img), + // KEEP_TURN_IMAGES : gardée par référence, à l'essai + // jusqu'à la mesure de son coût. + Some(k) => { + let stored = if k.wants(&messages) { + host.keep_image(&img).await + } else { + None + }; + k.relay(&mut messages, &mut extra, img, step_start, stored); + } + } } } // Compaction EN COURS DE TOUR, sur le contexte réel du dernier appel @@ -991,6 +1072,21 @@ pub async fn run_turn( used = base + estimate_tokens(&messages[step_start..], &ccfg.echo); s.last_est = used; } + // KEEP_TURN_IMAGES : les images gardées partent AVANT tout résumé ; + // la compaction ne suit que si elle reste nécessaire. + if has_kept_images(&messages) + && compact_needed(ccfg, &messages, used, s.peak_gen) + && let Some((c, freed)) = drop_kept_images(&messages) + { + log_kept_images_dropped("en-tour", freed); + messages = c; + if used > 0 { + used = (used - freed).max(1); + s.last_est = used; + } + extra.clear(); + publish_history(cb, kt, &messages); + } if compact_needed(ccfg, &messages, used, s.peak_gen) && !host.compact_refused(used, &messages) { @@ -1016,7 +1112,7 @@ pub async fn run_turn( s.last_est = 0; messages = c; extra.clear(); - new_history(cb, &messages); + publish_history(cb, kt, &messages); } } continue; @@ -1043,6 +1139,18 @@ pub async fn run_turn( && compact_would_trigger(ccfg, &messages, known) { s.length_replays += 1; + // KEEP_TURN_IMAGES : les images gardées d'abord ; le résumé ne + // suit que s'il reste nécessaire. + if let Some((c, freed)) = drop_kept_images(&messages) { + log_kept_images_dropped("fenêtre-pleine", freed); + messages = c; + extra.clear(); + publish_history(cb, kt, &messages); + if !compact_would_trigger(ccfg, &messages, (known - freed).max(1)) { + cb(StreamEvent::DropReasoning); + continue; + } + } cb(StreamEvent::Compacting(true)); let c = cancel .or_stop(host.compact(CompactPhase::FullWindow, &messages)) @@ -1063,7 +1171,7 @@ pub async fn run_turn( cb(StreamEvent::DropReasoning); messages = c; extra.clear(); - new_history(cb, &messages); + publish_history(cb, kt, &messages); continue; } } diff --git a/rust/loki-rs/src/chat/turn/tests.rs b/rust/loki-rs/src/chat/turn/tests.rs index ee756fa..d5750e8 100644 --- a/rust/loki-rs/src/chat/turn/tests.rs +++ b/rust/loki-rs/src/chat/turn/tests.rs @@ -42,6 +42,12 @@ struct Fake { /// Slot rendu par l'hôte (SIDE_SLOT) ; demandes reçues. slot: Mutex>, slot_asks: Mutex, + /// KEEP_TURN_IMAGES : budget (clé valable pour le tour) ; image relayée + /// par chaque outil ; rangements demandés (refusés si `keep_refuse`). + keep: Mutex>, + relay: Mutex>, + keep_stores: Mutex, + keep_refuse: Mutex, } impl Fake { @@ -120,6 +126,7 @@ impl TurnHost for Fake { .unwrap_or_else(|| format!("résultat de {name}")); ToolOutcome { result, + relay: self.relay.lock().unwrap().clone(), ..Default::default() } } @@ -133,8 +140,8 @@ impl TurnHost for Fake { fn drain_queued(&self) -> Vec { std::mem::take(&mut *self.queued.lock().unwrap()) } - fn image_data_url(&self, _r: &str) -> Option { - None + fn image_data_url(&self, r: &str) -> Option { + (r == "loki-img:aa.png").then(|| IMG_URL.to_string()) } fn effort_remember(&self, want: &str, got: &str) { self.efforts.lock().unwrap().push((want.into(), got.into())); @@ -153,12 +160,83 @@ impl TurnHost for Fake { self.perfs.lock().unwrap().push(rec); self.marks.lock().unwrap().clone() } + async fn keep_images(&self) -> Option { + *self.keep.lock().unwrap() + } + async fn keep_image(&self, m: &Message) -> Option<(Message, i64)> { + *self.keep_stores.lock().unwrap() += 1; + if *self.keep_refuse.lock().unwrap() { + return None; + } + let mut r = m.clone(); + r.content = Some(json!([ + m.content.as_ref().unwrap()[0].clone(), + {"type": "image_url", "image_url": {"url": "loki-img:aa.png"}} + ])); + Some((r, 1000)) + } async fn engine_slot(&self, _ep: &ChatEndpoint) -> Option { *self.slot_asks.lock().unwrap() += 1; *self.slot.lock().unwrap() } } +/// L'image d'un relais : la même data-URL que rend la référence rangée. +const IMG_URL: &str = "data:image/png;base64,AA=="; + +fn relay_msg() -> Message { + Message { + role: "user".into(), + content: Some(json!([ + {"type": "text", "text": "Voici la capture demandée."}, + {"type": "image_url", "image_url": {"url": IMG_URL}} + ])), + ..Default::default() + } +} + +/// Fin de complétion avec cet usage ; `prompt` 0 : aucun usage. +fn finish_u(reason: &str, prompt: i64, completion: i64) -> Vec { + let mut v = vec![format!( + "data: {}", + json!({"choices":[{"delta":{},"finish_reason": reason}]}) + )]; + if prompt > 0 { + v.push(format!( + "data: {}", + json!({"choices":[],"usage":{"prompt_tokens":prompt,"completion_tokens":completion}}) + )); + } + v +} + +fn see_image_step(prompt: i64) -> Script { + let mut l = vec![call("v1", "see_image", r#"{"file":"img.png"}"#)]; + l.extend(finish_u("tool_calls", prompt, 20)); + Script::Ok(l) +} + +fn answer_u(s: &str, prompt: i64) -> Script { + let mut l = vec![content(s)]; + l.extend(finish_u("stop", prompt, 5)); + Script::Ok(l) +} + +/// Lance un tour sur cet historique et rend (résultat, événements). +async fn run_hist( + host: &Fake, + tc: &TurnConfig, + hist: Vec, +) -> (TurnResult, Vec) { + let mut evs = Vec::new(); + let mut cb = |ev: StreamEvent| { + evs.push(ev); + true + }; + let r = run_turn(host, tc, hist, &Cancel::never(), &mut cb).await; + (r, evs) +} + fn content(s: &str) -> String { format!("data: {}", json!({"choices":[{"delta":{"content": s}}]})) } @@ -797,3 +875,183 @@ async fn telemetrie_incomplete() { "pas de statistique finale" ); } + +// ── KEEP_TURN_IMAGES (lot 32, cas de chat_keep_images_test.go) ── + +fn tool_res(s: &str) -> Message { + Message { + role: "tool".into(), + tool_call_id: "v1".into(), + content: Some(json!(s)), + ..Default::default() + } +} + +fn kept_relay(tokens: i64) -> Message { + let mut m = relay_msg(); + m.content.as_mut().unwrap()[1]["image_url"]["url"] = json!("loki-img:aa.png"); + m.img_relay = true; + m.img_tokens = tokens; + m +} + +/// Sans la clé : image éphémère, en data-URL, sans marque ; avec la clé, la +/// requête qui la montre est la MÊME à l'octet près, et l'image est gardée +/// par référence au coût mesuré (écart de prompt moins le texte ajouté). +#[tokio::test] +async fn images_gardees_requete_identique_et_mesuree() { + let text = crate::chat::compact::msg_tokens(&tool_res("résultat de see_image")); + let script = || vec![see_image_step(1000), answer_u("vu", 1020 + text + 300)]; + let off = Fake::new(script()); + *off.relay.lock().unwrap() = Some(relay_msg()); + let (r, _) = run(&off, &config(&["see_image"]), &Cancel::never()).await; + assert_eq!(r.err, None); + assert_eq!(r.extra.len(), 2, "sans la clé, l'image reste éphémère"); + assert!(r.extra.iter().all(|m| !m.img_relay && m.img_tokens == 0)); + assert_eq!(*off.keep_stores.lock().unwrap(), 0); + + let on = Fake::new(script()); + *on.relay.lock().unwrap() = Some(relay_msg()); + *on.keep.lock().unwrap() = Some(3200); + let (r, _) = run(&on, &config(&["see_image"]), &Cancel::never()).await; + assert_eq!(r.err, None); + assert_eq!(on.body(0), off.body(0)); + assert_eq!(on.body(1), off.body(1), "requête différente avec la clé"); + assert_eq!(r.extra.len(), 3, "{:?}", r.extra); + let k = &r.extra[2]; + assert!(k.img_relay && k.img_tokens == 300, "{k:?}"); + assert_eq!( + k.content.as_ref().unwrap()[1]["image_url"]["url"], + "loki-img:aa.png" + ); +} + +/// Budget dépassé, moteur sans usage, rangement refusé : éphémère. +#[tokio::test] +async fn images_gardees_refusees() { + let text = crate::chat::compact::msg_tokens(&tool_res("résultat de see_image")); + for (budget, prompts, refuse) in [ + (200, (1000, 1020 + text + 300), false), + (3200, (0, 0), false), + (3200, (1000, 1020 + text + 300), true), + ] { + let host = Fake::new(vec![see_image_step(prompts.0), answer_u("vu", prompts.1)]); + *host.relay.lock().unwrap() = Some(relay_msg()); + *host.keep.lock().unwrap() = Some(budget); + *host.keep_refuse.lock().unwrap() = refuse; + let (r, _) = run(&host, &config(&["see_image"]), &Cancel::never()).await; + assert_eq!(r.err, None); + assert_eq!( + r.extra.len(), + 2, + "budget={budget} {prompts:?} refus={refuse}" + ); + // Le modèle l'a vue quand même, en data-URL. + assert_eq!( + host.body(1)["messages"][3]["content"][1]["image_url"]["url"], + IMG_URL + ); + } +} + +/// Tour interrompu juste après l'image : jamais mesurée, éphémère. +#[tokio::test] +async fn image_a_l_essai_en_fin_de_tour() { + let host = Fake::new(vec![ + see_image_step(1000), + Script::Status(401, "clé refusée".into()), + ]); + *host.relay.lock().unwrap() = Some(relay_msg()); + *host.keep.lock().unwrap() = Some(3200); + let (r, _) = run(&host, &config(&["see_image"]), &Cancel::never()).await; + assert!(r.err.is_some()); + assert_eq!(r.extra.len(), 2, "{:?}", r.extra); + assert!(r.extra.iter().all(|m| !m.img_relay)); +} + +/// En cours de tour, au seuil : les images gardées partent d'abord, et le +/// résumé n'est pas demandé si leur retrait suffit. +#[tokio::test] +async fn images_retirees_en_cours_de_tour() { + let mut glob = vec![call("g1", "glob", r#"{"pattern":"*.go"}"#)]; + glob.extend(finish_u("tool_calls", 1600, 10)); + let host = Fake::new(vec![Script::Ok(glob), answer_u("fini", 0)]); + let mut tc = config(&["glob"]); + tc.compact.ctx_window = 2000; + let hist = vec![ + um("q"), + tool_res("[ok]"), + kept_relay(1000), + text_msg("assistant", "vu"), + um("suite"), + ]; + let (r, evs) = run_hist(&host, &tc, hist.clone()).await; + assert_eq!(r.err, None); + assert_eq!(host.sent(), 2); + assert!(host.phases.lock().unwrap().is_empty(), "pas de résumé"); + let published = evs + .iter() + .rev() + .find_map(|e| match e { + StreamEvent::NewHistory(h) => Some(h.clone()), + _ => None, + }) + .expect("historique publié"); + assert_eq!(published.len(), hist.len() + 2); + assert_eq!( + published[2].content, + Some(json!(format!( + "Voici la capture demandée.{}", + crate::chat::compact::IMAGE_LOST_MARKER + ))) + ); + assert_eq!( + host.body(1)["messages"][2]["content"], + published[2].content.clone().unwrap() + ); + assert!(r.extra.is_empty(), "{:?}", r.extra); +} + +/// Prompt refusé pour débordement : les images gardées partent d'abord, la +/// requête est rejouée sans elles, sans résumé. +#[tokio::test] +async fn images_retirees_avant_le_filet_reactif() { + let host = Fake::new(vec![ + Script::Status( + 400, + r#"{"error":{"code":400,"message":"request (2500 tokens) exceeds the available context size (2000 tokens), try increasing it"}}"#.into(), + ), + answer_u("fini", 0), + ]); + *host.compacted.lock().unwrap() = Some(vec![um("résumé")]); + let mut tc = config(&[]); + tc.compact.ctx_window = 2000; + let hist = vec![um("q"), tool_res("[ok]"), kept_relay(1000), um("suite")]; + let (r, evs) = run_hist(&host, &tc, hist.clone()).await; + assert_eq!(r.err, None); + assert_eq!(host.sent(), 2); + assert!(host.phases.lock().unwrap().is_empty(), "pas de résumé"); + assert!(evs.iter().any( + |e| matches!(e, StreamEvent::NewHistory(h) if h.len() == hist.len() && h[2].img_tokens == 0) + )); + assert!(host.body(1)["messages"][2]["content"].is_string()); +} + +/// Coupée par une fenêtre pleine : les images gardées d'abord ; leur retrait +/// suffit, l'étape est rejouée sans résumé. +#[tokio::test] +async fn images_retirees_fenetre_pleine() { + let mut cut = vec![]; + cut.extend(finish_u("length", 1500, 200)); + let host = Fake::new(vec![Script::Ok(cut), answer_u("fini", 0)]); + *host.compacted.lock().unwrap() = Some(vec![um("résumé")]); + let mut tc = config(&[]); + tc.compact.ctx_window = 2000; + let hist = vec![um("q"), tool_res("[ok]"), kept_relay(1000), um("suite")]; + let (r, evs) = run_hist(&host, &tc, hist).await; + assert_eq!(r.err, None); + assert_eq!(host.sent(), 2); + assert!(host.phases.lock().unwrap().is_empty(), "pas de résumé"); + assert!(evs.iter().any(|e| matches!(e, StreamEvent::DropReasoning))); + assert_eq!(text(&evs), "fini"); +} diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 32d46eb..8eb6ffa 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -132,6 +132,11 @@ pub struct FakeGo { /// reçues (« conv main »). pub turn_side_live: Mutex, pub turn_slots: Mutex>, + /// Message relayé par chaque outil exécuté ici (image d'une capture). + pub tool_relay: Mutex>, + /// /internal/v1/turn/keepimg : vision active ; rangements reçus. + pub keep_vision: Mutex, + pub keep_stores: Mutex>, } impl FakeGo { @@ -648,11 +653,14 @@ async fn handle(st: Arc, req: Request) -> Response let result = by_name .or_else(|| st.tool_result.lock().unwrap().clone()) .unwrap_or_else(|| format!("exec:{}", v["name"].as_str().unwrap())); - let out = serde_json::json!({ + let mut out = serde_json::json!({ "label": v["args"]["command"], "result": result, "caps_agent": v["caps"]["Agent"], }); + if let Some(r) = st.tool_relay.lock().unwrap().clone() { + out["relay"] = r; + } resp(200, out.to_string()) } "/internal/v1/oai/route" => { @@ -687,6 +695,24 @@ async fn handle(st: Arc, req: Request) -> Response } resp(200, r#"{"ticket":7}"#) } + "/internal/v1/turn/keepimg" => { + let v = serde_json::json!({ + "vision": *st.keep_vision.lock().unwrap(), + "budget": 3276, + }); + resp(200, v.to_string()) + } + "/internal/v1/turn/keepimg/store" => { + let body = req.into_body().collect().await.unwrap().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let mut msg = v["msg"].clone(); + st.keep_stores.lock().unwrap().push(msg.clone()); + msg["content"][1]["image_url"]["url"] = serde_json::json!("loki-img:k1.png"); + resp( + 200, + serde_json::json!({ "ok": true, "msg": msg, "weight": 4 }).to_string(), + ) + } "/internal/v1/turn/slot" => { let body = req.into_body().collect().await.unwrap().to_bytes(); let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index 08e02f5..aef62fe 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -1992,3 +1992,93 @@ async fn telemetrie_des_resumes() { ); assert_eq!(perfs[1]["complete"], false); } + +/// Lot 32 : KEEP_TURN_IMAGES tenu par la façade. L'image relayée par un +/// outil est rangée par le Go, gardée au coût mesuré par le moteur ; au tour +/// suivant, clé retirée, elle part d'abord (légende et marque gardées). +#[tokio::test] +async fn images_gardees_par_la_facade() { + use loki_rs::chat::compact::{IMAGE_LOST_MARKER, msg_tokens}; + use loki_rs::chat::types::Message; + let tool = Message { + role: "tool".into(), + tool_call_id: "c1".into(), + content: Some(serde_json::json!("exec:bash")), + ..Default::default() + }; + let prompt2 = 1020 + msg_tokens(&tool) + 300; + let (url, seen) = fake_engine(vec![ + sse(&[ + serde_json::json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","type":"function","function":{"name":"bash","arguments":"{\"command\":\"ls\"}"}}]}}]}), + serde_json::json!({"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":1000,"completion_tokens":20}}), + ]), + sse(&[ + serde_json::json!({"choices":[{"delta":{"content": "vu"}}]}), + serde_json::json!({"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":prompt2,"completion_tokens":5}}), + ]), + answer("ensuite"), + ]) + .await; + let (svc, go, _stop, t) = service(Some(&url)).await; + // Le fichier que le Go range (ici, le faux ne l'écrit pas). + std::fs::create_dir_all(t.path().join("chatimg")).unwrap(); + std::fs::write(t.path().join("chatimg/k1.png"), b"\x89PNG\r\n").unwrap(); + go.set("config", "KEEP_TURN_IMAGES", "on"); + *go.keep_vision.lock().unwrap() = true; + *go.tool_relay.lock().unwrap() = Some(serde_json::json!({ + "role": "user", + "content": [ + {"type": "text", "text": "Voici la capture demandée."}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0K"}} + ] + })); + svc.send(serde_json::json!({"message": "capture", "cid": "k1"})) + .await + .unwrap(); + until_done(&svc, 1).await; + until(|| go.turn_log.lock().unwrap().contains(&"done".to_string())).await; + // Rangée par le Go, vue en data-URL par le moteur. + assert_eq!(go.keep_stores.lock().unwrap().len(), 1); + let sent = seen.lock().unwrap()[1].1.clone(); + let relay = &sent["messages"].as_array().unwrap()[4]; + assert_eq!( + relay["content"][1]["image_url"]["url"], + "data:image/png;base64,iVBORw0K" + ); + assert!(relay.get("img_relay").is_none(), "marque envoyée au moteur"); + let msgs = svc.sessions.conv().with_state(|st| st.messages.clone()); + let kept = msgs.iter().find(|m| m.img_relay).expect("relais gardé"); + assert_eq!(kept.img_tokens, 300); + assert_eq!( + kept.content.as_ref().unwrap()[1]["image_url"]["url"], + "loki-img:k1.png" + ); + + // Clé retirée : l'image gardée part au début du tour suivant. + go.set("config", "KEEP_TURN_IMAGES", "off"); + svc.send(serde_json::json!({"message": "et ensuite ?", "cid": "k2"})) + .await + .unwrap(); + until_done(&svc, 2).await; + let msgs = svc.sessions.conv().with_state(|st| st.messages.clone()); + let stub = msgs + .iter() + .find(|m| m.img_relay) + .expect("relais gardé en légende"); + assert_eq!(stub.img_tokens, 0); + assert_eq!( + stub.content, + Some(serde_json::json!(format!( + "Voici la capture demandée.{IMAGE_LOST_MARKER}" + ))) + ); + let sent = seen.lock().unwrap()[2].1.clone(); + assert!( + sent["messages"] + .as_array() + .unwrap() + .iter() + .all(|m| !m["content"].to_string().contains("image_url")), + "image encore envoyée : {sent}" + ); +}