Lot 34 : bloc projet figé (PROJ_SNAPSHOT) dans le tour de la façade

PROJ_SNAPSHOT était le dernier réglage qui gardait la conversation au
Go : la façade la tient désormais dans tous les cas (LOKI_CHAT=go reste
la porte de sortie). facadeChatBlocker et la décision prise une fois par
processus disparaissent.

Go (facade_projsnap.go) :
- /internal/v1/turn/view, avec la clé et l'agent : tête du tour portant
  le bloc figé (ou le vivant à rafraîchir), préambule sans date ni
  dossier, et la décision projsnap (rafraîchir, mise à jour à poser,
  instantané à ranger, clé et système du tour) ; dry pour le
  préchauffage ;
- /internal/v1/turn/projsnap/refresh : bloc vivant et instantané qu'il
  devient, pour une compaction en plein tour.

Rust :
- ChatService::assemble_turn, partagé par le tour et la relance du
  builder : instantané rangé avec la discussion, mise à jour posée en tête
  du dernier message utilisateur (prependCtxUpdate), historique nettoyé
  au rafraîchissement ;
- TurnHost::project_refresh après une compaction ou une réduction en
  plein tour (replaceProjectHead) ; l'instantané posé devient celui de la
  discussion ;
- le préchauffage demande la vue en dry et garde les mises à jour tant
  que le bloc n'est pas rafraîchi.

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 03:13:27 +00:00
1 parent ff5c6d4571
commit ad52da18a9
19 files changed
+1063 -171

No files matched your search

+122
View File
@@ -510,6 +510,57 @@ pub(crate) fn strip_context_updates(msgs: Vec<Message>) -> Vec<Message> {
.collect()
}
/// `prependCtxUpdate` : la mise à jour du bloc projet (PROJ_SNAPSHOT) posée en
/// tête d'un message `user`, marqué `ctx_upd` ; `None` : pas un message où la
/// poser.
pub(crate) fn prepend_ctx_update(m: &Message, block: &str) -> Option<Message> {
if m.role != "user" {
return None;
}
let mut out = m.clone();
match &m.content {
Some(Value::String(c)) => out.content = Some(Value::String(format!("{block}{c}"))),
Some(Value::Array(parts)) => {
let mut v = Vec::with_capacity(parts.len() + 1);
v.push(serde_json::json!({ "type": "text", "text": block }));
v.extend(parts.iter().cloned());
out.content = Some(Value::Array(v));
}
_ => return None,
}
out.ctx_upd = true;
Some(out)
}
/// `replaceProjectHead` : le bloc projet de tête (hors rappel des pages lues,
/// qui appartient à l'historique) remplacé par `block`, à la même place.
pub(crate) fn replace_project_head(msgs: Vec<Message>, block: &[Message]) -> Vec<Message> {
let end = msgs.iter().take_while(|m| m.role == "system").count();
let mut out = Vec::with_capacity(msgs.len() + block.len());
let mut inserted = false;
let mut rest = msgs.into_iter();
for m in rest.by_ref().take(end) {
let reminder = matches!(&m.content, Some(Value::String(s)) if s.starts_with(super::wire::MEM_REMINDER_PREFIX));
if super::wire::is_project_system(&m) && !reminder {
if !inserted {
out.extend(block.iter().cloned());
inserted = true;
}
continue;
}
if reminder && !inserted {
out.extend(block.iter().cloned());
inserted = true;
}
out.push(m);
}
if !inserted {
out.extend(block.iter().cloned());
}
out.extend(rest);
out
}
// ---- bornes et assemblage --------------------------------------------------
/// `compactBounds` : frontières tête/queue. Tête = messages système de tête ;
@@ -1844,6 +1895,77 @@ impl<S: Summarizer, R: RecallArchive> Compactor<'_, S, R> {
#[cfg(test)]
mod tests {
use super::*;
// ── PROJ_SNAPSHOT (lot 34, cas de chat_projsnap_test.go) ──
#[test]
fn mise_a_jour_en_tete_du_message() {
let block = format!("{CTX_UPD_OPEN} at=\"x\">\n+ a\n{CTX_UPD_CLOSE}\n\n");
let m = prepend_ctx_update(&um("suite"), &block).expect("message user");
assert!(m.ctx_upd);
assert_eq!(m.content, Some(Value::String(format!("{block}suite"))));
// Retirée à l'identique.
assert_eq!(without_ctx_update(&m).unwrap().content, um("suite").content);
let parts = Message {
role: "user".into(),
content: Some(
serde_json::json!([{"type": "image_url", "image_url": {"url": "loki-img:a.png"}}]),
),
..Default::default()
};
let m = prepend_ctx_update(&parts, &block).unwrap();
assert_eq!(
m.content.as_ref().unwrap()[0],
serde_json::json!({"type": "text", "text": block})
);
assert_eq!(m.content.as_ref().unwrap()[1]["type"], "image_url");
assert!(prepend_ctx_update(&tm("r"), &block).is_none());
let empty = Message {
role: "user".into(),
..Default::default()
};
assert!(prepend_ctx_update(&empty, &block).is_none());
}
#[test]
fn bloc_projet_remplace_a_sa_place() {
use super::super::wire::{MEM_INDEX_PREFIX, MEM_REMINDER_PREFIX, PROJECT_CONTEXT_PREFIX};
let sys = |c: String| Message {
role: "system".into(),
content: Some(Value::String(c)),
..Default::default()
};
let msgs = vec![
sys("base".into()),
sys(format!("{PROJECT_CONTEXT_PREFIX} — vieux")),
sys(format!("{MEM_INDEX_PREFIX} vieux")),
sys(format!("{MEM_REMINDER_PREFIX} a.md")),
um("q"),
];
let block = vec![sys(format!("{MEM_INDEX_PREFIX} neuf"))];
let out = replace_project_head(msgs, &block);
let got: Vec<String> = out.iter().map(msg_text).collect();
assert_eq!(
got,
[
"base".to_string(),
format!("{MEM_INDEX_PREFIX} neuf"),
format!("{MEM_REMINDER_PREFIX} a.md"),
"q".into()
]
);
// Bloc vide au départ : inséré après le système commun.
let out = replace_project_head(vec![sys("base".into()), um("q")], &block);
assert_eq!(msg_text(&out[1]), format!("{MEM_INDEX_PREFIX} neuf"));
assert_eq!(msg_text(&out[2]), "q");
// Rappel des pages lues avant tout bloc : le bloc se pose devant lui.
let out = replace_project_head(
vec![sys(format!("{MEM_REMINDER_PREFIX} a.md")), um("q")],
&block,
);
assert_eq!(msg_text(&out[0]), format!("{MEM_INDEX_PREFIX} neuf"));
assert_eq!(msg_text(&out[1]), format!("{MEM_REMINDER_PREFIX} a.md"));
}
use crate::chat::types::{ToolCall, ToolCallFunc, ToolFunction};
use serde_json::json;
use std::sync::atomic::{AtomicBool, Ordering};
+164 -31
View File
@@ -22,8 +22,10 @@
//! corrections du builder) suit le tour, avant la compaction de fin
//! ([`super::verify`]) ; toute compaction rend l'état des critères au résumé.
//!
//! Écart assumé (reste au Go) : PROJ_SNAPSHOT (refusé par `/turn/view`). Le
//! préchauffage (PREWARM, lot 33) part en fin de tour ([`super::prewarm`]).
//! Le bloc projet figé (PROJ_SNAPSHOT, lot 34) : le Go décide avec la vue du
//! tour, la façade range l'instantané et pose les mises à jour
//! ([`ChatService::assemble_turn`]). Le préchauffage (PREWARM, lot 33) part en
//! fin de tour ([`super::prewarm`]).
//! 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
@@ -46,18 +48,19 @@ use tokio::sync::watch;
use super::compact::{
CompactCaps, CompactConfig, CompactOpts, Compactor, TurnView, compact_needed, compact_opts_for,
ctx_pending, drop_kept_images, estimate_tokens, log_compact, log_ctx_estimate,
log_kept_images_dropped, strip_context_updates,
log_kept_images_dropped, prepend_ctx_update, 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, keep_images_state,
ChatShared, EngineSummarizer, GoTurnHost, SnapTurn, 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::persist::ProjSnapshot;
use super::sessions::{GoSessionHooks, Sessions};
use super::turn::{Cancel, TurnConfig, run_turn};
use super::types::{Message, ReasoningEcho, StreamEvent, Tool};
@@ -177,6 +180,42 @@ pub(super) struct TurnEnv {
pub(super) summarizer: EngineSummarizer,
}
/// La décision PROJ_SNAPSHOT rendue avec la vue du tour (`projsnap`).
#[derive(Debug, Clone, Default, Deserialize)]
pub(super) struct SnapDecision {
/// Instantané neuf : blocs `<context_update>` retirés de l'historique.
#[serde(default)]
pub(super) refresh: bool,
/// Mise à jour à poser en tête du dernier message utilisateur.
#[serde(default)]
pub(super) delta: String,
/// L'instantané à ranger (absent en préchauffage).
#[serde(default)]
pub(super) snap: Option<ProjSnapshot>,
#[serde(default)]
pub(super) key: String,
#[serde(default)]
pub(super) sys_hash: String,
}
/// Les pièces de la vue d'envoi.
pub(super) struct ViewParts {
pub(super) head: Vec<Message>,
pub(super) prefix: String,
pub(super) tools: Vec<Tool>,
pub(super) snap: Option<SnapDecision>,
}
/// Une étape assemblée : ce qui part, les outils, l'historique d'où part la
/// vue (`historyFromView`), et de quoi rafraîchir le bloc projet en plein
/// tour (PROJ_SNAPSHOT).
pub(super) struct Assembled {
pub(super) sent: Vec<Message>,
pub(super) tools: Vec<Tool>,
pub(super) hist: Vec<Message>,
pub(super) snap_turn: Option<SnapTurn>,
}
/// Le chat tenu par la façade : la discussion ouverte, ses tours et sa file.
pub struct ChatService {
pub sessions: Arc<Sessions<GoSessionHooks>>,
@@ -523,23 +562,34 @@ impl ChatService {
Some(compacted)
}
/// `/internal/v1/turn/view` : tête (prompt du preset, consignes,
/// projet), préambule et outils du tour ; `Err` : le message à journaliser.
/// Les pièces de la vue d'envoi (`/internal/v1/turn/view`) : tête du Go
/// (prompt du preset, consignes, projet), préambule, outils ; avec
/// PROJ_SNAPSHOT et l'agent, la décision sur le bloc figé (`snap` :
/// l'instantané de la discussion ; `dry` : préchauffage, rien à ranger).
pub(super) async fn turn_view(
&self,
caps: &Value,
) -> Result<(Vec<Message>, String, Vec<Tool>), String> {
snap: Option<&ProjSnapshot>,
dry: bool,
) -> Result<ViewParts, String> {
match self
.shared
.store
.internal_post("/internal/v1/turn/view", &json!({ "caps": caps }))
.internal_post(
"/internal/v1/turn/view",
&json!({ "caps": caps, "snap": snap, "dry": dry }),
)
.await
{
Ok((s, v)) if s.as_u16() == 200 => Ok((
serde_json::from_value::<Vec<Message>>(v["head"].clone()).unwrap_or_default(),
v["prefix"].as_str().unwrap_or("").to_string(),
serde_json::from_value::<Vec<Tool>>(v["tools"].clone()).unwrap_or_default(),
)),
Ok((s, v)) if s.as_u16() == 200 => Ok(ViewParts {
head: serde_json::from_value::<Vec<Message>>(v["head"].clone()).unwrap_or_default(),
prefix: v["prefix"].as_str().unwrap_or("").to_string(),
tools: serde_json::from_value::<Vec<Tool>>(v["tools"].clone()).unwrap_or_default(),
snap: v
.get("projsnap")
.filter(|p| p.is_object())
.and_then(|p| serde_json::from_value(p.clone()).ok()),
}),
Ok((_, v)) => Err(v["error"]
.as_str()
.unwrap_or("vue du tour refusée")
@@ -548,6 +598,90 @@ impl ChatService {
}
}
/// `turnView` : la séquence envoyée pour une étape de la discussion (tour,
/// relance du builder) — tête du Go, historique, préambule. Avec
/// PROJ_SNAPSHOT : l'instantané de la discussion rangé, rafraîchi (blocs
/// `<context_update>` retirés de l'historique), ou complété d'une mise à
/// jour posée en tête du dernier message utilisateur, dans la discussion
/// et dans `msgs`. Sans la clé, une discussion qui porte encore des blocs
/// d'une époque où elle servait en est nettoyée.
pub(super) async fn assemble_turn(
&self,
epoch: i64,
caps: &Value,
mut msgs: Vec<Message>,
) -> Result<Assembled, String> {
let conv = self.sessions.conv();
let snap = conv.with_state(|st| st.proj_snap.clone());
let v = self.turn_view(caps, snap.as_ref(), false).await?;
let agent = caps.get("Agent").and_then(Value::as_bool).unwrap_or(false);
let mut snap_turn = None;
let hist = match v.snap {
Some(d) => {
let applied = conv.with_state_if(epoch, |st| {
if d.refresh {
st.messages = strip_context_updates(std::mem::take(&mut st.messages));
st.proj_snap = d.snap.clone();
return (true, false);
}
if d.delta.is_empty() {
return (false, false);
}
// Pas de message utilisateur où la poser : l'état annoncé
// ne bouge pas, elle partira au tour suivant.
let Some(m) = st
.messages
.last()
.and_then(|m| prepend_ctx_update(m, &d.delta))
else {
return (false, false);
};
if let Some(last) = st.messages.last_mut() {
*last = m;
}
st.proj_snap = d.snap.clone();
(false, true)
});
if let Some((refresh, updated)) = applied {
if refresh {
msgs = strip_context_updates(msgs);
}
if updated
&& let Some(m) = msgs.last().and_then(|m| prepend_ctx_update(m, &d.delta))
&& let Some(last) = msgs.last_mut()
{
*last = m;
}
}
snap_turn = Some(SnapTurn {
caps: caps.clone(),
key: d.key,
sys_hash: d.sys_hash,
});
msgs
}
None => {
if agent {
conv.with_state_if(epoch, |st| {
if st.proj_snap.is_some() || st.messages.iter().any(|m| m.ctx_upd) {
st.proj_snap = None;
st.messages = strip_context_updates(std::mem::take(&mut st.messages));
}
});
}
strip_context_updates(msgs)
}
};
let mut sent = v.head;
sent.extend(hist.iter().cloned());
Ok(Assembled {
sent: inject_prefix(sent, &v.prefix),
tools: v.tools,
hist,
snap_turn,
})
}
/// Configuration, point de chat et résumeur d'un tour (lus une fois).
pub(super) async fn turn_env(&self) -> TurnEnv {
let store = &self.shared.store;
@@ -827,30 +961,21 @@ impl ChatService {
}
}
// Vue d'envoi : tête du Go (prompt du preset, consignes, projet),
// l'historique sans bloc <context_update>, le préambule.
let (head, prefix, tools) = match self.turn_view(&prep.turn_caps).await {
Ok(v) => v,
// Vue d'envoi : tête du Go (prompt du preset, consignes, projet ou
// bloc figé), l'historique, le préambule.
let Assembled {
sent,
tools,
hist,
snap_turn,
} = match self.assemble_turn(epoch, &prep.turn_caps, msgs).await {
Ok(a) => a,
Err(e) => {
conv.append_delta(epoch, obj("error", e));
self.finish(epoch, &cancel, lease, turn_start).await;
return;
}
};
if agent {
// Une discussion qui porte encore des blocs d'une époque où
// PROJ_SNAPSHOT servait en est nettoyée (comme assembleTurn).
conv.with_state_if(epoch, |st| {
if st.proj_snap.is_some() || st.messages.iter().any(|m| m.ctx_upd) {
st.proj_snap = None;
st.messages = strip_context_updates(std::mem::take(&mut st.messages));
}
});
}
let hist = strip_context_updates(msgs);
let mut sent = head;
sent.extend(hist.iter().cloned());
let sent = inject_prefix(sent, &prefix);
let temperature = if prep.temperature == 0.0 {
0.7
@@ -889,6 +1014,8 @@ impl ChatService {
queued: Box::new(move || me.drain_queued(epoch)),
perf_kind: super::perf::KIND_MAIN,
side_slot: Default::default(),
projsnap: snap_turn,
snap_refreshed: Default::default(),
};
// Chaque événement devient un delta du journal.
@@ -967,6 +1094,9 @@ impl ChatService {
}
}
};
// Bloc projet rafraîchi par une compaction en cours de tour
// (PROJ_SNAPSHOT) : c'est lui que le tour suivant renvoie.
let refreshed = host.take_refreshed_snap();
drop(host);
// Vue modèle rangée : outils puis réponse finale — sauf si un reset
@@ -975,6 +1105,9 @@ impl ChatService {
if let Some(base) = new_base.take() {
st.messages = base;
}
if let Some(s) = refreshed {
st.proj_snap = Some(s);
}
st.messages.extend(res.extra.iter().cloned());
if !content.trim().is_empty() {
let a = Message {
+59 -1
View File
@@ -37,7 +37,7 @@ use super::compact::{
parse_summary_response,
};
use super::engine::{EngineHttp, classify, friendly_error};
use super::persist::{Persister, StoreSink, ToolResJob};
use super::persist::{Persister, ProjSnapshot, StoreSink, ToolResJob};
use super::stream::{CacheHintFn, PreflightFn, RetryPatterns};
use super::turn::{CompactPhase, PerfMarks, PerfRec, Reply, SendError, ToolOutcome, TurnHost};
use super::types::{AskEvent, DiffLine, Message, Tool};
@@ -366,6 +366,30 @@ pub struct GoTurnHost {
/// La requête qui part va sur le second slot (posé par `engine_slot`,
/// lu par `send` pour le ticket « en vol »).
pub side_slot: AtomicBool,
/// PROJ_SNAPSHOT : la clé et le système du tour (`None` sans la clé), et
/// l'instantané posé par un rafraîchissement en cours de tour.
pub projsnap: Option<SnapTurn>,
pub snap_refreshed: Mutex<Option<ProjSnapshot>>,
}
/// PROJ_SNAPSHOT (lot 34) : ce qu'un tour avec la clé garde pour rafraîchir
/// le bloc projet quand une compaction réécrit l'historique en plein tour
/// (`projSnapTurn`).
#[derive(Debug, Clone)]
pub struct SnapTurn {
pub caps: Value,
pub key: String,
pub sys_hash: String,
}
impl GoTurnHost {
/// L'instantané posé en cours de tour (`takeRefreshed`), s'il y en a un.
pub fn take_refreshed_snap(&self) -> Option<ProjSnapshot> {
self.snap_refreshed
.lock()
.unwrap_or_else(|e| e.into_inner())
.take()
}
}
/// Délai d'une demande de slot : le rechargement SLOT_PERSIST relit jusqu'à
@@ -738,6 +762,40 @@ impl TurnHost for GoTurnHost {
self.shared.echo_refusals.refuse(model);
}
async fn project_refresh(&self, msgs: Vec<Message>) -> Vec<Message> {
let Some(t) = &self.projsnap else {
return msgs;
};
let got = self
.shared
.store
.internal_post(
"/internal/v1/turn/projsnap/refresh",
&json!({ "caps": t.caps, "key": t.key, "sys_hash": t.sys_hash }),
)
.await;
let (block, snap) = match got {
Ok((code, v)) if code.is_success() => (
serde_json::from_value::<Vec<Message>>(v["block"].clone()).ok(),
serde_json::from_value::<ProjSnapshot>(v["snap"].clone()).ok(),
),
_ => (None, None),
};
// Go injoignable : le bloc figé reste, sans les mises à jour (le début
// du prompt change de toute façon) ; l'instantané n'est pas remplacé.
let msgs = super::compact::strip_context_updates(msgs);
match (block, snap) {
(Some(block), Some(snap)) => {
*self
.snap_refreshed
.lock()
.unwrap_or_else(|e| e.into_inner()) = Some(snap);
super::compact::replace_project_head(msgs, &block)
}
_ => msgs,
}
}
fn prewarm_keeps(&self, prefix: &[u64], payload: &Value) -> bool {
self.perf_kind == super::perf::KIND_MAIN
&& self.shared.prewarm.live()
+14 -7
View File
@@ -319,10 +319,10 @@ impl ChatService {
if conv.is_generating() {
return Err("moteur occupé");
}
let (mut msgs, ctx_now, peak) = conv.with_state(|st| {
let (mut msgs, ctx_now, peak, snap) = conv.with_state(|st| {
let msgs = st.messages.clone();
let used = ctx_pending(st.ctx_used, st.ctx_used_len, &msgs, &env.echo);
(msgs, used, st.gen_peak)
(msgs, used, st.gen_peak, st.proj_snap.clone())
});
// Sans message utilisateur, le contexte projet irait en tête de la
// sentinelle : rien à préparer.
@@ -338,13 +338,20 @@ impl ChatService {
if compact_needed(&env.ccfg, &msgs, ctx_now, peak) {
return Err("compaction attendue au prochain tour");
}
let (head, prefix, tools) = self
.turn_view(&turn_caps)
// Mêmes décisions que le vrai tour (PROJ_SNAPSHOT compris), sans rien
// ranger : bloc figé gardé avec ses mises à jour, ou rafraîchi.
let v = self
.turn_view(&turn_caps, snap.as_ref(), true)
.await
.map_err(|_| "vue du tour refusée")?;
let mut sent = head;
sent.extend(strip_context_updates(msgs));
let sent = inject_prefix(sent, &prefix);
let tools = v.tools;
let hist = match &v.snap {
Some(d) if !d.refresh => msgs,
_ => strip_context_updates(msgs),
};
let mut sent = v.head;
sent.extend(hist);
let sent = inject_prefix(sent, &v.prefix);
let reasoning = turn_reasoning(&env.cfg, &self.shared.effort);
let pol = turn_echo_policy(false, || env.echo.clone());
self.shared.images.prefetch(&self.shared.store, &sent).await;
+14 -6
View File
@@ -42,8 +42,9 @@
//! préchauffage en vol le garde ([`TurnHost::prewarm_keeps`]) ; toute autre
//! l'annule ([`super::prewarm`]).
//!
//! Non porté : PROJ_SNAPSHOT. Avec cette clé, le Go garde la conversation
//! (`facadeChatBlocker`, décidé au démarrage).
//! Bloc projet figé (PROJ_SNAPSHOT, lot 34) : une compaction ou une réduction
//! en plein tour fait remplacer le bloc figé par le vivant
//! ([`TurnHost::project_refresh`]).
use std::collections::HashMap;
use std::future::Future;
@@ -150,6 +151,13 @@ pub trait TurnHost: Send + Sync {
body: Vec<u8>,
keep_prewarm: bool,
) -> impl Future<Output = Result<Reply, SendError>> + Send;
/// PROJ_SNAPSHOT : une compaction ou une réduction vient de réécrire
/// l'historique en plein tour — le début du prompt change de toute façon,
/// le bloc projet figé est remplacé par le vivant et les mises à jour
/// retirées (`projSnapTurn.refresh`). Sans la clé : tel quel.
fn project_refresh(&self, msgs: Vec<Message>) -> impl Future<Output = Vec<Message>> + Send {
async { msgs }
}
/// PREWARM : la requête qui part (empreintes cumulées de ses messages,
/// corps) prolonge-t-elle exactement le préchauffage en vol ?
fn prewarm_keeps(&self, _prefix: &[u64], _payload: &Value) -> bool {
@@ -691,7 +699,7 @@ async fn run_turn_loop<H: TurnHost>(
if let Some(c) = c {
s.compacted_retry = true;
log_compact(ccfg, "réactif", 0, &messages, &c, true);
messages = c;
messages = host.project_refresh(c).await;
extra.clear();
publish_history(cb, kt, &messages);
continue;
@@ -703,7 +711,7 @@ async fn run_turn_loop<H: TurnHost>(
if changed {
s.shrink_retries += 1;
log_compact(ccfg, "réduction", actual, &messages, &c, true);
messages = c;
messages = host.project_refresh(c).await;
extra.clear();
publish_history(cb, kt, &messages);
continue;
@@ -1124,7 +1132,7 @@ async fn run_turn_loop<H: TurnHost>(
);
if let Some(c) = c {
s.last_est = 0;
messages = c;
messages = host.project_refresh(c).await;
extra.clear();
publish_history(cb, kt, &messages);
}
@@ -1183,7 +1191,7 @@ async fn run_turn_loop<H: TurnHost>(
);
if let Some(c) = c {
cb(StreamEvent::DropReasoning);
messages = c;
messages = host.project_refresh(c).await;
extra.clear();
publish_history(cb, kt, &messages);
continue;
+60
View File
@@ -52,6 +52,9 @@ struct Fake {
/// a reçu.
prewarm_match: Mutex<bool>,
sent_keeps: Mutex<Vec<bool>>,
/// PROJ_SNAPSHOT : clé posée pour ce tour ; rafraîchissements demandés.
snap_on: Mutex<bool>,
refreshes: Mutex<usize>,
}
impl Fake {
@@ -173,6 +176,13 @@ impl TurnHost for Fake {
async fn keep_images(&self) -> Option<i64> {
*self.keep.lock().unwrap()
}
async fn project_refresh(&self, mut msgs: Vec<Message>) -> Vec<Message> {
if *self.snap_on.lock().unwrap() {
*self.refreshes.lock().unwrap() += 1;
msgs.insert(0, text_msg("system", "BLOC VIVANT"));
}
msgs
}
fn prewarm_keeps(&self, prefix: &[u64], payload: &Value) -> bool {
// Ce que l'hôte reçoit : les empreintes des messages envoyés et le
// corps de la requête.
@@ -1094,3 +1104,53 @@ async fn prechauffage_garde_par_la_requete_qui_le_prolonge() {
assert_eq!(r.err, None);
assert_eq!(*host.sent_keeps.lock().unwrap(), [false]);
}
/// PROJ_SNAPSHOT : une compaction ou une réduction en plein tour fait
/// rafraîchir le bloc projet (le début du prompt change de toute façon) ;
/// l'historique publié et la requête rejouée le portent. Le retrait d'images
/// gardées, lui, ne touche pas au début du prompt.
#[tokio::test]
async fn bloc_projet_rafraichi_par_la_compaction() {
let host = Fake::new(vec![
Script::Status(400, "the request exceeds the available context size".into()),
answer("après compaction"),
]);
*host.compacted.lock().unwrap() = Some(vec![um("résumé")]);
*host.snap_on.lock().unwrap() = true;
let (r, evs) = run(&host, &config(&[]), &Cancel::never()).await;
assert_eq!(r.err, None, "{:?}", host.bodies.lock().unwrap());
assert_eq!(*host.refreshes.lock().unwrap(), 1);
assert!(evs.iter().any(|e| matches!(e, StreamEvent::NewHistory(h)
if h.first().and_then(|m| m.content.as_ref()) == Some(&json!("BLOC VIVANT")))));
assert_eq!(host.body(1)["messages"][0]["content"], "BLOC VIVANT");
// Réduction forcée (pas de résumé possible) : rafraîchi aussi.
let host = Fake::new(vec![
Script::Status(
400,
"request (400000 tokens) exceeds the available context size (32768 tokens)".into(),
),
answer("ok"),
]);
*host.snap_on.lock().unwrap() = true;
let hist: Vec<Message> = (0..12)
.map(|i| um(&format!("message {i} {}", "x".repeat(400))))
.collect();
let (r, _) = run_hist(&host, &config(&[]), hist).await;
assert_eq!(r.err, None);
assert_eq!(*host.refreshes.lock().unwrap(), 1);
assert!(
host.phases.lock().unwrap().len() == 1,
"résumé tenté puis réduction"
);
assert_eq!(host.body(1)["messages"][0]["content"], "BLOC VIVANT");
// Images gardées retirées au filet réactif : pas de rafraîchissement.
let host = Fake::new(vec![
Script::Status(400, "the request exceeds the available context size".into()),
answer_u("fini", 0),
]);
*host.snap_on.lock().unwrap() = true;
let hist = vec![um("q"), tool_res("[ok]"), kept_relay(1000), um("suite")];
let (r, _) = run_hist(&host, &config(&[]), hist).await;
assert_eq!(r.err, None);
assert_eq!(*host.refreshes.lock().unwrap(), 0);
}
+36 -13
View File
@@ -23,12 +23,13 @@ use serde::Deserialize;
use serde_json::{Value, json};
use tokio::sync::watch;
use super::compact::{CompactCaps, is_loki_injected, strip_context_updates, without_ctx_update};
use super::generate::{ChatService, TurnEnv, obj};
use super::history::{drop_stray_nudges, history_from_view, inject_prefix, tool_used_delta};
use super::host::{GoTurnHost, TurnCompact};
use super::compact::{CompactCaps, is_loki_injected, without_ctx_update};
use super::generate::{Assembled, ChatService, TurnEnv, obj};
use super::history::{drop_stray_nudges, history_from_view, tool_used_delta};
use super::host::{GoTurnHost, SnapTurn, TurnCompact};
use super::lease::{EngineLease, LeaseSignal};
use super::persist::BK_CHAT;
use super::persist::ProjSnapshot;
use super::turn::{Cancel, TurnConfig, run_turn};
use super::types::{Message, ReasoningEcho, StreamEvent, Tool};
use super::wire::{TurnReasoning, with_echo};
@@ -166,6 +167,8 @@ struct PassOut {
echo: Option<ReasoningEcho>,
saw_usage: bool,
asked: bool,
/// Bloc projet rafraîchi en cours de passe (PROJ_SNAPSHOT).
snap: Option<ProjSnapshot>,
}
fn cap_bool(caps: &Value, k: &str) -> bool {
@@ -304,8 +307,18 @@ impl ChatService {
// Travail annexe : le Go efface le slot à la fin s'il a servi
// (les requêtes de la passe le lui disent en passant).
let side = side_job_begin(&self.shared.store).await;
self.run_pass(pe, epoch, msgs, tools, v["caps"].clone(), true, &[], run)
.await;
self.run_pass(
pe,
epoch,
msgs,
tools,
v["caps"].clone(),
true,
&[],
None,
run,
)
.await;
side_job_end(&self.shared.store, side).await;
}
other => {
@@ -359,18 +372,19 @@ impl ChatService {
return false;
};
// Même contexte que le tour de build : sinon le début du prompt change
// et tout le cache est à refaire.
let (head, prefix, tools) = match self.turn_view(pe.turn_caps).await {
Ok(v) => v,
// et tout le cache est à refaire (PROJ_SNAPSHOT compris).
let Assembled {
sent,
tools,
hist,
snap_turn,
} = match self.assemble_turn(epoch, pe.turn_caps, msgs).await {
Ok(a) => a,
Err(e) => {
conv.append_delta(epoch, obj("error", e));
return false;
}
};
let hist = strip_context_updates(msgs);
let mut sent = head;
sent.extend(hist.iter().cloned());
let sent = inject_prefix(sent, &prefix);
let out = self
.run_pass(
pe,
@@ -380,6 +394,7 @@ impl ChatService {
pe.turn_caps.clone(),
false,
&hist,
snap_turn,
run,
)
.await;
@@ -387,6 +402,10 @@ impl ChatService {
if let Some(base) = out.new_base {
st.messages = base;
}
// Même règle que le tour : le bloc rafraîchi en cours de passe.
if let Some(s) = out.snap {
st.proj_snap = Some(s);
}
st.messages.extend(out.extra);
if !out.content.trim().is_empty() {
let a = Message {
@@ -421,6 +440,7 @@ impl ChatService {
caps: Value,
isolated: bool,
base: &[Message],
snap_turn: Option<SnapTurn>,
run: &mut PassRun<'_>,
) -> PassOut {
let conv = self.sessions.conv().clone();
@@ -463,6 +483,8 @@ impl ChatService {
super::perf::KIND_MAIN
},
side_slot: Default::default(),
projsnap: snap_turn,
snap_refreshed: Default::default(),
};
let mut out = PassOut::default();
let mut cb = |ev: StreamEvent| -> bool {
@@ -546,6 +568,7 @@ impl ChatService {
}
};
out.extra = res.extra;
out.snap = host.take_refreshed_snap();
out
}
}
+4 -5
View File
@@ -247,8 +247,8 @@ fn serve() -> ExitCode {
let sup = child::Supervisor::spawn(go_bin, go_port, env);
wait_for(backend, Duration::from_secs(30)).await;
let engine_paths = engine::Paths::from_env();
// Le Go peut garder la conversation malgré LOKI_CHAT=rust (réglage que
// la façade ne sait pas encore honorer) : sa décision fait foi.
// Le Go confirme qu'il remet la conversation (adresse et jeton bien
// reçus) : sa réponse fait foi.
let chat_rust = chat_rust
&& match go_keeps_chat(backend, &token).await {
Ok(None) => true,
@@ -415,9 +415,8 @@ fn serve() -> ExitCode {
}
/// Conversation tenue par la façade : c'est le défaut. `LOKI_CHAT=go` (ou
/// 0/off/false/no) la rend au Go — porte de sortie. Le Go la garde de toute
/// façon si un réglage que la façade ne sait pas honorer est posé
/// (`facadeChatBlocker`).
/// 0/off/false/no) la rend au Go — porte de sortie. Plus aucun réglage ne la
/// garde au Go (lot 34).
fn chat_held_by_facade() -> bool {
chat_wanted(std::env::var("LOKI_CHAT").ok().as_deref())
}
+23 -1
View File
@@ -141,6 +141,13 @@ pub struct FakeGo {
/// (« check », « begin {corps} », « end {corps} »).
pub prewarm_refuse: Mutex<Option<String>>,
pub prewarm_log: Mutex<Vec<String>>,
/// PROJ_SNAPSHOT : corps reçus par /internal/v1/turn/view ; décisions à
/// rendre (une par appel, dans l'ordre) ; réponse de
/// /internal/v1/turn/projsnap/refresh et corps reçus.
pub turn_view_bodies: Mutex<Vec<serde_json::Value>>,
pub projsnaps: Mutex<std::collections::VecDeque<serde_json::Value>>,
pub projsnap_refresh: Mutex<Option<serde_json::Value>>,
pub projsnap_refreshes: Mutex<Vec<serde_json::Value>>,
}
impl FakeGo {
@@ -360,16 +367,31 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
}
"/internal/v1/turn/view" => {
st.turn_log.lock().unwrap().push("view".into());
let body = req.into_body().collect().await.unwrap().to_bytes();
let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
st.turn_view_bodies.lock().unwrap().push(v);
if let Some(v) = st.turn_view.lock().unwrap().clone() {
return resp(200, v.to_string());
}
let out = serde_json::json!({
let mut out = serde_json::json!({
"head": [{"role": "system", "content": "Tu es Loki."}],
"prefix": "PRÉAMBULE",
"tools": [{"type": "function", "function": {"name": "bash", "description": "", "parameters": {"type": "object"}}}],
});
if let Some(d) = st.projsnaps.lock().unwrap().pop_front() {
out["projsnap"] = d;
}
resp(200, out.to_string())
}
"/internal/v1/turn/projsnap/refresh" => {
let body = req.into_body().collect().await.unwrap().to_bytes();
let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
st.projsnap_refreshes.lock().unwrap().push(v);
match st.projsnap_refresh.lock().unwrap().clone() {
Some(r) => resp(200, r.to_string()),
None => resp(500, r#"{"error":"non prévu"}"#),
}
}
"/internal/v1/chat/holder" => match st.go_keeps_chat.lock().unwrap().clone() {
None => resp(200, r#"{"holder":"rust"}"#),
Some(why) => resp(
+169
View File
@@ -695,6 +695,8 @@ async fn tour_complet_par_le_go() {
queued: Box::new(Vec::new),
perf_kind: "main",
side_slot: Default::default(),
projsnap: None,
snap_refreshed: Default::default(),
};
let bash = Tool {
kind: "function".into(),
@@ -1072,6 +1074,8 @@ async fn hote_et_sonde_de_gabarit() {
queued: Box::new(Vec::new),
perf_kind: "main",
side_slot: Default::default(),
projsnap: None,
snap_refreshed: Default::default(),
}
};
let bash = Tool {
@@ -2172,6 +2176,14 @@ async fn prechauffage_par_la_facade() {
.iter()
.any(|p| p["kind"] == "prewarm" && p["stats"]["prompt_tokens_total"] == 60)
);
// La vue du préchauffage est demandée sans rien à ranger (PROJ_SNAPSHOT).
assert!(
go.turn_view_bodies
.lock()
.unwrap()
.iter()
.any(|b| b["dry"] == true)
);
}
/// Le Go refuse (moteur occupé) : rien ne part ; annulé par une autre
@@ -2215,3 +2227,160 @@ async fn prechauffage_refuse_ou_annule() {
.all(|p| p["kind"] != "prewarm")
);
}
fn proj_snap(desc: &str, delta_tok: i64) -> serde_json::Value {
serde_json::json!({
"at": 1,
"msgs": [{"role": "system", "content": format!("Project context — {desc}")}],
"state": {"desc": desc},
"key": "k1",
"sys_hash": "h1",
"delta_tok": delta_tok
})
}
/// Lot 34 : PROJ_SNAPSHOT tenu par la façade. Le Go décide (instantané pris,
/// mise à jour, rafraîchissement) ; la façade range l'instantané avec la
/// discussion et pose la mise à jour en tête du message, dans la requête et
/// dans l'historique.
#[tokio::test]
async fn bloc_projet_fige_par_la_facade() {
use loki_rs::chat::persist::ProjSnapshot;
let (url, seen) = fake_engine(vec![answer("un"), answer("deux"), answer("trois")]).await;
let (svc, go, _stop, _t) = service(Some(&url)).await;
let delta = "<context_update from=\"loki\" at=\"x\">\n+ neuf\n</context_update>\n\n";
go.projsnaps.lock().unwrap().extend([
serde_json::json!({"refresh": true, "snap": proj_snap("figé", 0), "key": "k1", "sys_hash": "h1"}),
serde_json::json!({"refresh": false, "delta": delta, "snap": proj_snap("neuf", 5), "key": "k1", "sys_hash": "h1"}),
serde_json::json!({"refresh": true, "snap": proj_snap("frais", 0), "key": "k1", "sys_hash": "h1"}),
]);
let stored = |svc: &ChatService| -> ProjSnapshot {
svc.sessions
.conv()
.with_state(|st| st.proj_snap.clone())
.expect("instantané rangé")
};
let as_snap = |v: serde_json::Value| -> ProjSnapshot { serde_json::from_value(v).unwrap() };
svc.send(serde_json::json!({"message": "bonjour", "cid": "s1"}))
.await
.unwrap();
until_done(&svc, 1).await;
assert_eq!(stored(&svc), as_snap(proj_snap("figé", 0)));
let views = go.turn_view_bodies.lock().unwrap().clone();
assert_eq!(views[0]["snap"], serde_json::Value::Null);
assert_eq!(views[0]["dry"], false);
// Un changement : mise à jour en tête du nouveau message, rangée telle
// quelle (le tour suivant renvoie les mêmes octets).
svc.send(serde_json::json!({"message": "suite", "cid": "s2"}))
.await
.unwrap();
until_done(&svc, 2).await;
let views = go.turn_view_bodies.lock().unwrap().clone();
assert_eq!(
as_snap(views[1]["snap"].clone()),
as_snap(proj_snap("figé", 0))
);
let sent = seen.lock().unwrap()[1].1.clone();
let msgs = sent["messages"].as_array().unwrap();
assert_eq!(msgs.last().unwrap()["content"], format!("{delta}suite"));
assert!(
msgs.last().unwrap().get("ctx_upd").is_none(),
"marque envoyée au moteur"
);
let hist = svc.sessions.conv().with_state(|st| st.messages.clone());
let upd = hist.iter().find(|m| m.ctx_upd).expect("mise à jour rangée");
assert_eq!(
upd.content,
Some(serde_json::json!(format!("{delta}suite")))
);
assert_eq!(stored(&svc), as_snap(proj_snap("neuf", 5)));
// Rafraîchi : instantané neuf, mise à jour retirée de l'historique.
svc.send(serde_json::json!({"message": "fin", "cid": "s3"}))
.await
.unwrap();
until_done(&svc, 3).await;
let sent = seen.lock().unwrap()[2].1.clone();
assert!(
sent["messages"]
.as_array()
.unwrap()
.iter()
.any(|m| m["content"] == "suite"),
"{sent}"
);
let hist = svc.sessions.conv().with_state(|st| st.messages.clone());
assert!(hist.iter().all(|m| !m.ctx_upd));
assert_eq!(stored(&svc), as_snap(proj_snap("frais", 0)));
}
/// Réduction forcée en plein tour : le bloc vivant remplace le figé, et
/// devient l'instantané du tour suivant.
#[tokio::test]
async fn bloc_projet_rafraichi_en_plein_tour() {
use loki_rs::chat::persist::ProjSnapshot;
let (url, seen) = fake_engine(vec![
"STATUS 400\nrequest (400000 tokens) exceeds the available context size (32768 tokens)".into(),
serde_json::json!({"choices": [{"message": {"content": "résumé des échanges"}, "finish_reason": "stop"}]})
.to_string(),
answer("ok"),
])
.await;
let (svc, go, _stop, _t) = service(Some(&url)).await;
// Un historique assez long pour qu'il y ait de quoi résumer.
svc.sessions.conv().with_state(|st| {
for i in 0..8 {
st.messages.push(loki_rs::chat::types::Message {
role: if i % 2 == 0 { "user" } else { "assistant" }.into(),
content: Some(serde_json::json!(format!(
"échange {i} {}",
"x".repeat(800)
))),
..Default::default()
});
}
});
go.projsnaps.lock().unwrap().push_back(serde_json::json!(
{"refresh": true, "snap": proj_snap("figé", 0), "key": "k1", "sys_hash": "h1"}
));
*go.projsnap_refresh.lock().unwrap() = Some(serde_json::json!({
"block": [{"role": "system", "content": "Project context — vivant"}],
"snap": proj_snap("vivant", 0)
}));
*go.turn_view.lock().unwrap() = None;
svc.send(serde_json::json!({"message": "bonjour", "cid": "m1"}))
.await
.unwrap();
until_done(&svc, 1).await;
let refreshes = go.projsnap_refreshes.lock().unwrap().clone();
assert_eq!(refreshes.len(), 1, "{refreshes:?}");
assert_eq!(
(
refreshes[0]["key"].clone(),
refreshes[0]["sys_hash"].clone()
),
(serde_json::json!("k1"), serde_json::json!("h1"))
);
let sent = seen.lock().unwrap().last().unwrap().1.clone();
assert!(
sent["messages"]
.as_array()
.unwrap()
.iter()
.any(|m| m["content"]
.as_str()
.is_some_and(|c| c.contains("Project context — vivant"))),
"{sent}"
);
let stored: ProjSnapshot = svc
.sessions
.conv()
.with_state(|st| st.proj_snap.clone())
.unwrap();
assert_eq!(
stored,
serde_json::from_value(proj_snap("vivant", 0)).unwrap()
);
}