Lot 32 : images gardées (KEEP_TURN_IMAGES) dans le tour de la façade

KEEP_TURN_IMAGES ne garde plus la conversation au Go. L'état du tour vit
dans la façade ; les fichiers et la vision restent au Go.

Go (facade_keepimg.go) :
- /internal/v1/turn/keepimg : vision réellement active, budget des images
  gardées ;
- /internal/v1/turn/keepimg/store : une image relayée rangée par
  référence (refRelayImage), relue à l'octet près.

Rust :
- chat/keepimg.rs : port de keepImages (image à l'essai, mesure par le
  moteur répartie au prorata des octets, budget, publication, fin de tour).
- La boucle du tour : budget demandé pour le fil principal, image rangée
  par le Go, historique réécrit publié sans les relais non gardés ; les
  images gardées partent avant tout résumé (filet réactif, en cours de
  tour, fenêtre pleine).
- Début et fin de tour : retirées si le contexte le demande ou si elles
  ne doivent plus partir (clé retirée, vision absente).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv
This commit is contained in:
Claude committed 2026-10-16 02:30:59 +00:00
1 parent a005626e66
commit f79b74dac3
17 files changed
+1435 -37

No files matched your search

+2 -2
View File
@@ -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<Message>, i64)> {
pub fn drop_kept_images(msgs: &[Message]) -> Option<(Vec<Message>, i64)> {
if !msgs.iter().any(|m| m.img_tokens > 0) {
return None;
}
@@ -441,7 +441,7 @@ fn drop_kept_images(msgs: &[Message]) -> Option<(Vec<Message>, 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)");
}
+73 -7
View File
@@ -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<String, String>,
ccfg: &CompactConfig,
msgs: Vec<Message>,
ctx_used: i64,
peak: i64,
) -> (Vec<Message>, 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
+45
View File
@@ -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<i64> {
// 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<u8> {
if ep.external {
self.side_slot.store(false, Ordering::SeqCst);
+272
View File
@@ -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<String, String>) -> 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<Pend>,
/// 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<Message>,
extra: &mut Vec<Message>,
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<Message>,
) {
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<Message>) {
if self.pend.is_empty() {
return;
}
self.stopped = true;
let drop: HashSet<usize> = 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<Message>) {
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<Message> {
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;
+298
View File
@@ -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<Message>, Vec<Message>, 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)]));
}
+1
View File
@@ -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;
+121 -13
View File
@@ -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<PreflightFn> {
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<Output = Option<i64>> + 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<Output = Option<(Message, i64)>> + 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<H: TurnHost>(
host: &H,
tc: &TurnConfig,
messages: Vec<Message>,
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<KeepImages>,
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<H: TurnHost>(
host: &H,
tc: &TurnConfig,
mut messages: Vec<Message>,
cancel: &Cancel,
cb: &mut (dyn FnMut(StreamEvent) -> bool + Send),
kt: &mut Option<KeepImages>,
) -> TurnResult {
let mut extra: Vec<Message> = Vec::new();
let mut s = Loop::default();
@@ -451,9 +501,6 @@ pub async fn run_turn<H: TurnHost>(
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<Message>, err: String| {
cb(StreamEvent::Err(err.clone()));
@@ -524,6 +571,10 @@ pub async fn run_turn<H: TurnHost>(
)
};
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<H: TurnHost>(
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<H: TurnHost>(
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<H: TurnHost>(
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<H: TurnHost>(
}
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<H: TurnHost>(
// 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<H: TurnHost>(
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<H: TurnHost>(
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<H: TurnHost>(
&& 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<H: TurnHost>(
cb(StreamEvent::DropReasoning);
messages = c;
extra.clear();
new_history(cb, &messages);
publish_history(cb, kt, &messages);
continue;
}
}
+260 -2
View File
@@ -42,6 +42,12 @@ struct Fake {
/// Slot rendu par l'hôte (SIDE_SLOT) ; demandes reçues.
slot: Mutex<Option<u8>>,
slot_asks: Mutex<usize>,
/// 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<Option<i64>>,
relay: Mutex<Option<Message>>,
keep_stores: Mutex<usize>,
keep_refuse: Mutex<bool>,
}
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<Message> {
std::mem::take(&mut *self.queued.lock().unwrap())
}
fn image_data_url(&self, _r: &str) -> Option<String> {
None
fn image_data_url(&self, r: &str) -> Option<String> {
(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<i64> {
*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<u8> {
*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<String> {
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<Message>,
) -> (TurnResult, Vec<StreamEvent>) {
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");
}
+27 -1
View File
@@ -132,6 +132,11 @@ pub struct FakeGo {
/// reçues (« conv main »).
pub turn_side_live: Mutex<bool>,
pub turn_slots: Mutex<Vec<String>>,
/// Message relayé par chaque outil exécuté ici (image d'une capture).
pub tool_relay: Mutex<Option<serde_json::Value>>,
/// /internal/v1/turn/keepimg : vision active ; rangements reçus.
pub keep_vision: Mutex<bool>,
pub keep_stores: Mutex<Vec<serde_json::Value>>,
}
impl FakeGo {
@@ -648,11 +653,14 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
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<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
}
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();
+90
View File
@@ -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}"
);
}