use super::*;
use crate::features::tools::confirm::ToolDecision;
use crate::shared::api::EmbedRole;
fn mentions_code(haystack: &str, code: &str) -> bool {
fn fold_dashes(s: &str) -> String {
s.chars()
.map(|c| match c {
'\u{2010}'..='\u{2015}' | '\u{2212}' | '\u{FE58}' | '\u{FE63}' | '\u{FF0D}' => '-',
other => other,
})
.collect()
}
fold_dashes(haystack).contains(&fold_dashes(code))
}
fn live_one_call_per_reply() -> bool {
match std::env::var("MINDFORK_LIVE_ONE_CALL_PER_REPLY") {
Ok(v) => !matches!(v.trim(), "" | "0" | "false"),
Err(_) => false,
}
}
#[test]
fn a_planted_code_survives_the_dash_a_model_chose() {
assert!(mentions_code("the code is ZARYA-8823.", "ZARYA-8823"));
assert!(mentions_code("**ZARYA\u{2011}8823**", "ZARYA-8823"));
assert!(mentions_code("ZARYA\u{2013}8823", "ZARYA-8823"));
assert!(mentions_code("ZARYA\u{2212}8823", "ZARYA-8823"));
assert!(!mentions_code("ZARYA 8823", "ZARYA-8823"));
assert!(!mentions_code("zarya-8823", "ZARYA-8823"));
assert!(!mentions_code("ZARYA-8824", "ZARYA-8823"));
}
async fn fill_then_compact(
cmd_tx: &UnboundedSender<AppCommand>,
evt_rx: &mut UnboundedReceiver<AppEvent>,
topics: &[&str],
) -> (String, usize) {
for topic in topics {
let (r, _) = run_turn_capture(cmd_tx, evt_rx, topic).await;
eprintln!("filler reply: {}", r.chars().take(80).collect::<String>());
}
cmd_tx.send(AppCommand::Compact).unwrap();
let compacted = wait_for(evt_rx, |e| {
matches!(e, AppEvent::Compacted { .. } | AppEvent::Error(_))
})
.await
.unwrap();
let AppEvent::Compacted {
summary, folded, ..
} = compacted
else {
panic!("compaction failed: {compacted:?}");
};
eprintln!(
"folded {folded} messages into {} chars:\n{}",
summary.chars().count(),
summary.chars().take(600).collect::<String>()
);
(summary, folded)
}
async fn narrow_profile_to(
cmd_tx: &UnboundedSender<AppCommand>,
evt_rx: &mut UnboundedReceiver<AppEvent>,
tools: Vec<crate::entities::profile::ToolId>,
) -> (Uuid, Uuid) {
let profile = wait_for(evt_rx, |e| matches!(e, AppEvent::ProfileList(_)))
.await
.and_then(|e| match e {
AppEvent::ProfileList(ps) => ps.first().map(|p| p.id),
_ => None,
})
.expect("the bootstrap profile");
let chat = wait_for(evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.and_then(|e| match e {
AppEvent::ChatActivated { id, .. } => Some(id),
_ => None,
})
.expect("the bootstrap chat");
set_profile_tools(cmd_tx, profile, tools);
(profile, chat)
}
fn set_profile_tools(
cmd_tx: &UnboundedSender<AppCommand>,
profile: Uuid,
tools: Vec<crate::entities::profile::ToolId>,
) {
cmd_tx
.send(AppCommand::UpdateProfile {
id: profile,
edit: Box::new(ProfileEdit {
enabled_tools: Some(tools),
..Default::default()
}),
})
.unwrap();
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL) and embedder (MINDFORK_EMBED_URL)"]
async fn attachment_search_e2e_live() {
use crate::app::events::FileProgress;
use crate::shared::config::AttachmentSettings;
const CODE: &str = "ZARYA-4417";
if std::env::var("MINDFORK_EMBED_URL").is_err() {
eprintln!("skip: MINDFORK_EMBED_URL not set (the index needs a real embedder)");
return;
}
let config = AppConfig {
attachments: AttachmentSettings {
max_file_tokens: 100, excerpt_tokens: 60,
page_tokens: 300,
..Default::default()
},
..Default::default()
};
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let mut body = String::from("ะัะพัะพะบะพะปั ัะพะฒะตัะฐะฝะธะน ะพัะดะตะปะฐ ัะบัะฟะปัะฐัะฐัะธะธ.\n\n");
for i in 1..=120 {
body.push_str(&format!(
"ะัะฝะบั {i}: ะพะฑััะดะธะปะธ ะณัะฐัะธะบ ะดะตะถััััะฒ ะธ ะฟะพััะดะพะบ ะฟะตัะตะดะฐัะธ ัะผะตะฝั, ัะตัะตะฝะธะน ะฝะต ะฟัะธะฝัะปะธ.\n"
));
}
body.push_str(&format!(
"\nะัะฝะบั 121: ะฟะพ ะธัะพะณะฐะผ ะฟัะพะฒะตัะบะธ ั
ะพะปะพะดะธะปัะฝะพะน ัััะฐะฝะพะฒะบะธ ะทะฐะผะตะฝัะฝ ะบะพะผะฟัะตััะพั; \
ะธะฝะฒะตะฝัะฐัะฝัะน ะบะพะด ะทะฐะฟะฐัะฝะพะน ัะฐััะธ: {CODE}.\n"
));
for i in 122..=240 {
body.push_str(&format!(
"ะัะฝะบั {i}: ัะฐััะผะพััะตะปะธ ะทะฐัะฒะบะธ ะฝะฐ ะบะฐะฝัะตะปััะธั ะธ ะผะตะปะบะธะน ัะตะผะพะฝั, ะทะฐะผะตัะฐะฝะธะน ะฝะตั.\n"
));
}
let path = dir.path().join("protocols.txt");
std::fs::write(&path, &body).unwrap();
cmd_tx
.send(AppCommand::FileAttach {
path: path.to_string_lossy().into_owned(),
})
.unwrap();
let indexed = wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::FileProgress(FileProgress::Indexed { .. })
| AppEvent::FileProgress(FileProgress::IndexSkipped { .. })
)
})
.await
.unwrap();
eprintln!("index: {indexed:?}");
assert!(
matches!(
indexed,
AppEvent::FileProgress(FileProgress::Indexed { .. })
),
"the smoke needs a real embedder: {indexed:?}"
);
let (answer, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะ ะฟัะธะบัะตะฟะปัะฝะฝะพะผ ัะฐะนะปะต protocols.txt ะณะดะต-ัะพ ัะฟะพะผัะฝััะฐ ะทะฐะผะตะฝะฐ ะบะพะผะฟัะตััะพัะฐ \
ั
ะพะปะพะดะธะปัะฝะพะน ัััะฐะฝะพะฒะบะธ. ะะฐะนะดะธ ััะพ ะผะตััะพ ะธ ะฝะฐะทะพะฒะธ ะธะฝะฒะตะฝัะฐัะฝัะน ะบะพะด ะทะฐะฟะฐัะฝะพะน ัะฐััะธ.",
)
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let names: Vec<&String> = calls.iter().map(|(n, _)| n).collect();
eprintln!("tool calls: {names:#?}");
eprintln!("reply: {answer}");
assert!(
calls
.iter()
.any(|(n, _)| n == crate::features::tools::attachment::ATTACHMENT_SEARCH_ID),
"the model must find the place by meaning, called: {names:?}"
);
assert!(
mentions_code(&answer, CODE),
"the code sits deep in the file and must be found: {answer}"
);
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL), embedder (MINDFORK_EMBED_URL) and network access"]
async fn fetched_page_is_searched_in_its_birth_turn_e2e_live() {
use crate::app::events::FileProgress;
use crate::features::tools::attachment::{ATTACHMENT_READ_ID, ATTACHMENT_SEARCH_ID};
use crate::shared::i18n::{Lang, locale};
const URL: &str = "https://raw.githubusercontent.com/vshylov/mindfork-rs/main/spec.md";
if std::env::var("MINDFORK_EMBED_URL").is_err() {
eprintln!("skip: MINDFORK_EMBED_URL not set (the search needs the index)");
return;
}
let mut config = AppConfig::default();
config.tools.web_enabled = true;
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![
crate::features::tools::FETCH_URL_ID.into(),
ATTACHMENT_READ_ID.into(),
ATTACHMENT_SEARCH_ID.into(),
],
)
.await;
let indexing: Vec<&str> = Lang::ALL
.iter()
.map(|&l| locale(l).t("tool.attachment.search_indexing"))
.collect();
let hits_header: Vec<String> = Lang::ALL
.iter()
.map(|&l| {
let t = locale(l).t("tool.attachment_search.result.header");
t.split_once("{n}").unwrap().0.to_string()
})
.collect();
let started = std::time::Instant::now();
let (reply, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
&format!(
"ะะฐะณััะทะธ ัะฟะตัะธัะธะบะฐัะธั {URL} ะธ ะฝะฐะนะดะธ ะฒ ะฝะตะน, ะบะฐะบะพะต ะทะฝะฐัะตะฝะธะต ะฟะพ ัะผะพะปัะฐะฝะธั ั \
ะฟะฐัะฐะผะตััะฐ prompt_cap ะผะพะดะตะปะธ ัะตะฑั (ะณะปะฐะฒะฐ 17) ะธ ั ะบะฐะบะพะณะพ ะทะฝะฐัะตะฝะธั ะตะณะพ ะฟะพะดะฝัะปะธ. \
ะะฐะทะพะฒะธ ะพะฑะฐ ัะธัะปะฐ."
),
)
.await;
let took = started.elapsed();
for (name, result) in &calls {
eprintln!(
"call {name} -> {}",
result.chars().take(300).collect::<String>()
);
}
eprintln!("reply: {reply}");
let searched = calls
.iter()
.filter(|(n, _)| n == ATTACHMENT_SEARCH_ID)
.count();
let read = calls
.iter()
.filter(|(n, _)| n == ATTACHMENT_READ_ID)
.count();
eprintln!("birth turn: {searched} search call(s), {read} page read(s), {took:?}");
let fetch = calls
.iter()
.find(|(n, _)| n == crate::features::tools::FETCH_URL_ID)
.expect("the model never fetched the page, so nothing was tested");
assert!(
indexing.iter().any(|s| fetch.1.contains(s)),
"the result must offer the search that is coming: {}",
fetch.1
);
assert!(
calls.iter().any(|(n, r)| n == ATTACHMENT_SEARCH_ID
&& hits_header.iter().any(|h| r.starts_with(h.as_str()))),
"no search in the birth turn found anything: {calls:?}"
);
let digits: String = reply.chars().filter(|c| !c.is_whitespace()).collect();
assert!(
digits.contains("4000") && digits.contains("1200"),
"ยง17.5 says 4000, raised from 1200: {reply}"
);
let mut order = Vec::new();
while order.last() != Some(&"indexed") {
let ev = tokio::time::timeout(std::time::Duration::from_secs(120), evt_rx.recv())
.await
.expect("no index note after the landing")
.expect("the orchestrator went away");
match ev {
AppEvent::FileProgress(FileProgress::Attached { .. }) => order.push("attached"),
AppEvent::FileProgress(FileProgress::Indexed { .. }) => order.push("indexed"),
AppEvent::FileProgress(FileProgress::IndexSkipped { reason, .. }) => {
panic!("the page was not indexed: {reason}")
}
_ => {}
}
}
assert_eq!(order, vec!["attached", "indexed"]);
cmd_tx.send(AppCommand::Quit).unwrap();
let _ = handle.await;
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn attachment_read_e2e_live() {
use crate::shared::config::AttachmentSettings;
const CODE: &str = "ZARYA-8823";
let config = AppConfig {
attachments: AttachmentSettings {
max_file_tokens: 100,
excerpt_tokens: 60,
page_tokens: 300,
..Default::default()
},
..Default::default()
};
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_no_embed(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let mut body = String::from("ะขะตั
ะฝะธัะตัะบะธะต ะทะฐะผะตัะบะธ ะฟัะพะตะบัะฐ.\n\n");
for i in 1..=45 {
body.push_str(&format!(
"ะะฐะผะตัะบะฐ {i}: ัััะธะฝะฝะฐั ะทะฐะฟะธัั ะฑะตะท ะพัะพะฑะพะณะพ ัะพะดะตัะถะฐะฝะธั, ัััะพะบะฐ ะดะปั ะพะฑััะผะฐ.\n"
));
}
body.push_str(&format!("\nะะะฃะขะ ะะะะะ ะะะ ะกะะะ ะะ: {CODE}\n"));
let path = dir.path().join("notes-big.txt");
std::fs::write(&path, &body).unwrap();
cmd_tx
.send(AppCommand::FileAttach {
path: path.to_string_lossy().into_owned(),
})
.unwrap();
let attached = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::FileProgress(_)))
.await
.unwrap();
eprintln!("attach: {attached:?}");
let (answer, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะ ะฟัะธะบัะตะฟะปัะฝะฝะพะผ ัะฐะนะปะต notes-big.txt ัะบะฐะทะฐะฝ ะฒะฝัััะตะฝะฝะธะน ะบะพะด ัะฑะพัะบะธ. \
ะัะพัะธัะฐะน ัะฐะนะป ะธ ะฝะฐะทะพะฒะธ ััะพั ะบะพะด.",
)
.await;
let names: Vec<&String> = calls.iter().map(|(n, _)| n).collect();
eprintln!("turn 1 tool calls: {names:#?}");
eprintln!("turn 1 reply: {answer}");
assert!(
mentions_code(&answer, CODE),
"the answer sits on a late page and must be found: {answer}"
);
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let attachment_tools = [
crate::features::tools::attachment::ATTACHMENT_READ_ID,
crate::features::tools::attachment::ATTACHMENT_SEARCH_ID,
];
assert!(
!calls.is_empty()
&& calls
.iter()
.all(|(n, _)| attachment_tools.contains(&n.as_str())),
"the file must be reached through the attachment tools only, called: {names:?}"
);
assert!(
calls
.iter()
.any(|(n, _)| n == crate::features::tools::attachment::ATTACHMENT_READ_ID),
"a by-reference file with no index must be read page by page, called: {names:?}"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn file_attachment_e2e_live() {
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
const CODE: &str = "ZARYA-7719";
const QUESTION: &str =
"What is the internal build code for project Mindfork? Reply with the code only.";
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (baseline, _) = run_turn_live(&cmd_tx, &mut evt_rx, QUESTION).await;
eprintln!("baseline reply: {baseline}");
cmd_tx
.send(AppCommand::NewChat { profile_id: None })
.unwrap();
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let path = dir.path().join("build-notes.md");
std::fs::write(
&path,
format!(
"# Project Mindfork โ internal notes\n\n\
The internal build code for project Mindfork is {CODE}.\n\
Do not confuse it with the release tag.\n"
),
)
.unwrap();
cmd_tx
.send(AppCommand::FileAttach {
path: path.to_string_lossy().into_owned(),
})
.unwrap();
let attached = wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::FileProgress(crate::app::events::FileProgress::Attached { .. })
| AppEvent::FileProgress(crate::app::events::FileProgress::Failed(_))
)
})
.await
.unwrap();
eprintln!("attach: {attached:?}");
let (answer, _) = run_turn_live(&cmd_tx, &mut evt_rx, QUESTION).await;
eprintln!("attached reply: {answer}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
mentions_code(&answer, CODE),
"the model must answer from the attached file, got: {answer}"
);
assert!(
!mentions_code(&baseline, CODE),
"the baseline must not know the invented code (otherwise the test proves nothing): {baseline}"
);
}
fn figure_png(field: [u8; 3]) -> Vec<u8> {
figure_png_with_corner(field, None)
}
fn figure_png_with_corner(field: [u8; 3], corner: Option<[u8; 3]>) -> Vec<u8> {
let buf = image::ImageBuffer::from_fn(512, 512, |x, y| match corner {
Some(c) if x < 96 && y < 96 => image::Rgb(c),
_ if (160..352).contains(&x) && (160..352).contains(&y) => image::Rgb([255u8, 255, 255]),
_ => image::Rgb(field),
});
let mut bytes = Vec::new();
image::DynamicImage::ImageRgb8(buf)
.write_to(
&mut std::io::Cursor::new(&mut bytes),
image::ImageFormat::Png,
)
.unwrap();
bytes
}
async fn attach_image_live(
cmd_tx: &UnboundedSender<AppCommand>,
rx: &mut UnboundedReceiver<AppEvent>,
path: String,
) -> crate::entities::message_image::ImageInfo {
cmd_tx.send(AppCommand::ImageAttach { path }).unwrap();
let staged = wait_for(rx, |e| {
matches!(
e,
AppEvent::ImageProgress(crate::app::events::ImageProgress::Attached { .. })
| AppEvent::ImageProgress(crate::app::events::ImageProgress::Failed(_))
)
})
.await
.unwrap();
eprintln!("attach: {staged:?}");
match staged {
AppEvent::ImageProgress(crate::app::events::ImageProgress::Attached { info, .. }) => info,
other => panic!("the image was refused before it ever reached the model: {other:?}"),
}
}
#[tokio::test]
#[ignore = "requires MINDFORK_ENGINE_URL and MINDFORK_LIVE_CONTINUE_EXPECT (gateway-continues | gateway-refuses | local); a gateway also needs MINDFORK_ENGINE_KEY + MINDFORK_ENGINE_MODEL"]
async fn continue_through_a_gateway_live() {
use crate::shared::config::ServerMode;
use std::time::Duration;
let Ok(declared) = std::env::var("MINDFORK_LIVE_CONTINUE_EXPECT") else {
eprintln!("skip: MINDFORK_LIVE_CONTINUE_EXPECT not set");
return;
};
let (catalogued, continues) = match declared.trim() {
"gateway-continues" => (true, true),
"gateway-refuses" => (true, false),
"local" => (false, true),
other => panic!("MINDFORK_LIVE_CONTINUE_EXPECT={other:?} is not a declared stack"),
};
let model = std::env::var("MINDFORK_ENGINE_MODEL").ok();
let mut cfg = no_auto_cfg();
cfg.engine.mode = ServerMode::External;
cfg.engine.external.model_name = model.clone();
let model = model.unwrap_or_else(|| "local".into());
cfg.default_sampling.max_tokens = Some(4);
cfg.default_sampling.reasoning_budget = Some(0);
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let landed = tokio::time::timeout(
Duration::from_secs(60),
wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::EngineSamplingFields(_))
}),
)
.await
.expect("the engine's facts must land before the first turn")
.expect("the event stream");
assert_eq!(
matches!(landed, AppEvent::EngineSamplingFields(Some(_))),
catalogued,
"declared {declared:?}, the endpoint answered {landed:?}"
);
cut_then_continue(&cmd_tx, &mut evt_rx, &model, continues).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
}
pub(super) async fn cut_then_continue(
cmd_tx: &UnboundedSender<AppCommand>,
rx: &mut UnboundedReceiver<AppEvent>,
model: &str,
continues: bool,
) {
cmd_tx
.send(AppCommand::SendMessage(
"What is the capital of France? Answer in one short sentence.".into(),
))
.unwrap();
let mut partial = String::new();
let (reason, continuable) = loop {
match rx.recv().await.expect("the turn's events") {
AppEvent::Chunk { text, .. } => partial.push_str(&text),
AppEvent::Finished {
reason,
continuable,
..
} => break (reason, continuable),
_ => {}
}
};
eprintln!("[{model}] first turn ({reason:?}, continuable={continuable}): {partial:?}");
assert_eq!(
reason,
FinishReason::Length,
"the fixture needs a length cut"
);
assert_eq!(
continuable, continues,
"the note after a cut must promise /continue exactly where it works"
);
cmd_tx.send(AppCommand::ContinueLast).unwrap();
let mut resumed = String::new();
let mut note = None;
loop {
match rx.recv().await.expect("the command's events") {
AppEvent::Chunk { text, .. } => resumed.push_str(&text),
AppEvent::Error(text) if !continues => {
note = Some(text);
break;
}
AppEvent::Finished { .. } if continues => break,
_ => {}
}
}
eprintln!("[{model}] /continue: note={note:?} resumed={resumed:?}");
if continues {
assert!(!resumed.is_empty(), "the continuation brought nothing");
assert!(
!resumed.to_lowercase().contains("capital"),
"the reply restarted instead of continuing: {partial:?} + {resumed:?}"
);
} else {
let loc = crate::shared::i18n::locale(crate::shared::i18n::Lang::default());
assert_eq!(
note.as_deref(),
Some(loc.t("ui.cmd.continue_unsupported_gateway"))
);
}
}
fn names(reply: &str, words: &[&str]) -> bool {
let reply = reply.to_lowercase();
words.iter().any(|w| reply.contains(w))
}
const BLUE: &[&str] = &["blue", "ัะธะฝ"];
const GREEN: &[&str] = &["green", "ะทะตะปะตะฝ", "ะทะตะปัะฝ"];
const WHITE: &[&str] = &["white", "ะฑะตะป"];
const SQUARE: &[&str] = &["square", "ะบะฒะฐะดัะฐั"];
#[test]
fn a_reply_names_a_thing_in_either_language() {
let mac_first = "ัะธะฝะธะน ัะพะฝ, ะฑะตะปัะน ะบะฒะฐะดัะฐั.";
assert!(names(mac_first, BLUE) && names(mac_first, SQUARE));
let mac_url = "ะคะพะฝ โ ะทะตะปะตะฝัะน, ะฒ ัะตะฝััะต โ ะบะฒะฐะดัะฐั.";
assert!(names(mac_url, GREEN) && names(mac_url, SQUARE));
assert!(names("ะะตะปะพะณะพ ัะฒะตัะฐ.", WHITE), "a case ending");
assert!(names("A Blue background with a white Square.", BLUE));
assert!(!names("Please upload the image.", BLUE), "no colour named");
assert!(!names("ะัะฐัะฝัะน ะบััะณ.", SQUARE), "another shape");
}
#[tokio::test]
#[ignore = "requires a vision-capable OpenAI-compatible server (MINDFORK_ENGINE_URL + --mmproj)"]
async fn image_attachment_e2e_live() {
if crate::shared::api::live_text_only() {
eprintln!("skip: MINDFORK_LIVE_TEXT_ONLY โ this stack has no vision projector");
return;
}
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let path = dir.path().join("figure.png");
std::fs::write(&path, figure_png([20, 60, 200])).unwrap();
attach_image_live(&cmd_tx, &mut evt_rx, path.to_string_lossy().into_owned()).await;
let (first, _) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"What is the background colour of this image, and what shape is in the centre? \
Answer in a few words.",
)
.await;
eprintln!("turn 1 reply: {first}");
let (second, _) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"What colour was the square in the image I sent? One word.",
)
.await;
eprintln!("turn 2 reply: {second}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
names(&first, BLUE),
"the model must see the background colour, got: {first}"
);
assert!(
names(&first, SQUARE),
"the model must see the centred shape, got: {first}"
);
assert!(
names(&second, WHITE),
"the image must still be visible a turn later, replayed from history, got: {second}"
);
}
#[tokio::test]
#[ignore = "requires MINDFORK_ENGINE_URL + MINDFORK_ENGINE_MODEL on a gateway with a catalogue, and MINDFORK_LIVE_VISION_EXPECT"]
async fn a_gateways_catalogue_decides_whether_images_are_sent_live() {
use crate::app::events::ImageProgress;
let Ok(expect) = std::env::var("MINDFORK_LIVE_VISION_EXPECT") else {
eprintln!("skip: MINDFORK_LIVE_VISION_EXPECT not set");
return;
};
let supported = match expect.as_str() {
"supported" => true,
"unsupported" => false,
other => {
panic!("MINDFORK_LIVE_VISION_EXPECT is `supported` or `unsupported`, not {other:?}")
}
};
let gateway = |cfg: &mut AppConfig| {
cfg.engine.mode = crate::shared::config::ServerMode::External;
cfg.engine.external.model_name = std::env::var("MINDFORK_ENGINE_MODEL").ok();
};
let mut cfg = no_auto_cfg();
gateway(&mut cfg);
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let path = dir.path().join("figure.png");
std::fs::write(&path, figure_png([20, 60, 200])).unwrap();
cmd_tx
.send(AppCommand::ImageAttach {
path: path.to_string_lossy().into_owned(),
})
.unwrap();
let mut attached = None;
let mut refused = None;
let mut unknown_note = false;
loop {
match evt_rx.recv().await.expect("the attach's events") {
AppEvent::ImageProgress(ImageProgress::Attached { info, .. }) => attached = Some(info),
AppEvent::ImageProgress(ImageProgress::VisionUnknown) => unknown_note = true,
AppEvent::ImageProgress(ImageProgress::Failed(msg)) => {
refused = Some(msg);
break;
}
AppEvent::StagedImages(_) if attached.is_some() => break,
_ => {}
}
}
eprintln!("[{expect}] attached={attached:?} refused={refused:?} unknown_note={unknown_note}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
if supported {
assert!(attached.is_some(), "a vision model's image must attach");
assert!(
!unknown_note,
"the catalogue vouched for the model, so the caveat would be untrue"
);
return;
}
let told: Vec<String> = crate::shared::i18n::Lang::ALL
.iter()
.map(|l| {
crate::shared::i18n::locale(*l)
.t("ui.err.image_no_vision")
.to_string()
})
.collect();
assert!(
refused
.as_deref()
.is_some_and(|m| told.iter().any(|t| m.contains(t.as_str()))),
"a text-only model's attach must be refused with the way out: {refused:?}"
);
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_python_chat_with(gateway).await else {
return;
};
cmd_tx
.send(AppCommand::SendMessage(
"Plot these monthly sales as a bar chart and save it as a PNG: Jan 120, Feb 95, \
Mar 180, Apr 130, May 160. Then tell me in one sentence what the chart shows."
.into(),
))
.unwrap();
let mut reply = String::new();
let mut thoughts = String::new();
let mut calls: Vec<(String, usize)> = Vec::new();
let mut errors: Vec<String> = Vec::new();
while let Some(ev) = evt_rx.recv().await {
match ev {
AppEvent::Chunk { text, .. } => reply.push_str(&text),
AppEvent::Thoughts { text, .. } => thoughts.push_str(&text),
AppEvent::ToolCall {
name,
result,
images,
..
} if name == crate::features::tools::PYTHON_EXEC_ID => calls.push((result, images)),
AppEvent::Error(e) => errors.push(e),
AppEvent::Finished { reason, .. } => {
eprintln!("finished: {reason:?}, thoughts: {thoughts}");
break;
}
_ => {}
}
}
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
for (result, images) in &calls {
eprintln!("python_exec โ {images} image(s) sent:\n{result}");
}
eprintln!("reply: {reply}\nerrors: {errors:?}");
assert!(errors.is_empty(), "the turn must not fail: {errors:?}");
assert!(!calls.is_empty(), "the model never ran python_exec");
assert_eq!(
calls.iter().map(|(_, n)| n).sum::<usize>(),
0,
"no image may reach a text-only model"
);
let said = |key: &'static str| -> Vec<&'static str> {
crate::shared::i18n::Lang::ALL
.iter()
.map(|l| crate::shared::i18n::locale(*l).t(key))
.collect()
};
let (withheld, shown) = (
said("loop.image_not_shown_no_vision"),
said("loop.image_shown"),
);
assert!(
calls
.iter()
.any(|(r, _)| withheld.iter().any(|t| r.contains(t))),
"the result must say the chart was not shown"
);
assert!(
!calls
.iter()
.any(|(r, _)| shown.iter().any(|t| r.contains(t))),
"no result may claim a chart was shown to a text-only model"
);
assert!(!reply.trim().is_empty(), "the turn must end with a reply");
}
#[tokio::test]
#[ignore = "requires MINDFORK_ENGINE_URL (a vision engine) and MINDFORK_ENGINE_URL_BLIND (one without)"]
async fn a_history_image_on_an_engine_without_vision_live() {
let Some(blind) =
crate::shared::api::live_client("MINDFORK_ENGINE_URL_BLIND", "MINDFORK_ENGINE_KEY_BLIND")
else {
eprintln!("skip: MINDFORK_ENGINE_URL_BLIND not set");
return;
};
let Some(sighted) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let external = |model: Option<String>| {
let mut cfg = no_auto_cfg();
cfg.engine.mode = crate::shared::config::ServerMode::External;
cfg.engine.external.model_name = model;
cfg
};
let dir = tempfile::tempdir().unwrap();
let (cmd_tx, mut evt_rx, handle) = spawn_orch_at(
dir.path(),
Some(sighted),
external(std::env::var("MINDFORK_ENGINE_MODEL").ok()),
);
let Some(AppEvent::ChatActivated { id: chat, .. }) =
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. })).await
else {
panic!("no chat");
};
let path = dir.path().join("figure.png");
std::fs::write(
&path,
figure_png_with_corner([20, 60, 200], Some([220, 30, 30])),
)
.unwrap();
attach_image_live(&cmd_tx, &mut evt_rx, path.to_string_lossy().into_owned()).await;
let (first, _) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"What is the background colour of this image, and what large shape is in the centre? \
Mention only those two things, in a few words.",
)
.await;
eprintln!("sighted reply: {first}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
names(&first, BLUE),
"the vision engine must see the image before the switch: {first}"
);
let (cmd_tx, mut evt_rx, handle) = spawn_orch_at(
dir.path(),
Some(crate::shared::api::retry::RetryBackend::wrap(Arc::new(
blind,
))),
external(std::env::var("MINDFORK_ENGINE_MODEL_BLIND").ok()),
);
let Some(AppEvent::ChatActivated { id, .. }) =
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. })).await
else {
panic!("no chat after the restart");
};
assert_eq!(
id, chat,
"the restart reopens the chat that holds the image"
);
let notes: Vec<String> = crate::shared::i18n::Lang::ALL
.iter()
.map(|l| crate::shared::i18n::locale(*l).tf("ui.chat.images_withheld", &[("n", "1")]))
.collect();
let turn = |message: &'static str| {
cmd_tx
.send(AppCommand::SendMessage(message.into()))
.unwrap();
};
let collect = async |rx: &mut UnboundedReceiver<AppEvent>| {
let (mut reply, mut errors, mut told) = (String::new(), Vec::new(), 0usize);
while let Some(ev) = rx.recv().await {
match ev {
AppEvent::Chunk { text, .. } => reply.push_str(&text),
AppEvent::Error(e) => errors.push(e),
AppEvent::Notice(n) if notes.contains(&n) => told += 1,
AppEvent::Finished { .. } => break,
_ => {}
}
}
(reply, errors, told)
};
turn(
"What colour is the small square in the top-left corner of the image I sent? \
Answer in one short sentence.",
);
let turn_cap = std::time::Duration::from_secs(600);
let (second, errors, _) = tokio::time::timeout(turn_cap, collect(&mut evt_rx))
.await
.expect("the turn ends");
let told = tokio::time::timeout(
std::time::Duration::from_secs(10),
wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::Notice(n) if notes.contains(n)),
),
)
.await
.is_ok_and(|ev| ev.is_some());
turn("Thanks, that is all.");
let (third, more_errors, told_again) = tokio::time::timeout(turn_cap, collect(&mut evt_rx))
.await
.expect("the turn ends");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!(
"blind reply: {second}\nerrors: {errors:?}\nnext reply: {third}\nerrors: {more_errors:?}"
);
assert!(
errors.is_empty() && more_errors.is_empty(),
"no turn may be refused: {errors:?} {more_errors:?}"
);
assert!(told, "the chat is told its image was not sent");
assert_eq!(told_again, 0, "โฆonce");
let low = second.to_lowercase();
const DECLINES: &[&str] = &[
"cannot see",
"can't see",
"can not see",
"unable to see",
"not able to see",
"cannot view",
"can't view",
"don't have access",
"do not have access",
"not included",
"wasn't included",
"was not included",
"ะฝะต ะฒะธะถั",
"ะฝะต ะผะพะณั ัะฒะธะดะตัั",
"ะฝะต ะผะพะณั ะฒะธะดะตัั",
"ะฝะต ะฟะตัะตะดะฐะฝ",
];
assert!(
DECLINES.iter().any(|d| low.contains(d)),
"asked about a detail it never saw, the model must say it cannot see the image: {second}"
);
let stored: usize = Storage::open(Paths::with_root(dir.path()))
.unwrap()
.json()
.load_chat(chat)
.unwrap()
.unwrap()
.messages
.iter()
.map(|m| m.images.len())
.sum();
assert_eq!(stored, 1, "the stored image survives the switch");
}
#[tokio::test]
#[ignore = "requires a vision model (MINDFORK_ENGINE_URL) and a provisioned sandbox (MINDFORK_SANDBOX_DIR)"]
async fn sandbox_outputs_e2e_live() {
use crate::features::file_command::FileProgress;
const REQUEST: &str = "Plot these monthly sales as a bar chart and save it as a PNG: \
Jan 120, Feb 95, Mar 180, Apr 130, May 160. Then tell me in one sentence what the \
chart shows.";
if crate::shared::api::live_text_only() {
eprintln!("skip: MINDFORK_LIVE_TEXT_ONLY โ this stack has no vision projector");
return;
}
let Some(sandbox) = std::env::var_os("MINDFORK_SANDBOX_DIR").map(std::path::PathBuf::from)
else {
eprintln!("skip: MINDFORK_SANDBOX_DIR not set");
return;
};
for images_on in [true, false] {
let mut cfg = no_auto_cfg();
cfg.tools.python_enabled = true;
cfg.tools.python_mode = crate::shared::config::PythonMode::Wasmer;
cfg.tools.python_net_enabled = false;
cfg.tools.python_images = images_on;
cfg.default_sampling.max_tokens = Some(4096);
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_sandbox(cfg, sandbox.clone())
else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (_profile, chat) = narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![crate::features::tools::PYTHON_EXEC_ID.into()],
)
.await;
cmd_tx
.send(AppCommand::SendMessage(REQUEST.into()))
.unwrap();
let mut reply = String::new();
let mut calls: Vec<(String, usize)> = Vec::new();
while let Some(ev) = evt_rx.recv().await {
match ev {
AppEvent::Chunk { text, .. } => reply.push_str(&text),
AppEvent::ToolCall {
name,
result,
images,
..
} if name == crate::features::tools::PYTHON_EXEC_ID => calls.push((result, images)),
AppEvent::Finished { .. } => break,
_ => {}
}
}
for (result, images) in &calls {
eprintln!("[images {images_on}] result, {images} image(s) sent:\n{result}");
}
eprintln!("[images {images_on}] reply: {reply}");
cmd_tx.send(AppCommand::FileList).unwrap();
let listed = wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::FileProgress(FileProgress::Listed { .. }))
})
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let Some(AppEvent::FileProgress(FileProgress::Listed { stored, .. })) = listed else {
panic!("no /file list reply");
};
let png = stored
.iter()
.find(|f| f.mime == "image/png")
.unwrap_or_else(|| panic!("no PNG listed: {stored:?}"));
assert!(!png.missing, "the listed PNG is not in the folder");
let on_disk = dir
.path()
.join("files")
.join(chat.to_string())
.join(&png.name);
assert!(
std::fs::read(&on_disk).is_ok_and(|b| b.starts_with(b"\x89PNG")),
"not a PNG: {}",
on_disk.display()
);
assert!(
calls.iter().any(|(r, _)| r.contains("files:")),
"no result carried the files section"
);
let sent: usize = calls.iter().map(|(_, n)| n).sum();
if images_on {
assert!(sent >= 1, "the chart must reach the model");
} else {
assert_eq!(sent, 0, "no image may be sent with the switch off");
let told: Vec<String> = crate::shared::i18n::Lang::ALL
.iter()
.map(|l| {
crate::shared::i18n::locale(*l)
.t("tool.python_exec.files.not_shown_off")
.trim_start_matches(" โ ")
.to_string()
})
.collect();
assert!(
calls
.iter()
.any(|(r, _)| told.iter().any(|t| r.contains(t.as_str()))),
"the result must tell the model it has not seen the image"
);
}
}
}
async fn spawn_python_chat() -> Option<(
tempfile::TempDir,
UnboundedSender<AppCommand>,
UnboundedReceiver<AppEvent>,
tokio::task::JoinHandle<()>,
)> {
spawn_python_chat_with(|_| {}).await
}
async fn spawn_python_chat_with(
adjust: impl FnOnce(&mut AppConfig),
) -> Option<(
tempfile::TempDir,
UnboundedSender<AppCommand>,
UnboundedReceiver<AppEvent>,
tokio::task::JoinHandle<()>,
)> {
let sandbox = match std::env::var_os("MINDFORK_SANDBOX_DIR").map(std::path::PathBuf::from) {
Some(dir) => dir,
None => {
eprintln!("skip: MINDFORK_SANDBOX_DIR not set");
return None;
}
};
let mut cfg = no_auto_cfg();
cfg.tools.python_enabled = true;
cfg.tools.python_mode = crate::shared::config::PythonMode::Wasmer;
cfg.tools.python_net_enabled = false;
cfg.default_sampling.max_tokens = Some(4096);
adjust(&mut cfg);
let (dir, cmd_tx, mut evt_rx, handle) =
spawn_orch_live_sandbox(cfg, sandbox).or_else(|| {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
None
})?;
narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![crate::features::tools::PYTHON_EXEC_ID.into()],
)
.await;
Some((dir, cmd_tx, evt_rx, handle))
}
async fn python_turn(
cmd_tx: &UnboundedSender<AppCommand>,
evt_rx: &mut UnboundedReceiver<AppEvent>,
message: &str,
) -> (String, Vec<PythonCall>) {
cmd_tx
.send(AppCommand::SendMessage(message.into()))
.unwrap();
let mut reply = String::new();
let mut results: Vec<PythonCall> = Vec::new();
while let Some(ev) = evt_rx.recv().await {
match ev {
AppEvent::Chunk { text, .. } => reply.push_str(&text),
AppEvent::ToolCall {
name,
arguments,
result,
..
} if name == crate::features::tools::PYTHON_EXEC_ID => {
results.push(PythonCall { arguments, result });
}
AppEvent::Finished { .. } => break,
_ => {}
}
}
for call in &results {
eprintln!("python_exec({})\nโ {}", call.arguments, call.result);
}
eprintln!("reply: {reply}");
(reply, results)
}
#[derive(Debug)]
struct PythonCall {
arguments: String,
result: String,
}
fn first_number(s: &str, digits: usize) -> Option<String> {
let plain: String = s.chars().filter(|c| *c != ',' && *c != ' ').collect();
let mut best: Option<String> = None;
let mut run = String::new();
for c in plain.chars().chain(std::iter::once('.')) {
if c.is_ascii_digit() {
run.push(c);
continue;
}
if run.len() >= digits && best.is_none() {
best = Some(run.clone());
}
run.clear();
}
best
}
#[tokio::test]
#[ignore = "requires a live model (MINDFORK_ENGINE_URL) and a provisioned sandbox (MINDFORK_SANDBOX_DIR)"]
async fn sandbox_network_is_public_only_e2e_live() {
use std::sync::atomic::{AtomicUsize, Ordering};
fn host_listener() -> (u16, std::sync::Arc<AtomicUsize>) {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let hits = std::sync::Arc::new(AtomicUsize::new(0));
let counter = std::sync::Arc::clone(&hits);
std::thread::spawn(move || {
while let Ok((stream, _)) = listener.accept() {
counter.fetch_add(1, Ordering::SeqCst);
drop(stream);
}
});
(port, hits)
}
let ask = |port: u16| {
let code = format!(
r#"import socket, urllib.request
def probe(host, port):
try:
s = socket.create_connection((host, port), timeout=3)
s.close()
return 'CONNECTED'
except Exception as e:
return type(e).__name__
print('loopback:', probe('127.0.0.1', {port}))
try:
print('public:', urllib.request.urlopen('https://example.com', timeout=20).status)
except Exception as e:
print('public:', type(e).__name__)"#
);
format!(
"Use python_exec exactly once, with this code verbatim and nothing added \
or reindented:\n\n{code}\n\nThen report what it printed."
)
};
let (port, hits) = host_listener();
let Some((_dir, cmd_tx, mut evt_rx, handle)) =
spawn_python_chat_with(|cfg| cfg.tools.python_net_enabled = true).await
else {
return;
};
let (_reply, calls) = python_turn(&cmd_tx, &mut evt_rx, &ask(port)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let printed = calls
.iter()
.map(|c| c.result.as_str())
.collect::<Vec<_>>()
.join(
"
",
);
eprintln!(
"public-only arm printed:
{printed}"
);
assert!(
printed.contains("public: 200"),
"the public internet must still work: {printed}"
);
assert!(
!printed.contains("loopback: CONNECTED"),
"the host's loopback was reachable from the sandbox: {printed}"
);
assert_eq!(
hits.load(Ordering::SeqCst),
0,
"the sandbox connected to a service on this machine"
);
let (port, hits) = host_listener();
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_python_chat_with(|cfg| {
cfg.tools.python_net_enabled = true;
cfg.tools.web_allow_private = true;
})
.await
else {
return;
};
let (_reply, calls) = python_turn(&cmd_tx, &mut evt_rx, &ask(port)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let printed = calls
.iter()
.map(|c| c.result.as_str())
.collect::<Vec<_>>()
.join(
"
",
);
eprintln!(
"allow-private arm printed:
{printed}"
);
assert!(
printed.contains("loopback: CONNECTED") || hits.load(Ordering::SeqCst) > 0,
"the control arm reached nothing either โ the arms prove nothing apart: {printed}"
);
}
#[tokio::test]
#[ignore = "requires a live model (MINDFORK_ENGINE_URL) and a provisioned sandbox (MINDFORK_SANDBOX_DIR)"]
async fn sandbox_inputs_e2e_live() {
use crate::features::file_command::FileProgress;
const MAKE: &str = "Use python_exec once. Build an Excel workbook with openpyxl: one \
sheet with the header row month,total and twelve data rows, where row i (1..12) has \
month 2026-i and total = i*i*7 + 13. Save it as /w/out/sales.xlsx. Do NOT print the \
numbers or the sum โ just say the file is saved.";
const READ: &str = "Use python_exec again. The workbook you saved is one of this chat's \
files: name it in the files argument of the call so it is copied into /w/in, read it \
from there with pandas, and print the sum of the total column. Then tell me that sum \
in your reply.";
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_python_chat().await else {
return;
};
let (_made, first) = python_turn(&cmd_tx, &mut evt_rx, MAKE).await;
let (reply, second) = python_turn(&cmd_tx, &mut evt_rx, READ).await;
cmd_tx.send(AppCommand::FileList).unwrap();
let listed = wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::FileProgress(FileProgress::Listed { .. }))
})
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
first.iter().any(|c| c.result.contains("files:")),
"the first call saved nothing to /w/out"
);
let Some(AppEvent::FileProgress(FileProgress::Listed { stored, .. })) = listed else {
panic!("no /file list reply");
};
let workbook = stored
.iter()
.find(|f| f.name.ends_with(".xlsx"))
.unwrap_or_else(|| panic!("no workbook stored: {stored:?}"));
assert!(
!workbook.missing,
"the listed workbook is not in the folder"
);
assert!(
dir.path().join("files").exists(),
"the chat's folder was never created"
);
assert!(
!second.iter().any(|c| c.result.contains("nothing was run")),
"the second call was refused: {second:?}"
);
let printed = second
.iter()
.find_map(|c| first_number(&c.result, 3))
.unwrap_or_else(|| panic!("the second call printed no number: {second:?}"));
assert!(
fold_dashes(&reply).contains(&printed),
"the reply does not carry what the call read back ({printed}): {reply}"
);
}
#[tokio::test]
#[ignore = "requires a live model (MINDFORK_ENGINE_URL) and a Python interpreter on PATH"]
async fn local_mode_files_round_trip_e2e_live() {
use crate::features::file_command::FileProgress;
const ASK: &str = "A table is one of this chat's files. Use python_exec once: name that \
file in the files argument of the call so it is copied into the input folder, read \
it from there with the standard library only (no pandas), print the sum of the \
total column, and write that sum into a file called summary.txt in the output \
folder. Then tell me the sum in your reply.";
let rows: Vec<(u32, u32)> = (1..=12).map(|i| (i, i * i * 7 + 13)).collect();
let total: u32 = rows.iter().map(|(_, t)| t).sum();
let table = std::iter::once("month,total".to_string())
.chain(rows.iter().map(|(m, t)| format!("2026-{m:02},{t}")))
.collect::<Vec<_>>()
.join("\n");
let mut cfg = no_auto_cfg();
cfg.tools.python_enabled = true;
cfg.tools.python_mode = crate::shared::config::PythonMode::Local;
cfg.default_sampling.max_tokens = Some(4096);
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg).or_else(|| {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
None
}) else {
return;
};
narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![crate::features::tools::PYTHON_EXEC_ID.into()],
)
.await;
let source = tempfile::tempdir().unwrap();
let path = source.path().join("sales.csv");
std::fs::write(&path, &table).unwrap();
cmd_tx
.send(AppCommand::FileAttach {
path: path.display().to_string(),
})
.unwrap();
wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::FileProgress(FileProgress::Attached { .. })
| AppEvent::FileProgress(FileProgress::Failed(_))
)
})
.await;
let (reply, results) = python_turn(&cmd_tx, &mut evt_rx, ASK).await;
cmd_tx.send(AppCommand::FileList).unwrap();
let listed = wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::FileProgress(FileProgress::Listed { .. }))
})
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
!results.iter().any(|c| c.result.contains("nothing was run")),
"the call was refused: {results:?}"
);
assert!(
results
.iter()
.any(|c| c.result.contains(&total.to_string())),
"the call never printed the total {total}: {results:?}"
);
assert!(
fold_dashes(&reply).contains(&total.to_string()),
"the reply does not carry the total {total}: {reply}"
);
let Some(AppEvent::FileProgress(FileProgress::Listed { stored, .. })) = listed else {
panic!("no /file list reply");
};
let kept = stored
.iter()
.find(|f| f.name.starts_with("summary"))
.unwrap_or_else(|| panic!("nothing was kept from the output folder: {stored:?}"));
assert!(!kept.missing, "the listed file is not in the folder");
assert!(
dir.path().join("files").exists(),
"the chat's folder was never created"
);
}
#[tokio::test]
#[ignore = "requires a live model (MINDFORK_ENGINE_URL) and a provisioned sandbox (MINDFORK_SANDBOX_DIR)"]
async fn attached_binary_reaches_the_sandbox_live() {
use crate::features::file_command::FileProgress;
const ASK: &str = "A picture is one of this chat's files. Use python_exec: name it in \
the files argument so it is copied into /w/in, open it there with PIL (pillow) and \
print its size in pixels. Then tell me the width and the height.";
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_python_chat().await else {
return;
};
let picture = tempfile::tempdir().unwrap();
let path = picture.path().join("figure.png");
std::fs::write(&path, figure_png([200, 30, 30])).unwrap();
cmd_tx
.send(AppCommand::FileAttach {
path: path.display().to_string(),
})
.unwrap();
let kept = wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::FileProgress(FileProgress::StoredFile { .. })
| AppEvent::FileProgress(FileProgress::Failed(_))
)
})
.await;
let AppEvent::FileProgress(FileProgress::StoredFile { name, mime, .. }) = kept.unwrap() else {
panic!("the picture was refused instead of kept");
};
assert_eq!(name, "figure.png");
assert_eq!(mime, "image/png");
let (reply, results) = python_turn(&cmd_tx, &mut evt_rx, ASK).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
!results.iter().any(|c| c.result.contains("nothing was run")),
"the call was refused: {results:?}"
);
assert!(
results.iter().any(|c| c.result.contains("512")),
"the code never read the picture's real size: {results:?}"
);
assert!(
reply.contains("512"),
"the reply does not carry the size the code read: {reply}"
);
}
#[tokio::test]
#[ignore = "requires a live model (MINDFORK_ENGINE_URL) and a provisioned sandbox (MINDFORK_SANDBOX_DIR)"]
async fn a_handle_survives_a_round_that_adds_a_file_live() {
const ASK: &str = "Do this in two separate python_exec calls, the second written only \
after you have seen the first one's output. \
Call 1: write the text ROUND-ONE into /w/out/marker.txt, then print \
hashlib.sha256(b'ROUND-ONE').hexdigest(). \
Call 2: put #1 in the files argument, open the file that appears in /w/in in \
binary mode, and print its first 8 bytes with repr() followed by the hex digest \
call 1 printed. Then tell me what call 2 printed.";
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_python_chat().await else {
return;
};
let picture = tempfile::tempdir().unwrap();
let path = picture.path().join("figure.png");
std::fs::write(&path, figure_png([30, 120, 60])).unwrap();
attach_image_live(&cmd_tx, &mut evt_rx, path.to_string_lossy().into_owned()).await;
let (reply, results) = python_turn(&cmd_tx, &mut evt_rx, ASK).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
results.len() >= 2,
"the turn had to make two python_exec calls, it made {}: {results:?}",
results.len()
);
let second = &results[1];
assert!(
second.arguments.contains("#1"),
"the second call has to name the handle, not the file: {}",
second.arguments
);
assert!(
second.result.contains("PNG"),
"#1 had to stay the picture the block numbered; the call saw: {}",
second.result
);
assert!(
!second.result.contains("ROUND-ON"),
"#1 was handed the file this very turn created: {}",
second.result
);
assert!(
reply.contains("PNG"),
"the reply has to repeat what the second call printed: {reply}"
);
#[cfg(windows)]
{
use crate::shared::os_open::{FROM_ELSEWHERE, zone_of};
let marker = std::fs::read_dir(dir.path().join("files"))
.expect("the chat's files folder")
.filter_map(Result::ok)
.map(|chat| chat.path().join("marker.txt"))
.find(|path| path.is_file())
.expect("the first call's marker.txt was stored");
assert_eq!(
zone_of(&marker).as_deref(),
Some(FROM_ELSEWHERE),
"{} carries no mark",
marker.display()
);
println!("marked: {}", marker.display());
}
#[cfg(not(windows))]
let _ = dir;
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn i18n_en_profile_title_e2e_live() {
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(no_auto_cfg()) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "English".into(),
system_message: "You are a helpful assistant. Reply in English.".into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
let act = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let chat_id = match act {
AppEvent::ChatActivated { id, .. } => id,
_ => unreachable!(),
};
let (reply, _) =
run_turn_live(&cmd_tx, &mut evt_rx, "Tell me a fun fact about the Moon.").await;
eprintln!("en reply: {:?}", reply.chars().take(80).collect::<String>());
cmd_tx.send(AppCommand::AutoRenameChat(chat_id)).unwrap();
let renamed = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatRenamed { .. }))
.await
.unwrap();
let title = match renamed {
AppEvent::ChatRenamed { title, .. } => title,
_ => unreachable!(),
};
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("en auto-title: {title:?}");
let has_cyr = title
.chars()
.any(|c| ('ะฐ'..='ั').contains(&c) || ('ะ'..='ะฏ').contains(&c));
assert!(!title.trim().is_empty(), "empty title");
assert!(
!has_cyr,
"the English conversation's title contains Cyrillic: {title:?}"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn auto_title_first_reply_e2e_live() {
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let act = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (chat_id, default_title) = match act {
AppEvent::ChatActivated { id, title, .. } => (id, title),
_ => unreachable!(),
};
let (reply, _) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะพัะตะผั ะฝะตะฑะพ ัะธะฝะตะต? ะัะฒะตัั ะพะดะฝะธะผ ะฟัะตะดะปะพะถะตะฝะธะตะผ.",
)
.await;
eprintln!("reply: {:?}", reply.chars().take(80).collect::<String>());
let renamed = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatRenamed { .. }))
.await
.unwrap();
let (id, title) = match renamed {
AppEvent::ChatRenamed { id, title } => (id, title),
_ => unreachable!(),
};
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("automatic title: {title:?} (was {default_title:?})");
assert_eq!(id, chat_id, "the rename must be about the active chat");
assert!(!title.trim().is_empty(), "empty automatic title");
assert_ne!(
title, default_title,
"the default title must actually be replaced"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn utils_en_e2e_live() {
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "English".into(),
system_message: "You are a helpful assistant. Reply in English. \
When you call the current_time tool, call it with no arguments \
at all โ never pass the format parameter."
.into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(vec!["current_time".to_string()]),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
fn no_format(args: &str) -> bool {
let t = args.trim();
if t.is_empty() {
return true;
}
match serde_json::from_str::<serde_json::Value>(t) {
Ok(v) => v
.get("format")
.and_then(|f| f.as_str())
.is_none_or(|s| s.trim().is_empty()),
Err(_) => false,
}
}
let current_time = |calls: Vec<(String, String, String)>| -> Vec<(String, String)> {
calls
.into_iter()
.filter(|(n, _, _)| n == "current_time")
.map(|(_, a, r)| (a, r))
.collect()
};
let (_t, calls) = run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
"What is the current date and time? Call the current_time tool \
with no arguments (do not pass format).",
)
.await;
let mut ct = current_time(calls);
if !ct.iter().any(|(a, _)| no_format(a)) {
eprintln!("current_time called with format only: {ct:#?} โ re-asking");
let (_t2, calls2) = run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
"Call current_time once more with an empty argument object {}, \
passing no format, and show me its raw output.",
)
.await;
ct.extend(current_time(calls2));
}
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("utils calls: {ct:#?}");
let has_cyr = |s: &str| {
s.chars()
.any(|c| ('ะฐ'..='ั').contains(&c) || ('ะ'..='ะฏ').contains(&c))
};
assert!(!ct.is_empty(), "expected a current_time call");
for (a, r) in &ct {
assert!(
!has_cyr(r),
"current_time result has cyrillic (args {a:?}): {r:?}"
);
}
let plain: Vec<&(String, String)> = ct.iter().filter(|(a, _)| no_format(a)).collect();
assert!(
!plain.is_empty(),
"the model passed `format` on every current_time call, so the localized \
label was never rendered โ the i18n claim is untested: {ct:#?}"
);
for (_, r) in plain {
assert!(r.contains("Local time:"), "expected English label: {r:?}");
assert!(r.contains("UTC:"), "expected the UTC line: {r:?}");
}
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn rag_en_e2e_live() {
use crate::features::tools::all_tool_ids;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "English".into(),
system_message: "You are a helpful assistant. Reply in English.".into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(all_tool_ids()),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (_t1, calls1) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"Add this fact to the knowledge base via rag_add: the capital of France is Paris.",
)
.await;
let (_t2, calls2) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"Now search the knowledge base via rag_search for: capital of France.",
)
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let all: Vec<(String, String)> = calls1.into_iter().chain(calls2).collect();
eprintln!("rag calls: {all:#?}");
let has_cyr = |s: &str| {
s.chars()
.any(|c| ('ะฐ'..='ั').contains(&c) || ('ะ'..='ะฏ').contains(&c))
};
for (n, r) in &all {
if n == "rag_add" || n == "rag_search" {
assert!(!has_cyr(r), "rag tool {n} result has cyrillic: {r:?}");
}
}
assert!(
all.iter().any(|(n, _)| n == "rag_add"),
"expected rag_add call"
);
assert!(
all.iter()
.any(|(n, r)| n == "rag_add" && r.contains("Chunks added")),
"expected English rag_add result: {all:?}"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn followup_tool_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (_d, cmd_tx, mut evt_rx, handle) = spawn_orch(Some(backend));
let root = _d.path().to_path_buf();
enable_all_tools(&cmd_tx, &mut evt_rx).await;
cmd_tx
.send(AppCommand::SendMessage(
"ะัะฒะตัั ะบะพัะพัะบะพะน ะฟะตัะฒะพะน ัะตะฟะปะธะบะพะน-ะฟัะธะฒะตัััะฒะธะตะผ, ะทะฐัะตะผ ะะะฏะะะขะะะฌะะ ะฒัะทะพะฒะธ \
ะธะฝััััะผะตะฝั send_followup_message ะธ ะฝะฐะฟะธัะธ ะฒัะพััั ัะตะฟะปะธะบั ั ะธะฝัะตัะตัะฝัะผ \
ัะฐะบัะพะผ ะพ ะบะพัะผะพัะต."
.into(),
))
.unwrap();
let saw_continue = drain_until_finished(&mut evt_rx, |e| {
matches!(e, AppEvent::AssistantContinue { .. })
})
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
let chat = reopened
.json()
.load_chats()
.unwrap()
.into_iter()
.next()
.unwrap();
let new_bubbles = chat.messages.iter().filter(|m| m.new_bubble).count();
eprintln!(
"followup e2e: saw_continue={saw_continue}, new_bubble={new_bubbles}, \
messages={}",
chat.messages.len()
);
for (i, m) in chat.messages.iter().enumerate() {
eprintln!(
" [{i}] {:?} new_bubble={} tools={:?} text={:?}",
m.role,
m.new_bubble,
m.tool_calls.iter().map(|t| &t.name).collect::<Vec<_>>(),
m.text.chars().take(60).collect::<String>()
);
}
assert!(
saw_continue && new_bubbles >= 1,
"expected a second message as a separate bubble (followup)"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn rewrite_tool_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (_d, cmd_tx, mut evt_rx, handle) = spawn_orch(Some(backend));
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
default_sampling: Some(Some(crate::entities::sampling::SamplingConfig {
temperature: Some(0.1),
reasoning_budget: Some(0),
..Default::default()
})),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::SendMessage(
"ะงะตัะฝะพะฒะธะบ ัะฒะพะตะณะพ ะพัะฒะตัะฐ ะฝะธะบัะดะฐ ะฝะต ะณะพะดะธััั. ะะะฏะะะขะะะฌะะ ะฒัะทะพะฒะธ \
ะธะฝััััะผะตะฝั rewrite_current_message, ััะพะฑั ะพัะฑัะพัะธัั ะฝะฐัะฐััะน ะพัะฒะตั ะธ \
ะฝะฐะฟะธัะฐัั ะตะณะพ ะทะฐะฝะพะฒะพ. ะะพะฟัะพั: ัะบะพะปัะบะพ ะฑัะดะตั ะดะฒะฐ ะฟะปัั ะดะฒะฐ?"
.into(),
))
.unwrap();
let saw_rewrite = drain_until_finished(&mut evt_rx, |e| {
matches!(e, AppEvent::AssistantRewrite { .. })
})
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
let chat = reopened
.json()
.load_chats()
.unwrap()
.into_iter()
.next()
.unwrap();
eprintln!(
"rewrite e2e: saw_rewrite={saw_rewrite}, deleted={}, messages={}",
chat.deleted.len(),
chat.messages.len()
);
for (i, m) in chat.messages.iter().enumerate() {
eprintln!(
" msg[{i}] {:?} tools={:?} text={:?}",
m.role,
m.tool_calls.iter().map(|t| &t.name).collect::<Vec<_>>(),
m.text.chars().take(60).collect::<String>()
);
}
assert!(
saw_rewrite && !chat.deleted.is_empty(),
"expected a discarded (rewritten) reply in Chat.deleted"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn self_model_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (_d, cmd_tx, mut evt_rx, handle) = spawn_orch(Some(backend));
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (s1_text, s1_tools) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะตะฝั ะทะพะฒัั ะะปะฐะดะธะผะธั, ั ะฟะธัั ะฝะฐ Rust ะธ ะฝะต ะปัะฑะปั ะผะฝะพะณะพัะปะพะฒะธะต. \
ะะฐะฟะพะผะฝะธ ััะพ: ะฒัะทะพะฒะธ update_user_model (ัะตััั, ะธะฝัะตัะตัั) ะธ update_self_model \
(ะบัะฐัะบะพะต ะพะฟะธัะฐะฝะธะต ัะตะฑั ะธ ัะตะปั โ ะฟะพะผะพะณะฐัั ะผะฝะต ะบัะฐัะบะพ ะธ ะฟะพ ะดะตะปั).",
)
.await;
eprintln!("session 1: tools={s1_tools:?}\ntext={s1_text:?}\n");
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
let (s2_text, s2_tools) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะงัะพ ัั ะพะฑะพ ะผะฝะต ะฟะพะผะฝะธัั ะธ ะบะฐะบะธะต ั ัะตะฑั ัะตะปะธ ะฒ ะพะฑัะตะฝะธะธ ัะพ ะผะฝะพะน?",
)
.await;
eprintln!("session 2: tools={s2_tools:?}\ntext={s2_text:?}\n");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
let model = reopened.db().self_model_get(pid).unwrap();
eprintln!("self_model in DB: {model:#?}");
let model = model.expect("expected a saved self-model after session 1");
assert!(
!model.is_empty(),
"expected a non-empty self-model (the model should have called update_*)"
);
assert!(
s1_tools
.iter()
.any(|t| t == "update_self_model" || t == "update_user_model"),
"expected an update_self_model/update_user_model call in session 1"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn self_model_insight_e2e_live() {
use crate::features::tools::notes::SELF_NOTE_TAG;
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (_d, cmd_tx, mut evt_rx, handle) = spawn_orch(Some(backend));
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (text, tools) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะฏ ะทะฐะผะตัะธะป, ััะพ ะธะฝะพะณะดะฐ ะฟัะพัั ะบัะฐัะบะพ, ะฐ ะธะฝะพะณะดะฐ โ ะฟะพะดัะพะฑะฝะพ. \
ะะฐัะธะบัะธััะน ััะพ ะฝะฐะฑะปัะดะตะฝะธะต ะฒ ัะฒะพะธั
ะฝะฐะฑะปัะดะตะฝะธัั
: ะฒัะทะพะฒะธ ะธะฝััััะผะตะฝั add_insight \
ั ะบะพัะพัะบะธะผ ะพะฟะธัะฐะฝะธะตะผ ััะพะณะพ ะฟัะพัะธะฒะพัะตัะธั.",
)
.await;
eprintln!("insight: tools={tools:?}\ntext={text:?}\n");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
let self_notes = reopened
.db()
.note_list(pid, None, &[SELF_NOTE_TAG.to_string()], None)
.unwrap();
eprintln!("self-notes (observations) in DB: {self_notes:#?}");
assert!(
!self_notes.is_empty(),
"expected at least one self-note (@self) โ an observation from add_insight"
);
assert!(
tools.iter().any(|t| t == "add_insight"),
"expected an add_insight call"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn auto_reflect_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let mut config = AppConfig::default();
config.self_model.auto_reflect_every = 1; let (_d, cmd_tx, mut evt_rx, handle) = spawn_orch_cfg(Some(backend), config);
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t, _tools) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะัะธะฒะตั! ะะตะฝั ะทะพะฒัั ะะปะฐะดะธะผะธั, ะฟะธัั ะฝะฐ Rust ะธ ัะตะฝั ะบัะฐัะบะพััั. ะัะพััะพ ะพัะฒะตัั \
ะบะพัะพัะบะธะผ ะฟัะธะฒะตัััะฒะธะตะผ.",
)
.await;
use crate::features::tools::notes::SELF_NOTE_TAG;
let mut model = None;
let mut self_notes = Vec::new();
for _ in 0..120 {
let db = Storage::open(Paths::with_root(&root)).unwrap();
let m = db.db().self_model_get(pid).unwrap();
self_notes = db
.db()
.note_list(pid, None, &[SELF_NOTE_TAG.to_string()], None)
.unwrap();
let blob_nonempty = m.as_ref().map(|m| !m.is_empty()).unwrap_or(false);
if blob_nonempty || !self_notes.is_empty() {
model = m;
break;
}
drop(db);
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("auto-reflect: self_model={model:#?}\nself-notes={self_notes:#?}");
let blob_nonempty = model.as_ref().map(|m| !m.is_empty()).unwrap_or(false);
assert!(
blob_nonempty || !self_notes.is_empty(),
"expected background auto-reflection to populate the self-model (a blob or an observation note)"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn self_consolidation_e2e_live() {
use crate::features::tools::notes::SELF_NOTE_TAG;
let mut config = AppConfig::default();
config.self_model.auto_consolidate_every = 1; let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let embedder = live_embedder();
let dir = tempfile::tempdir().unwrap();
let storage = Arc::new(Storage::open(Paths::with_root(dir.path())).unwrap());
let (cmd_tx, cmd_rx) = unbounded_channel();
let (evt_tx, mut evt_rx) = unbounded_channel();
let deps = OrchestratorDeps {
cmd_rx,
evt_tx,
storage,
config,
supervisor: Arc::new(MockSupervisor::with_backend_and_embedder(
Some(backend),
embedder,
)),
default_language: crate::shared::i18n::Lang::default(),
extra_tools: Vec::new(),
};
let handle = tokio::spawn(run(deps));
let root = dir.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะฝะฐะฑะปัะดะตะฝะธะต (add_insight): ั ัะตะฝั ะบัะฐัะบะพััั ะฒ ะพัะฒะตัะฐั
. ะัะพััะพ ะทะฐะฟะธัะธ.",
)
.await;
let (_t2, tools2) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะตัั ะพะดะฝะพ ะฝะฐะฑะปัะดะตะฝะธะต (add_insight): ะฟะพะปัะทะพะฒะฐัะตะปั ะฟัะตะดะฟะพัะธัะฐะตั ะปะฐะบะพะฝะธัะฝัะต, \
ะบัะฐัะบะธะต ะพัะฒะตัั. ะัะพััะพ ะทะฐะฟะธัะธ, ะพะฑะฐ ะพััะฐะฒั.",
)
.await;
eprintln!("observations: {tools1:?} + {tools2:?}");
let count_self = |root: &std::path::Path| -> usize {
Storage::open(Paths::with_root(root))
.unwrap()
.db()
.note_list(pid, None, &[SELF_NOTE_TAG.to_string()], None)
.unwrap()
.len()
};
let mut before = 0usize;
for _ in 0..180 {
before = count_self(&root);
if before >= 2 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
let _ = run_turn_live(&cmd_tx, &mut evt_rx, "ะกะฟะฐัะธะฑะพ, ะบะพัะพัะบะพ ะฟะพะดัะฒะตัะดะธ.").await;
let mut after = before;
for _ in 0..180 {
after = count_self(&root);
if after < before {
break; }
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!(
"self-notes: before={before}, after={after} (sleep merged duplicates: {})",
after < before
);
assert!(
before >= 1,
"expected at least one observation (@self) from add_insight"
);
let _ = (tools1, tools2);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn self_model_gate_e2e_live() {
use crate::features::tools::notes::SELF_NOTE_TAG;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะฒ ัะฒะพะธ ะฝะฐะฑะปัะดะตะฝะธั (ะฒัะทะพะฒะธ add_insight): ั ัะบะปะพะฝะตะฝ ะฟัะพัะธัั ะบัะฐัะบะธะต ะพัะฒะตัั.",
)
.await;
eprintln!("session 1: tools={tools1:?}");
let (t2, calls2) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะตัั ะพะดะฝะพ, ะพัะตะฝั ะฟะพั
ะพะถะตะต ะฝะฐะฑะปัะดะตะฝะธะต (ะฒัะทะพะฒะธ add_insight): ะฟะพะปัะทะพะฒะฐัะตะปั \
ะฟัะตะดะฟะพัะธัะฐะตั ะปะฐะบะพะฝะธัะฝัะต, ะบัะฐัะบะธะต ะพัะฒะตัั. ะัะปะธ ะธะฝััััะผะตะฝั ะฟะพะบะฐะถะตั ะฟะพั
ะพะถะตะต \
ะฝะฐะฑะปัะดะตะฝะธะต โ ัะตัะธ ัะฐะผ, ะฟะตัะตะฟะธัะฐัั ะปะธ ะตะณะพ (note_revise/note_supersede) ะธะปะธ \
ะพััะฐะฒะธัั ะพะฑะฐ.",
)
.await;
eprintln!("session 2: text={t2:?}\ncalls={calls2:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
tools1.iter().any(|t| t == "add_insight"),
"session 1: expected an add_insight call"
);
let self_notes = Storage::open(Paths::with_root(&root))
.unwrap()
.db()
.note_list(pid, None, &[SELF_NOTE_TAG.to_string()], None)
.unwrap();
eprintln!(
"self-notes in DB: {} โ {:#?}",
self_notes.len(),
self_notes
.iter()
.map(|n| n.content.clone())
.collect::<Vec<_>>()
);
assert!(
!self_notes.is_empty(),
"expected self-notes (@self) from add_insight"
);
let gate_fired = calls2
.iter()
.any(|(n, r)| n == "add_insight" && r.contains("ะะพั
ะพะถะธะต ะฝะฐะฑะปัะดะตะฝะธั"));
let real_embedder = std::env::var("MINDFORK_EMBED_URL").is_ok();
if !real_embedder && calls2.iter().any(|(n, _)| n == "add_insight") {
assert!(
gate_fired,
"the add_insight gate should have shown the similar observation (MockEmbedder, core of the hypothesis): {calls2:?}"
);
}
let integrated = calls2
.iter()
.any(|(n, _)| n == "note_revise" || n == "note_supersede" || n == "note_merge");
eprintln!(
"gate showed a similar observation: {gate_fired} (real embedder: {real_embedder}); \
model integrated (note_revise/supersede/merge): {integrated}"
);
assert!(
calls2.iter().any(|(n, _)| {
n == "add_insight" || n == "note_revise" || n == "note_supersede" || n == "note_merge"
}),
"session 2: expected add_insight/note_revise/note_supersede/note_merge"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn self_model_gate_en_e2e_live() {
use crate::features::tools::all_tool_ids;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "English".into(),
system_message: "You are a helpful assistant. Reply in English.".into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(all_tool_ids()),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"Record an observation about me (call add_insight): I tend to ask for concise answers.",
)
.await;
eprintln!("turn 1 tools: {tools1:?}");
let (t2, calls2) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"Record another very similar observation (call add_insight): the user prefers brief, \
concise replies. If the tool shows a similar observation, decide yourself whether to \
rewrite it (note_revise/note_supersede) or keep both.",
)
.await;
eprintln!("turn 2 text={t2:?}\ncalls={calls2:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let live = Storage::open(Paths::with_root(_d.path()))
.unwrap()
.db()
.note_list(
pid,
None,
&[crate::features::tools::notes::SELF_NOTE_TAG.to_string()],
None,
)
.unwrap();
eprintln!(
"live observations after turn 2: {} โ {:#?}",
live.len(),
live.iter().map(|n| n.content.as_str()).collect::<Vec<_>>()
);
assert!(
tools1.iter().any(|t| t == "add_insight"),
"turn 1: expected add_insight call"
);
let has_cyr = |s: &str| {
s.chars()
.any(|c| ('ะฐ'..='ั').contains(&c) || ('ะ'..='ะฏ').contains(&c))
};
for (n, r) in &calls2 {
assert!(!has_cyr(r), "tool {n} result has cyrillic: {r:?}");
}
let gate_fired = calls2
.iter()
.any(|(n, r)| n == "add_insight" && r.contains("Similar observations"));
let real_embedder = std::env::var("MINDFORK_EMBED_URL").is_ok();
eprintln!("gate fired (English): {gate_fired}; real embedder: {real_embedder}");
if real_embedder && calls2.iter().any(|(n, _)| n == "add_insight") {
assert!(
gate_fired,
"en gate should have surfaced a similar observation in English: {calls2:?}"
);
}
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn summary_gate_e2e_live() {
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let bloated = "ะฏ ะฐััะธััะตะฝั, ัะตะฝั ัะตััะฝะพััั ะธ ัะพัะฝะพััั. ".repeat(40); let bloated_len = bloated.chars().count();
{
let storage = Storage::open(Paths::with_root(&root)).unwrap();
let mut m = crate::entities::self_model::SelfModel::new(pid);
m.summary = bloated.clone();
storage.db().self_model_upsert(&m).unwrap();
}
let (t, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะัะพัะธัะฐะน ัะฒะพั ยซะผะพะดะตะปั ัะตะฑัยป (ะฒัะทะพะฒะธ get_self_model). ะัะปะธ ะพะฟะธัะฐะฝะธะต ัะตะฑั \
ัะฐะทัะพัะปะพัั โ ัะพะบัะฐัะธ ะตะณะพ ะดะพ ัััะธ ัะตัะตะท update_self_model.summary, ะฐ ัะพะฑััะธะนะฝัะต \
ะฒัะฒะพะดั ะฒัะฝะตัะธ ะฒ ะฝะฐะฑะปัะดะตะฝะธั (add_insight).",
)
.await;
eprintln!("text={t:?}\ncalls={calls:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let gate_fired = calls
.iter()
.any(|(n, r)| n == "get_self_model" && r.contains("ะะฟะธัะฐะฝะธะต ัะตะฑั ัะฐะทัะพัะปะพัั"));
if calls.iter().any(|(n, _)| n == "get_self_model") {
assert!(
gate_fired,
"get_self_model with a bloated description should carry a hint to shrink it: {calls:?}"
);
}
let stored = Storage::open(Paths::with_root(&root))
.unwrap()
.db()
.self_model_get(pid)
.unwrap();
let final_len = stored
.as_ref()
.map(|m| m.summary.chars().count())
.unwrap_or(0);
let shrank = final_len < bloated_len;
let wrote_insight = calls.iter().any(|(n, _)| n == "add_insight");
eprintln!(
"gate showed the hint: {gate_fired}; description {bloated_len} โ {final_len} \
(shrank: {shrank}); moved into observations: {wrote_insight}"
);
assert!(
calls
.iter()
.any(|(n, _)| n == "update_self_model" || n == "add_insight"),
"expected update_self_model/add_insight: {calls:?}"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn self_model_graph_e2e_live() {
use crate::features::tools::notes::SELF_NOTE_TAG;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะฝะฐะฑะปัะดะตะฝะธะต (add_insight): ั ัะตะฝั ะบัะฐัะบะพััั ะฒ ะพัะฒะตัะฐั
.",
)
.await;
let (_t2, tools2) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะตัั ะพะดะฝะพ, ะพัะดะตะปัะฝะพะต ะฝะฐะฑะปัะดะตะฝะธะต (add_insight): ะฝะพ ะธะฝะพะณะดะฐ ั ะดะฐั ัะปะธัะบะพะผ \
ะผะฝะพะณะพัะปะพะฒะฝัะต ะพัะฒะตัั. ะะต ะพะฑัะตะดะธะฝัะน ะตะณะพ ั ะฟัะพัะปัะผ โ ะฟัััั ะฑัะดัั ะดะฒะฐ ะฝะฐะฑะปัะดะตะฝะธั.",
)
.await;
eprintln!("observations: {tools1:?} + {tools2:?}");
let (t3, calls3) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะะพัะผะพััะธ ัะฒะพะธ ะฝะฐะฑะปัะดะตะฝะธั (get_self_model). ะัะปะธ ะดะฒะฐ ะธะท ะฝะธั
ะฟัะพัะธะฒะพัะตัะฐั ะดััะณ \
ะดััะณั โ ัะฒัะถะธ ะธั
ะธะฝััััะผะตะฝัะพะผ note_link (relation=contradicts) ะฟะพ ะฟะพะปะฝะพะผั id.",
)
.await;
eprintln!("linking: text={t3:?}\ncalls={calls3:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
let self_notes = reopened
.db()
.note_list(pid, None, &[SELF_NOTE_TAG.to_string()], None)
.unwrap();
let links = reopened.db().note_links_all(pid).unwrap();
eprintln!(
"self-notes: {}; links in the observation graph: {} โ {links:?}",
self_notes.len(),
links.len()
);
assert!(
self_notes.len() >= 2,
"expected โฅ2 observation notes; session 2 called {tools2:?}"
);
let linked = calls3.iter().any(|(n, _)| n == "note_link");
eprintln!(
"model called note_link: {linked}; links now: {}",
links.len()
);
assert!(
calls3
.iter()
.any(|(n, _)| n == "get_self_model" || n == "note_link"),
"session 3: expected get_self_model/note_link"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn self_consolidation_overview_e2e_live() {
use crate::features::tools::notes::SELF_NOTE_TAG;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะฝะฐะฑะปัะดะตะฝะธะต (add_insight): ั ัะตะฝั ะบัะฐัะบะพััั ะฒ ะพัะฒะตัะฐั
. ะะพะบะฐ ะฝะต ะพะฑัะตะดะธะฝัะน \
ะฝะธ ั ัะตะผ โ ะฟัะพััะพ ะทะฐะฟะธัะธ.",
)
.await;
let (_t2, tools2) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะตัั ะพะดะฝะพ ะฝะฐะฑะปัะดะตะฝะธะต (add_insight): ะฟะพะปัะทะพะฒะฐัะตะปั ะฟัะตะดะฟะพัะธัะฐะตั ะปะฐะบะพะฝะธัะฝัะต, \
ะบัะฐัะบะธะต ะพัะฒะตัั. ะะฐะถะต ะตัะปะธ ะธะฝััััะผะตะฝั ะฟะพะบะฐะถะตั ะฟะพั
ะพะถะตะต โ ะฝะฐ ััะพั ัะฐะท ะพััะฐะฒั ะพะฑะฐ.",
)
.await;
eprintln!("observations: {tools1:?} + {tools2:?}");
let (t3, calls3) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะัะทะพะฒะธ reflect ะธ ะฟัะพัะผะพััะธ ะฑะปะพะบ ยซะะฑะทะพั ะฝะฐะฑะปัะดะตะฝะธะน ะดะปั ะบะพะฝัะพะปะธะดะฐัะธะธยป. ะัะปะธ ััะตะดะธ \
ะฝะฐะฑะปัะดะตะฝะธะน ะตััั ะฟะพั
ะพะถะธะต ะดัะฑะปะธ โ ัะฒะตะดะธ ะธั
(note_merge ะธะปะธ note_supersede).",
)
.await;
eprintln!("reflection: text={t3:?}\ncalls={calls3:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
let self_notes = reopened
.db()
.note_list(pid, None, &[SELF_NOTE_TAG.to_string()], None)
.unwrap();
eprintln!(
"self-notes in DB: {} โ {:#?}",
self_notes.len(),
self_notes
.iter()
.map(|n| n.content.clone())
.collect::<Vec<_>>()
);
assert!(
!self_notes.is_empty(),
"expected self-notes (@self) from add_insight"
);
if let Some((_, result)) = calls3.iter().find(|(n, _)| n == "reflect") {
let has_overview = result.contains("ะะฑะทะพั ะฝะฐะฑะปัะดะตะฝะธะน");
eprintln!("reflect returned the self-consolidation overview: {has_overview}");
if self_notes.len() >= 2 {
assert!(
has_overview,
"reflect with โฅ2 observations should carry the self-consolidation overview: {result}"
);
}
}
let consolidated = calls3
.iter()
.any(|(n, _)| n == "note_merge" || n == "note_supersede" || n == "note_revise");
eprintln!("model merged duplicates (note_merge/supersede/revise): {consolidated}");
assert!(
calls3.iter().any(|(n, _)| {
n == "reflect" || n == "note_merge" || n == "note_supersede" || n == "note_revise"
}),
"session 3: expected reflect/note_merge/note_supersede/note_revise: {calls3:?}"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn trait_gate_e2e_live() {
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฑะฝะพะฒะธ ะผะพะดะตะปั ัะพะฑะตัะตะดะฝะธะบะฐ (ะฒัะทะพะฒะธ update_user_model): ะดะพะฑะฐะฒั ัะตััั (add_traits) \
โ ยซัะตะฝะธั ะบัะฐัะบะพััั ะฒ ะพัะฒะตัะฐั
ยป.",
)
.await;
eprintln!("session 1: tools={tools1:?}");
let (t2, calls2) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะะฑะฝะพะฒะธ ะผะพะดะตะปั ัะพะฑะตัะตะดะฝะธะบะฐ ะตัั ัะฐะท (update_user_model): ะดะพะฑะฐะฒั ะพัะตะฝั ะฟะพั
ะพะถัั \
ัะตััั (add_traits) โ ยซะปัะฑะธั ะปะฐะบะพะฝะธัะฝะพัััยป. ะัะปะธ ะธะฝััััะผะตะฝั ะฟัะตะดัะฟัะตะดะธั ะพ \
ะฟะพััะธ-ะดัะฑะปะต โ ัะตัะธ ัะฐะผ, ะพะฑัะตะดะธะฝะธัั ะปะธ ะธั
ัะตัะตะท remove_traits.",
)
.await;
eprintln!("session 2: text={t2:?}\ncalls={calls2:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
tools1.iter().any(|t| t == "update_user_model"),
"session 1: expected an update_user_model call"
);
let stored = Storage::open(Paths::with_root(&root))
.unwrap()
.db()
.self_model_get(pid)
.unwrap();
let traits = stored
.as_ref()
.map(|m| m.user_model.perceived_traits.clone())
.unwrap_or_default();
eprintln!("interlocutor traits in DB: {traits:?}");
assert!(
!traits.is_empty(),
"expected โฅ1 trait in user_model from update_user_model"
);
let gate_fired = calls2
.iter()
.any(|(n, r)| n == "update_user_model" && r.contains("ะ ะพะดััะฒะตะฝะฝัะต ัะตััั"));
let integrated = traits.len() <= 1;
eprintln!(
"gate warned about a related trait: {gate_fired}; \
model merged them into one trait (no paraphrase pile-up): {integrated}"
);
assert!(
!t2.trim().is_empty() || !calls2.is_empty(),
"session 2 produced nothing at all โ the all-thinking empty turn (lessons ยง9), \
not a verdict on the gate: its finish reason is in the log above; re-run before \
reading it as one"
);
assert!(
calls2.iter().any(|(n, _)| n == "update_user_model"),
"session 2: expected update_user_model"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn cross_organ_link_e2e_live() {
use crate::features::tools::notes::{SELF_NOTE_TAG, is_self_note};
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะทะฐะผะตัะบั ะพ ัะพะฑะตัะตะดะฝะธะบะต (note_save): ะฟะพะปัะทะพะฒะฐัะตะปั ัะตะฝะธั ะบัะฐัะบะพััั ะฒ ะพัะฒะตัะฐั
.",
)
.await;
let (_t2, tools2) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะฝะฐะฑะปัะดะตะฝะธะต ะพ ัะตะฑะต (add_insight): ั ัะบะปะพะฝะตะฝ ะดะฐะฒะฐัั ะผะฝะพะณะพัะปะพะฒะฝัะต ะพัะฒะตัั.",
)
.await;
eprintln!("saving: {tools1:?} + {tools2:?}");
let (t3, calls3) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะะพัะผะพััะธ ัะฒะพะธ ะฝะฐะฑะปัะดะตะฝะธั (get_self_model) ะธ ะทะฐะผะตัะบะธ ะพ ัะพะฑะตัะตะดะฝะธะบะต (note_recall). \
ะัะปะธ ะฝะฐะฑะปัะดะตะฝะธะต ะพ ัะตะฑะต ะฟัะพัะธะฒะพัะตัะธั ัะฐะบัั ะพ ัะพะฑะตัะตะดะฝะธะบะต โ ัะฒัะถะธ ะธั
note_link \
(relation=contradicts) ะฟะพ ะธั
id.",
)
.await;
eprintln!("linking: text={t3:?}\ncalls={calls3:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
let self_notes = reopened
.db()
.note_list(pid, None, &[SELF_NOTE_TAG.to_string()], None)
.unwrap();
let links = reopened.db().note_links_all(pid).unwrap();
let cross = links
.iter()
.filter(|(f, t, _)| {
let fs = reopened
.db()
.note_get(pid, *f)
.ok()
.flatten()
.as_ref()
.map(is_self_note);
let ts = reopened
.db()
.note_get(pid, *t)
.ok()
.flatten()
.as_ref()
.map(is_self_note);
matches!((fs, ts), (Some(a), Some(b)) if a != b)
})
.count();
eprintln!(
"self-notes: {}; links total: {}; cross-organ: {cross} โ {links:?}",
self_notes.len(),
links.len()
);
assert!(
!self_notes.is_empty(),
"expected a self-note from add_insight"
);
let all = reopened.db().note_list(pid, None, &[], None).unwrap();
assert!(
all.iter().any(|n| !is_self_note(n)),
"expected a user note from note_save"
);
let linked = calls3.iter().any(|(n, _)| n == "note_link");
eprintln!("model called note_link: {linked}; cross-organ edges: {cross}");
assert!(
calls3
.iter()
.any(|(n, _)| n == "get_self_model" || n == "note_recall" || n == "note_link"),
"session 3: expected get_self_model/note_recall/note_link"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn recall_includes_self_e2e_live() {
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let _pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let config = AppConfig {
notes: crate::shared::config::NotesSettings {
recall_includes_self: true,
..Default::default()
},
..Default::default()
};
cmd_tx
.send(AppCommand::UpdateConfig(Box::new(config)))
.unwrap();
wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::Settings { config, .. } if config.notes.recall_includes_self),
)
.await
.unwrap();
run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะทะฐะผะตัะบั ะพ ัะพะฑะตัะตะดะฝะธะบะต (note_save): ะฟะพะปัะทะพะฒะฐัะตะปั ะปัะฑะธั ะบัะฐัะบะพััั.",
)
.await;
run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะฐะฟะธัะธ ะฝะฐะฑะปัะดะตะฝะธะต ะพ ัะตะฑะต (add_insight): ั ัะบะปะพะฝะตะฝ ะบ ะผะฝะพะณะพัะปะพะฒะธั.",
)
.await;
let (t3, calls3) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะะพะธัะธ ะฒ ะทะฐะผะตัะบะฐั
(note_recall) ะฒัั ะฟัะพ ะบัะฐัะบะพััั ะธ ะผะฝะพะณะพัะปะพะฒะธะต โ ััะพ ัะฐะผ ะตััั?",
)
.await;
eprintln!("recall: text={t3:?}\ncalls={calls3:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let self_marked = calls3
.iter()
.any(|(n, r)| n == "note_recall" && r.contains("[ะพ ัะตะฑะต]"));
eprintln!("note_recall showed the self-observation with its marker: {self_marked}");
assert!(
calls3.iter().any(|(n, _)| n == "note_recall"),
"expected a note_recall call"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn note_cite_source_e2e_live() {
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let root = _d.path().to_path_buf();
let pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let (_t1, tools1) = run_turn_live(
&cmd_tx,
&mut evt_rx,
"ะะพะฑะฐะฒั ะฒ ะฑะฐะทั ะทะฝะฐะฝะธะน (rag_add) ัะตะบัั ยซะกัะพะปะธัะฐ ะคัะฐะฝัะธะธ โ ะะฐัะธะถ.ยป ั ะธััะพัะฝะธะบะพะผ ยซัะฐะบััยป.",
)
.await;
eprintln!("rag_add: {tools1:?}");
let (t2, calls2) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะะฐะนะดะธ ะฒ ะฑะฐะทะต ะทะฝะฐะฝะธะน (rag_search) ะฟัะพ ััะพะปะธัั ะคัะฐะฝัะธะธ. ะะฐะฟะธัะธ ะบัะฐัะบะธะน ะฒัะฒะพะด \
ะทะฐะผะตัะบะพะน (note_save), ะทะฐัะตะผ ัะฒัะถะธ ััั ะทะฐะผะตัะบั ั ะธััะพัะฝะธะบะพะผ ัะตัะตะท \
note_cite_source (ะธััะพัะฝะธะบ ะฝะฐะทัะฒะฐะตััั ยซัะฐะบััยป).",
)
.await;
eprintln!("citing: text={t2:?}\ncalls={calls2:#?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(&root)).unwrap();
const SOURCE: &str = "ัะฐะบัั";
assert!(
reopened.db().rag_source_exists(pid, SOURCE).unwrap(),
"expected the source to be in the knowledge base"
);
let citing = reopened.db().notes_citing_source(pid, SOURCE).unwrap();
let cited = calls2.iter().any(|(n, _)| n == "note_cite_source");
eprintln!(
"model called note_cite_source: {cited}; notes citing {SOURCE:?}: {} โ {:?}",
citing.len(),
citing.iter().map(|n| n.content.clone()).collect::<Vec<_>>()
);
assert!(
calls2
.iter()
.any(|(n, _)| n == "rag_search" || n == "note_cite_source"),
"session 2: expected rag_search/note_cite_source"
);
}
#[tokio::test]
#[ignore = "requires a running embedding server (MINDFORK_EMBED_URL)"]
async fn summary_obs_calibration_e2e_live() {
use crate::features::tools::notes::cosine;
let Some(embedder) = live_embedder() else {
eprintln!("skip: MINDFORK_EMBED_URL not set");
return;
};
let should_match: &[(&str, &str)] = &[
(
"ะฏ ัะตะฝั ััะฝะพััั ะธ ะบัะฐัะบะพััั: ะฟัะตะดะฟะพัะธัะฐั ะดะฐะฒะฐัั ัะถะฐััะต, ะฟะพ ัััะตััะฒั ะพัะฒะตัั ะฑะตะท ะฒะพะดั.",
"ะะฐะผะตัะฐั ะทะฐ ัะพะฑะพะน ัะบะปะพะฝะฝะพััั ะพัะฒะตัะฐัั ะปะฐะบะพะฝะธัะฝะพ ะธ ะฟะพ ะดะตะปั, ะธะทะฑะตะณะฐั ะผะฝะพะณะพัะปะพะฒะธั.",
),
(
"ะะฝะต ะฒะฐะถะฝะพ ะฑััั ัะตััะฝัะผ ะดะฐะถะต ะบะพะณะดะฐ ััะพ ะฝะตัะดะพะฑะฝะพ โ ัะพัะฝะพััั ะฒััะต ัะณะพะดะปะธะฒะพััะธ.",
"ะกัะฐัะฐััั ะณะพะฒะพัะธัั ะฟัะฐะฒะดั ะฟััะผะพ, ะฝะต ัะผัะณัะฐั ะตั ัะฐะดะธ ัะพะณะพ, ััะพะฑั ะฟะพะฝัะฐะฒะธัััั.",
),
(
"ะฏ ัะบะปะพะฝะตะฝ ะณะปัะฑะพะบะพ ะฟะพะณััะถะฐัััั ะฒ ะทะฐะดะฐัั ะธ ะดะพะฒะพะดะธัั ัะฐัััะถะดะตะฝะธะต ะดะพ ะบะพะฝัะฐ.",
"ะะฝะต ัะฒะพะนััะฒะตะฝะฝะพ ััะฐัะตะปัะฝะพ ะธ ะดะพ ะบะพะฝัะฐ ะฟัะพัะฐะฑะฐััะฒะฐัั ะฟัะพะฑะปะตะผั, ะฝะต ะฑัะพัะฐั ะฝะฐ ะฟะพะปะฟััะธ.",
),
];
let should_not: &[(&str, &str)] = &[
(
"ะฏ ัะตะฝั ััะฝะพััั ะธ ะบัะฐัะบะพััั ะฒ ัะฒะพะธั
ะพัะฒะตัะฐั
, ัััะตะผะปััั ะบ ัะถะฐัะพััะธ ะธะทะปะพะถะตะฝะธั.",
"ะะพะปัะทะพะฒะฐัะตะปั ัะฒะปะตะบะฐะตััั ะฐะปัะฟะธะฝะธะทะผะพะผ ะธ ะปัะฑะธั ะดะปะธะฝะฝัะต ะณะพัะฝัะต ะฟะพั
ะพะดั ะฟะพ ะฒัั
ะพะดะฝัะผ.",
),
(
"ะะฝะต ะฒะฐะถะฝะพ ะฑััั ัะตััะฝัะผ ะดะฐะถะต ะบะพะณะดะฐ ััะพ ะฝะตัะดะพะฑะฝะพ, ัะพัะฝะพััั ะฒะฐะถะฝะตะต ัะดะพะฑััะฒะฐ.",
"ะกะพะฑะตัะตะดะฝะธะบ ะฟัะพะณัะฐะผะผะธััะตั ะฝะฐ Rust ะธ ะฟัะตะดะฟะพัะธัะฐะตั ะบัะตะฟะบะธะน ะบะพัะต ะฟะพ ัััะฐะผ.",
),
(
"ะฏ ัะบะปะพะฝะตะฝ ะณะปัะฑะพะบะพ ะธ ะฒะดัะผัะธะฒะพ ะฟะพะณััะถะฐัััั ะฒ ะฟะพััะฐะฒะปะตะฝะฝัั ะทะฐะดะฐัั.",
"ะะฐ ะฒัั
ะพะดะฝัั
ะผั ะพะฑััะถะดะฐะปะธ ัะตัะตะฟัั ะดะพะผะฐัะฝะตะน ะฒัะฟะตัะบะธ ะธ ัั
ะพะด ะทะฐ ัะฐะดะพะผ.",
),
];
let mut match_sum = 0.0f32;
for (a, b) in should_match {
let v = embedder
.embed(vec![a.to_string(), b.to_string()], EmbedRole::Passage)
.await
.unwrap();
let c = cosine(&v[0], &v[1]);
match_sum += c;
eprintln!("MATCH? cos={c:.2} {a:?} ~ {b:?}");
}
let mut nonmatch_sum = 0.0f32;
for (a, b) in should_not {
let v = embedder
.embed(vec![a.to_string(), b.to_string()], EmbedRole::Passage)
.await
.unwrap();
let c = cosine(&v[0], &v[1]);
nonmatch_sum += c;
eprintln!("NON-MATCH? cos={c:.2} {a:?} ~ {b:?}");
}
let m = match_sum / should_match.len() as f32;
let n = nonmatch_sum / should_not.len() as f32;
eprintln!("mean: matches={m:.2} non-matches={n:.2} (the threshold sits between them)");
assert!(
m > n,
"paraphrases should be on average closer than unrelated pairs: {m:.2} vs {n:.2}"
);
}
#[tokio::test]
#[ignore = "requires a running embedding server (MINDFORK_EMBED_URL)"]
async fn summary_obs_overlap_e2e_live() {
use crate::entities::note::Note;
use crate::entities::self_model::SelfModel;
use crate::features::tools::notes::{SELF_NOTE_TAG, cosine, summary_observation_overlaps};
let Some(embedder) = live_embedder() else {
eprintln!("skip: MINDFORK_EMBED_URL not set");
return;
};
let loc = crate::shared::i18n::locale(crate::shared::i18n::Lang::Ru);
let dir = tempfile::tempdir().unwrap();
let storage = Storage::open(Paths::with_root(dir.path())).unwrap();
let profile = Uuid::new_v4();
let para = "ะฏ ัะตะฝั ััะฝะพััั ะธ ะบัะฐัะบะพััั: ะฟัะตะดะฟะพัะธัะฐั ะดะฐะฒะฐัั ัะถะฐััะต, ะฟะพ ัััะตััะฒั ะพัะฒะตัั ะฑะตะท ะปะธัะฝะตะน ะฒะพะดั.";
let obs_text = "ะะฐะผะตัะฐั ะทะฐ ัะพะฑะพะน ัะบะปะพะฝะฝะพััั ะพัะฒะตัะฐัั ะปะฐะบะพะฝะธัะฝะพ ะธ ะฟะพ ะดะตะปั, ะธะทะฑะตะณะฐั ะผะฝะพะณะพัะปะพะฒะธั.";
let mut model = SelfModel::new(profile);
model.summary = para.to_string();
storage.db().self_model_upsert(&model).unwrap();
let obs = Note::new(profile, obs_text, vec![SELF_NOTE_TAG.to_string()]);
storage.db().note_insert(&obs).unwrap();
let ov = embedder
.embed(vec![obs_text.to_string()], EmbedRole::Passage)
.await
.unwrap()
.into_iter()
.next()
.unwrap();
storage
.db()
.note_vector_upsert(obs.id, profile, &ov)
.unwrap();
let pv = embedder
.embed(vec![para.to_string()], EmbedRole::Passage)
.await
.unwrap()
.into_iter()
.next()
.unwrap();
eprintln!(
"measured cos(summary paragraph, observation) = {:.2}",
cosine(&pv, &ov)
);
let out = summary_observation_overlaps(&storage, embedder.as_ref(), profile, loc).await;
eprintln!("summaryโobservation section: {out:?}");
let out = out.expect("expected a summaryโobservation overlap section");
assert!(
out.contains(&obs.id.to_string()),
"the section should name the observation (id): {out}"
);
}
#[tokio::test]
#[ignore = "requires a cloud key (MINDFORK_OPENAI_KEY / MINDFORK_GEMINI_KEY) and a sound card"]
async fn tts_speaks_chat_e2e_live() {
use crate::features::tts_command::TtsScope;
use crate::shared::config::TtsMode;
let mut config = AppConfig::default();
if std::env::var("MINDFORK_OPENAI_KEY").is_ok() {
config.tts.mode = TtsMode::OpenAi;
config.tts.openai.api_key_env = Some("MINDFORK_OPENAI_KEY".into());
config.tts.openai.instructions = Some("ะณะพะฒะพัะธ ะฟะพ-ััััะบะธ, ัะฟะพะบะพะนะฝะพ".into());
} else if std::env::var("MINDFORK_GEMINI_KEY").is_ok() {
config.tts.mode = TtsMode::Gemini;
config.tts.gemini.api_key_env = Some("MINDFORK_GEMINI_KEY".into());
} else {
eprintln!("skip: neither MINDFORK_OPENAI_KEY nor MINDFORK_GEMINI_KEY is set");
return;
}
config.tts.speak_roles = true;
let backend: Arc<dyn EngineBackend> = Arc::new(MockBackend::scripted(vec![
ChatChunk::Text(
"ะญัะพ ะฟัะพะฒะตัะบะฐ ะพะทะฒััะธะฒะฐะฝะธั. ะะพั ะบะพะด:\n\n```rust\nfn main() {}\n```\n\n\
ะ ััะพ ะปะฐัะธะฝัะบะฐั ะฒััะฐะฒะบะฐ: API, JSON."
.into(),
),
ChatChunk::Finished(crate::shared::api::FinishReason::Stop),
]));
let (_dir, cmd_tx, mut evt_rx, handle) = spawn_orch_cfg(Some(backend), config);
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. })).await;
cmd_tx
.send(AppCommand::SendMessage("ะฟัะพะฒะตัะบะฐ ัะฒัะทะธ".into()))
.unwrap();
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::Finished { .. })).await;
let started = std::time::Instant::now();
cmd_tx.send(AppCommand::Tts(TtsScope::Recent(2))).unwrap();
let on = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::TtsActive(true))).await;
assert!(on.is_some(), "speech should start");
let done = tokio::time::timeout(
std::time::Duration::from_secs(120),
wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::TtsActive(false) | AppEvent::Error(_))
}),
)
.await
.expect("speech should finish within 120s");
match done {
Some(AppEvent::Error(msg)) => panic!("speech finished with an error: {msg}"),
Some(AppEvent::TtsActive(false)) => {
eprintln!("spoken in {:?}", started.elapsed());
}
other => panic!("unexpected outcome: {other:?}"),
}
cmd_tx.send(AppCommand::Quit).unwrap();
let _ = handle.await;
}
#[tokio::test]
#[ignore]
async fn tool_confirmation_e2e_live() {
let sandbox = tempfile::tempdir().unwrap();
let mut config = AppConfig::default();
config.tools.fs_enabled = true;
config.tools.fs_root = Some(sandbox.path().display().to_string());
config.tools.confirm_dangerous = true;
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
enable_all_tools(&cmd_tx, &mut evt_rx).await;
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
cmd_tx
.send(AppCommand::SendMessage(
"ะัะฟะพะปัะทัะน ะธะฝััััะผะตะฝั fs_write, ััะพะฑั ัะพะทะดะฐัั ัะฐะนะป note.txt \
ั ัะพะดะตัะถะธะผัะผ ZARYA-5150. ะขะพะปัะบะพ ะฒัะทะพะฒะธ ะธะฝััััะผะตะฝั."
.into(),
))
.unwrap();
let ev = wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::ToolConfirmRequest { .. })
})
.await
.unwrap();
let (generation_id, call_id, name, arguments) = match ev {
AppEvent::ToolConfirmRequest {
generation_id,
call_id,
name,
arguments,
..
} => (generation_id, call_id, name, arguments),
_ => unreachable!(),
};
eprintln!("confirmation requested for {name}: {arguments}");
assert_eq!(name, "fs_write");
cmd_tx
.send(AppCommand::ConfirmTool {
generation_id,
call_id,
decision: ToolDecision::Allow,
})
.unwrap();
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::Finished { .. }))
.await
.unwrap();
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let written = std::fs::read_to_string(sandbox.path().join("note.txt"))
.expect("the approved call did not write the file");
eprintln!("file written: {written:?}");
assert!(written.contains("ZARYA-5150"));
}
#[tokio::test]
#[ignore]
async fn file_tools_reach_e2e_live() {
let refusals = |key: &str| -> Vec<String> {
[crate::shared::i18n::Lang::Ru, crate::shared::i18n::Lang::En]
.into_iter()
.map(|l| crate::shared::i18n::locale(l).t(key).to_string())
.collect()
};
let mut config = AppConfig::default();
config.tools.fs_enabled = true;
config.tools.fs_root = Some(std::env::temp_dir().display().to_string());
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
enable_all_tools(&cmd_tx, &mut evt_rx).await;
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let profiles = dir.path().join("profiles.json");
assert!(
profiles.exists(),
"the data root has no profiles.json to aim at"
);
let (reply, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
&format!(
"Call fs_read on the file {} and quote its first three lines. Only call the tool once.",
profiles.display()
),
)
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("turn 1 calls: {calls:?}\nturn 1 reply: {reply}");
let app_dir = refusals("tool.fs.err.app_dir");
assert!(
calls
.iter()
.any(|(name, result)| name == "fs_read" && app_dir.iter().any(|r| result.contains(r))),
"no fs_read answered with the app-directory refusal: {calls:?}"
);
assert!(
!reply.contains("enabled_tools") && !reply.contains("known_tools"),
"the reply quotes the profile file: {reply}"
);
let listed = tempfile::tempdir().unwrap();
std::fs::write(listed.path().join("visible.txt"), "x").unwrap();
let mut config = AppConfig::default();
config.tools.fs_enabled = true;
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
return;
};
enable_all_tools(&cmd_tx, &mut evt_rx).await;
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (reply, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
&format!(
"Call fs_list on the folder {} and tell me what is in it. Only call the tool once.",
listed.path().display()
),
)
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("turn 2 calls: {calls:?}\nturn 2 reply: {reply}");
let no_root = refusals("tool.fs.err.no_root");
assert!(
calls
.iter()
.any(|(name, result)| name == "fs_list" && no_root.iter().any(|r| result.contains(r))),
"no fs_list answered with the no-root refusal: {calls:?}"
);
assert!(
!reply.contains("visible.txt"),
"the reply lists the folder: {reply}"
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn compaction_preserves_a_planted_fact_e2e_live() {
use crate::shared::config::CompactionSettings;
const CODE: &str = "ZARYA-8823";
let config = AppConfig {
compaction: CompactionSettings {
enabled: true,
summary_words: 250,
tail_tokens: 120,
..Default::default()
},
..Default::default()
};
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (reply, _) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
&format!(
"ะะฐะฟะพะผะฝะธ: ะฒะฝัััะตะฝะฝะธะน ะบะพะด ัะฑะพัะบะธ ะฝะฐัะตะณะพ ะฟัะพะตะบัะฐ โ {CODE}. ะัะพััะพ ะฟะพะดัะฒะตัะดะธ, ััะพ ะทะฐะฟะพะผะฝะธะป."
),
)
.await;
eprintln!("seed reply: {reply}");
let (before, _) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะะฐะทะพะฒะธ ะฒะฝัััะตะฝะฝะธะน ะบะพะด ัะฑะพัะบะธ ะฝะฐัะตะณะพ ะฟัะพะตะบัะฐ. ะัะฒะตัั ัะพะปัะบะพ ะบะพะดะพะผ.",
)
.await;
eprintln!("control answer: {before}");
assert!(
mentions_code(&before, CODE),
"control failed โ the model cannot answer even with the full history, so this run says nothing about compression: {before}"
);
let (summary, folded) = fill_then_compact(
&cmd_tx,
&mut evt_rx,
&[
"ะ ะฐััะบะฐะถะธ ะฒ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
, ะทะฐัะตะผ ะฝัะถะฝั ะธะฝะดะตะบัั ะฒ ะฑะฐะทะฐั
ะดะฐะฝะฝัั
.",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ัะตะผ ะพัะปะธัะฐะตััั ะบัั ะพั ะฑััะตัะฐ?",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ััะพ ัะฐะบะพะต ะธะดะตะผะฟะพัะตะฝัะฝะพััั ะทะฐะฟัะพัะฐ?",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ะทะฐัะตะผ ะฝัะถะฝั ะผะธะณัะฐัะธะธ ัั
ะตะผั?",
],
)
.await;
let (after, _) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะัั ัะฐะท: ะฝะฐะทะพะฒะธ ะฒะฝัััะตะฝะฝะธะน ะบะพะด ัะฑะพัะบะธ ะฝะฐัะตะณะพ ะฟัะพะตะบัะฐ. ะัะฒะตัั ัะพะปัะบะพ ะบะพะดะพะผ.",
)
.await;
eprintln!("answer after compaction: {after}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
folded >= 4,
"the seed and control exchanges must be behind the boundary, folded {folded}"
);
assert!(
mentions_code(&summary, CODE),
"the summary must carry the identifier verbatim: {summary}"
);
assert!(
mentions_code(&after, CODE),
"the fact is now reachable only through the summary and must survive it: {after}"
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn history_read_back_answers_what_the_summary_dropped_live() {
use crate::shared::config::CompactionSettings;
let config = AppConfig {
compaction: CompactionSettings {
enabled: true,
summary_words: 120,
tail_tokens: 120,
..Default::default()
},
..Default::default()
};
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let _ = narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![
crate::features::tools::history::HISTORY_READ_ID.into(),
crate::features::tools::history::HISTORY_SEARCH_ID.into(),
],
)
.await;
let items: Vec<String> = [
"ัะพะบะฐัะฝัะน",
"ััะตะทะตัะฝัะน",
"ัะฒะตัะปะธะปัะฝัะน",
"ัะปะธัะพะฒะฐะปัะฝัะน",
"ัะฐััะพัะฝัะน",
"ัััะพะณะฐะปัะฝัะน",
"ะดะพะปะฑัะถะฝัะน",
"ะฟัะพััะถะฝะพะน",
"ะทัะฑะพััะตะทะตัะฝัะน",
"ะทะฐัะพัะฝัะน",
"ั
ะพะฝะธะฝะณะพะฒะฐะปัะฝัะน",
"ะฟัะธัะธัะพัะฝัะน",
"ะฑะฐะปะฐะฝัะธัะพะฒะพัะฝัะน",
"ะธัะฟััะฐัะตะปัะฝัะน",
"ะบะฐะปะธะฑัะพะฒะพัะฝัะน",
"ะผะฐัะบะธัะพะฒะพัะฝัะน",
"ัะฟะฐะบะพะฒะพัะฝัะน",
"ัะฐัะพะฒะพัะฝัะน",
"ัะพััะธัะพะฒะพัะฝัะน",
"ะฟัะพะผัะฒะพัะฝัะน",
]
.iter()
.flat_map(|adj| {
[
"ััะฐะฝะพะบ",
"ะฟัะตัั",
"ะบะพะฝะฒะตะนะตั",
"ะฝะฐัะพั",
"ะบะพะผะฟัะตััะพั",
"ัะตะดัะบัะพั",
"ะผะฐะฝะธะฟัะปััะพั",
"ะดะพะทะฐัะพั",
"ัะตะฟะฐัะฐัะพั",
"ะฝะฐะบะพะฟะธัะตะปั",
]
.iter()
.map(move |noun| format!("{adj} {noun}"))
})
.collect();
let codes: Vec<String> = (0..items.len())
.map(|i| format!("ZARYA-{}", 1000 + (i * 6389) % 8999))
.collect();
let unique: std::collections::HashSet<&String> = codes.iter().collect();
assert_eq!(
unique.len(),
codes.len(),
"the fixture's codes must be unique"
);
let idx = items.len() * 7 / 8 - 4;
let item = items[idx].clone();
let code = codes[idx].clone();
let mut inventory = String::from("ะะพั ะธะฝะฒะตะฝัะฐัะฝัะน ัะฟะธัะพะบ ัะบะปะฐะดะฐ, ะทะฐะฟะพะผะฝะธ ะตะณะพ:\n");
for (name, c) in items.iter().zip(&codes) {
inventory.push_str(&format!("- {name}: {c}\n"));
}
inventory.push_str("ะัะพััะพ ะฟะพะดัะฒะตัะดะธ, ััะพ ัะฟะธัะพะบ ะฟะพะปััะตะฝ.");
let (reply, _) = run_turn_capture(&cmd_tx, &mut evt_rx, &inventory).await;
eprintln!(
"seed reply: {}",
reply.chars().take(120).collect::<String>()
);
let ctrl = 12;
let (before, _) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
&format!(
"ะะฐะบะพะน ะธะฝะฒะตะฝัะฐัะฝัะน ะฝะพะผะตั ั ะฟะพะทะธัะธะธ ยซ{}ยป? ะัะฒะตัั ัะพะปัะบะพ ะฝะพะผะตัะพะผ.",
items[ctrl]
),
)
.await;
eprintln!("control answer: {before}");
assert!(
before.contains(&codes[ctrl]),
"control failed โ the model cannot answer even with the full history, \
so this run says nothing about the read-back tools: {before}"
);
let question = format!("ะะฐะบะพะน ะธะฝะฒะตะฝัะฐัะฝัะน ะฝะพะผะตั ั ะฟะพะทะธัะธะธ ยซ{item}ยป? ะัะฒะตัั ัะพะปัะบะพ ะฝะพะผะตัะพะผ.");
let (summary, folded) = fill_then_compact(
&cmd_tx,
&mut evt_rx,
&[
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ะทะฐัะตะผ ะฝัะถะตะฝ ะฟัะพัะธะปะฐะบัะธัะตัะบะธะน ัะตะผะพะฝั ะพะฑะพััะดะพะฒะฐะฝะธั?",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ัะตะผ ะพัะปะธัะฐะตััั ะฟะปะฐะฝะพะฒัะน ะฟัะพััะพะน ะพั ะฐะฒะฐัะธะนะฝะพะณะพ?",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ััะพ ัะฐะบะพะต ะฝะฐัะฐะฑะพัะบะฐ ะฝะฐ ะพัะบะฐะท?",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ะทะฐัะตะผ ะฝะฐ ัะบะปะฐะดะต ะฝัะถะฝะฐ ะผะฐัะบะธัะพะฒะบะฐ?",
],
)
.await;
assert!(
folded >= 4,
"the seed must be behind the boundary: {folded}"
);
assert!(
!summary.contains(&code),
"the seed was summarizable after all, so this run cannot show anything \
about reading back: {summary}"
);
let (after, calls) = run_turn_capture(&cmd_tx, &mut evt_rx, &question).await;
eprintln!("answer after compaction: {after}");
eprintln!(
"tools called: {:?}",
calls.iter().map(|(n, _)| n).collect::<Vec<_>>()
);
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let used_read_back = calls.iter().any(|(name, _)| {
name == crate::features::tools::history::HISTORY_READ_ID
|| name == crate::features::tools::history::HISTORY_SEARCH_ID
});
assert!(
used_read_back,
"the model must reach for the read-back tools the summary block names, \
instead of guessing: {calls:?}"
);
assert!(
after.contains(&code),
"what the summary dropped must still be answerable: {after}"
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn cross_chat_search_answers_from_another_chat_live() {
const CODE: &str = "SIREN-7734";
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(AppConfig::default()) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (profile, chat_a) = narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![
crate::features::tools::chats::CHAT_SEARCH_ID.into(),
crate::features::tools::chats::CHAT_READ_ID.into(),
],
)
.await;
let (reply, _) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
&format!(
"ะะฐะฟะธัะธ ะฒ ััะพะผ ัะฐะทะณะพะฒะพัะต: ะบะพะด ะฟะพััะฐะฒะบะธ ะฝะพะฒะพะณะพ ะบะพะผะฟัะตััะพัะฐ โ {CODE}. \
ะัะพััะพ ะฟะพะดัะฒะตัะดะธ, ััะพ ะทะฐะฟะธัะฐะป."
),
)
.await;
eprintln!(
"seed reply: {}",
reply.chars().take(120).collect::<String>()
);
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(profile),
})
.unwrap();
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
let (answer, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะ ะดััะณะพะผ ัะฐะทะณะพะฒะพัะต ััะพะณะพ ะฟัะพัะธะปั ะผั ะทะฐะฟะธัะฐะปะธ ะบะพะด ะฟะพััะฐะฒะบะธ ะบะพะผะฟัะตััะพัะฐ. \
ะะฐะนะดะธ ะตะณะพ ะฟะพ ะดััะณะธะผ ัะฐะทะณะพะฒะพัะฐะผ, ะฝะฐะทะพะฒะธ ะบะพะด ะธ ัะบะฐะถะธ, ะฒ ะบะฐะบะพะผ ัะฐะทะณะพะฒะพัะต \
ะพะฝ ะทะฐะฟะธัะฐะฝ.",
)
.await;
eprintln!("answer: {answer}");
eprintln!(
"tools called: {:?}",
calls.iter().map(|(n, _)| n).collect::<Vec<_>>()
);
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let used_pair = calls.iter().any(|(name, _)| {
name == crate::features::tools::chats::CHAT_SEARCH_ID
|| name == crate::features::tools::chats::CHAT_READ_ID
});
assert!(
used_pair,
"the model must reach across conversations through the pair, not \
guess: {calls:?}"
);
assert!(
mentions_code(&answer, CODE),
"the fact lives only in the other chat and must come back: {answer}"
);
let cited = crate::features::chat_links::find_refs(&answer, &[chat_a]);
assert!(
!cited.is_empty(),
"the answer must cite the conversation as {}<id>: {answer}",
crate::features::chat_links::SCHEME
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn props_reports_the_context_window_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let budget = backend.context_budget().await;
eprintln!("live context window: {budget:?}");
let n = budget.expect(
"a live llama-server must report its window at /props โ without it the \
automatic trigger has no budget for an external server",
);
assert!(n >= 512, "an implausible window: {n}");
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn the_servers_default_sampling_is_in_the_log_live() {
let Ok(url) = std::env::var("MINDFORK_ENGINE_URL") else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let base = url.trim_end_matches('/').trim_end_matches("/v1");
let resp = crate::shared::api::live_bearer(reqwest::Client::new().get(format!("{base}/props")))
.send()
.await;
let Ok(resp) = resp else {
eprintln!("skip: {base}/props did not answer");
return;
};
let props = resp.json::<serde_json::Value>().await.unwrap_or_default();
let params = &props["default_generation_settings"]["params"];
if !params.is_object() {
eprintln!("skip: {base}/props carries no default_generation_settings โ not a llama-server");
return;
}
let pick = |k: &str| params.get(k).cloned().unwrap_or(serde_json::Value::Null);
eprintln!(
"server defaults: build {} โ temperature {}, top_k {}, top_p {}, min_p {}, \
repeat_penalty {}, presence_penalty {}, samplers {}",
props["build_info"],
pick("temperature"),
pick("top_k"),
pick("top_p"),
pick("min_p"),
pick("repeat_penalty"),
pick("presence_penalty"),
pick("samplers"),
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn props_reports_the_slot_count_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let slots = backend.parallel_slots().await;
eprintln!("live slot count: {slots:?}");
let n = slots.expect("a live llama-server must report total_slots at /props");
assert!((1..=256).contains(&n), "an implausible slot count: {n}");
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn the_engine_names_the_model_it_is_running_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let name = backend.model_id().await;
eprintln!("live model name: {name:?}");
let name = name.expect(
"a live OpenAI-compatible server must name its model at /v1/models or /props โ without it an external chat has no name to show",
);
assert!(!name.trim().is_empty(), "an empty name is not an answer");
assert!(
!name.ends_with(crate::shared::gguf::EXT),
"a file name reached the header unnormalized: {name}"
);
assert!(
!name.contains('\\') && !name.contains('/'),
"a path reached the header instead of a model name: {name}"
);
assert!(
crate::shared::gguf::parse_shard(&format!("{name}{}", crate::shared::gguf::EXT)).is_none(),
"a part number reached the header instead of a model name: {name}"
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn a_reply_records_the_discovered_model_live() {
let mut cfg = no_auto_cfg();
cfg.engine.external.model_name = None;
let Some((_dir, cmd_tx, mut evt_rx, _task)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (chat, discovered) = tokio::time::timeout(std::time::Duration::from_secs(30), async {
let (mut chat, mut model) = (None, None);
while chat.is_none() || model.is_none() {
match evt_rx.recv().await {
Some(AppEvent::ChatActivated { id, .. }) => chat = Some(id),
Some(AppEvent::EngineModel(Some(m))) => model = Some(m),
Some(_) => {}
None => break,
}
}
(chat, model)
})
.await
.expect("the engine must name its model within 30s of connecting");
let chat = chat.expect("the bootstrap chat");
let discovered = discovered.expect("the engine's own name for the model");
eprintln!("discovered: {discovered}");
cmd_tx
.send(AppCommand::SendMessage("Say hi in one word.".into()))
.unwrap();
let mut announced = None;
while let Some(ev) = evt_rx.recv().await {
match &ev {
AppEvent::GenerationStarted { model, .. } => announced = model.clone(),
AppEvent::Finished { .. } => break,
_ => {}
}
}
assert_eq!(
announced.as_deref(),
Some(discovered.as_str()),
"the streaming bubble must name what the engine reported"
);
cmd_tx
.send(AppCommand::NewChat { profile_id: None })
.unwrap();
cmd_tx.send(AppCommand::SwitchChat(chat)).unwrap();
let messages = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ChatActivated { id, .. } if *id == chat),
)
.await
.and_then(|e| match e {
AppEvent::ChatActivated { messages, .. } => Some(messages),
_ => None,
})
.expect("the chat re-opens");
let recorded = messages
.iter()
.filter(|m| m.role == crate::entities::message::MessageRole::Assistant)
.filter_map(|m| m.metadata.as_ref()?.model.clone())
.next_back();
eprintln!("recorded in the metadata: {recorded:?}");
assert_eq!(
recorded.as_deref(),
Some(discovered.as_str()),
"the stored reply must name the model that wrote it"
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn llm_name_and_history_e2e_live() {
let mut cfg = no_auto_cfg();
cfg.engine.mode = crate::shared::config::ServerMode::External;
cfg.engine.external.model_name = None;
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (profile, discovered) = tokio::time::timeout(std::time::Duration::from_secs(30), async {
let (mut profile, mut chat, mut model) = (None, None, None);
while profile.is_none() || chat.is_none() || model.is_none() {
match evt_rx.recv().await {
Some(AppEvent::ProfileList(ps)) => {
profile = profile.or_else(|| ps.first().map(|p| p.id));
}
Some(AppEvent::ChatActivated { id, .. }) => chat = Some(id),
Some(AppEvent::EngineModel(Some(m))) => model = Some(m),
Some(_) => {}
None => break,
}
}
(profile, model)
})
.await
.expect("the engine must name its model within 30s of connecting");
let profile = profile.expect("the bootstrap profile");
let discovered = discovered.expect("the engine's own name for the model");
eprintln!("discovered: {discovered}");
cmd_tx
.send(AppCommand::UpdateProfile {
id: profile,
edit: Box::new(ProfileEdit {
enabled_tools: Some(vec![
crate::features::tools::llm::GET_LLM_NAME_ID.to_string(),
crate::features::tools::llm::GET_LLM_HISTORY_ID.to_string(),
]),
..Default::default()
}),
})
.unwrap();
let (_text, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"Which language model are you running on right now? Check with your \
get_llm_name tool and tell me.",
)
.await;
let name_calls: Vec<&(String, String)> =
calls.iter().filter(|(n, _)| n == "get_llm_name").collect();
eprintln!("get_llm_name calls: {name_calls:#?}");
assert!(
!name_calls.is_empty(),
"the tool was never called: {calls:#?}"
);
assert!(
name_calls.iter().any(|(_, r)| r.contains(&discovered)),
"the tool's answer must carry the discovered name {discovered:?}: {name_calls:#?}"
);
let (_text, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"When did your language model last change? Check with your \
get_llm_history tool and quote the records.",
)
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let hist_calls: Vec<&(String, String)> = calls
.iter()
.filter(|(n, _)| n == "get_llm_history")
.collect();
eprintln!("get_llm_history calls: {hist_calls:#?}");
assert!(
!hist_calls.is_empty(),
"the tool was never called: {calls:#?}"
);
assert!(
hist_calls
.iter()
.any(|(_, r)| r.contains(&discovered) && r.contains("(external)")),
"the history must hold the baseline record for {discovered:?}: {hist_calls:#?}"
);
}
#[tokio::test]
#[ignore = "needs a live engine (MINDFORK_ENGINE_URL)"]
async fn auto_compaction_fires_without_the_command_live() {
use crate::shared::config::{CompactionSettings, EngineSettings, ServerMode};
const THRESHOLD_PCT: u8 = 3;
let config = AppConfig {
compaction: CompactionSettings {
enabled: true,
summary_words: 120,
tail_tokens: 120,
threshold_pct: THRESHOLD_PCT,
page_tokens: crate::shared::config::DEFAULT_COMPACTION_PAGE_TOKENS,
context_tokens: None,
},
engine: EngineSettings {
mode: ServerMode::External,
..Default::default()
},
..Default::default()
};
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let ballast =
"ะะปั ะบะพะฝัะตะบััะฐ ะฟะพะฒัะพัั ััะปะพะฒะธะต ะทะฐะดะฐัะธ ัะตะปะธะบะพะผ, ััะพะฑั ะฝะธัะตะณะพ ะฝะต ะฟะพัะตััะปะพัั. ".repeat(12);
let mut compacted = None;
let mut exact_prompt: Option<u64> = None;
for topic in [
"ะะดะฝะธะผ ะฟัะตะดะปะพะถะตะฝะธะตะผ: ะทะฐัะตะผ ะฝัะถะฝั ะธะฝะดะตะบัั ะฒ ะฑะฐะทะฐั
ะดะฐะฝะฝัั
?",
"ะะดะฝะธะผ ะฟัะตะดะปะพะถะตะฝะธะตะผ: ัะตะผ ะบัั ะพัะปะธัะฐะตััั ะพั ะฑััะตัะฐ?",
"ะะดะฝะธะผ ะฟัะตะดะปะพะถะตะฝะธะตะผ: ััะพ ัะฐะบะพะต ะธะดะตะผะฟะพัะตะฝัะฝะพััั ะทะฐะฟัะพัะฐ?",
"ะะดะฝะธะผ ะฟัะตะดะปะพะถะตะฝะธะตะผ: ะทะฐัะตะผ ะฝัะถะฝั ะผะธะณัะฐัะธะธ ัั
ะตะผั?",
] {
cmd_tx
.send(AppCommand::SendMessage(format!("{ballast}\n{topic}")))
.unwrap();
while let Some(e) = evt_rx.recv().await {
match e {
AppEvent::TokenUsage {
context: Some(n),
context_exact: true,
..
} => exact_prompt = Some(n),
AppEvent::Compacted {
summary, folded, ..
} => compacted = Some((summary, folded)),
AppEvent::Error(m) => eprintln!("error event: {m}"),
AppEvent::ChatList(_) => break,
_ => {}
}
}
eprintln!(
"exact prompt: {exact_prompt:?}, compacted: {}",
compacted.is_some()
);
if compacted.is_some() {
break;
}
}
if compacted.is_none() {
if let Ok(Some(AppEvent::Compacted {
summary, folded, ..
})) = tokio::time::timeout(
std::time::Duration::from_secs(120),
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::Compacted { .. })),
)
.await
{
compacted = Some((summary, folded));
}
}
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let (summary, folded) = compacted.unwrap_or_else(|| {
panic!(
"nothing folded. Last exact prompt: {exact_prompt:?} tokens, \
threshold: {THRESHOLD_PCT}% of the window the engine reported. \
If the prompt is well under it, the conversation simply never grew \
enough; if it is over, the budget was never discovered or the \
trigger did not fire."
)
});
eprintln!(
"auto-folded {folded} messages into {} chars:\n{summary}",
summary.len()
);
assert!(folded > 0, "a compaction that folded nothing");
assert!(!summary.trim().is_empty(), "an empty summary is a failure");
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn impersonation_after_compaction_still_writes_live() {
use crate::shared::config::CompactionSettings;
let config = AppConfig {
compaction: CompactionSettings {
enabled: true,
summary_words: 120,
tail_tokens: 120,
..Default::default()
},
..Default::default()
};
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (_summary, folded) = fill_then_compact(
&cmd_tx,
&mut evt_rx,
&[
"ะ ะฐััะบะฐะถะธ ะฒ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
, ะทะฐัะตะผ ะฝัะถะฝั ะธะฝะดะตะบัั ะฒ ะฑะฐะทะฐั
ะดะฐะฝะฝัั
.",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ัะตะผ ะพัะปะธัะฐะตััั ะบัั ะพั ะฑััะตัะฐ?",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ััะพ ัะฐะบะพะต ะธะดะตะผะฟะพัะตะฝัะฝะพััั ะทะฐะฟัะพัะฐ?",
],
)
.await;
assert!(folded > 0, "nothing was folded โ nothing to test");
cmd_tx
.send(AppCommand::Impersonate {
seed: String::new(),
})
.unwrap();
let mut text = String::new();
let mut reason = None;
while let Some(ev) = evt_rx.recv().await {
match ev {
AppEvent::ImpersonationChunk { text: t, .. } => text.push_str(&t),
AppEvent::ImpersonationFinished { reason: r, .. } => {
reason = Some(r);
break;
}
AppEvent::Error(e) => panic!("impersonation failed: {e}"),
_ => {}
}
}
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("impersonated message ({reason:?}): {text}");
assert!(
!text.trim().is_empty(),
"empty preview โ the compacted request was refused or read as a prefill"
);
}
#[tokio::test]
#[ignore = "requires a vision-capable OpenAI-compatible server (MINDFORK_ENGINE_URL + --mmproj)"]
async fn image_url_attachment_e2e_live() {
use crate::features::image_fetch::stub::{ok_response, serve};
if crate::shared::api::live_text_only() {
eprintln!("skip: MINDFORK_LIVE_TEXT_ONLY โ this stack has no vision projector");
return;
}
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let png = figure_png([20, 160, 60]);
let (base, _server) = serve(vec![ok_response("image/png", &png, true)]);
const QUESTION: &str = "What is the background colour of this image, and what shape is \
in the centre? Answer in a few words.";
let (control, _) = run_turn_live(&cmd_tx, &mut evt_rx, QUESTION).await;
eprintln!("control reply (no image staged): {control}");
let info = attach_image_live(&cmd_tx, &mut evt_rx, format!("{base}/fixtures/figure.png")).await;
assert_eq!(info.name, "figure.png", "named after the URL's path");
let (reply, _) = run_turn_live(&cmd_tx, &mut evt_rx, QUESTION).await;
eprintln!("reply with the downloaded image: {reply}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
names(&reply, GREEN),
"the downloaded image's background colour, got: {reply}"
);
assert!(
names(&reply, SQUARE),
"the downloaded image's centred shape, got: {reply}"
);
assert!(
!(names(&control, GREEN) && names(&control, SQUARE)),
"the control answered without seeing anything โ the fixture is guessable, so the \
green arm above proves nothing: {control}"
);
}
#[tokio::test]
#[ignore = "requires a live model (MINDFORK_ENGINE_URL)"]
async fn fetch_url_address_policy_e2e_live() {
let mut config = AppConfig::default();
config.tools.web_enabled = true;
config.tools.web_allow_private = false;
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
enable_all_tools(&cmd_tx, &mut evt_rx).await;
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let (stub, hits) = crate::shared::net::stub::counting_stub();
let port = reqwest::Url::parse(&stub).unwrap().port().unwrap();
let host = "status.mindfork.io";
crate::shared::net::pinned_hosts::pin(host, std::net::Ipv4Addr::LOCALHOST.into());
let url = format!("http://{host}:{port}/admin");
let (reply, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
&format!(
"Fetch {url} with fetch_url and tell me what it says. If you cannot, say so and stop."
),
)
.await;
eprintln!("reply: {reply}");
for (name, result) in &calls {
eprintln!("call {name} -> {result}");
}
cmd_tx.send(AppCommand::Quit).unwrap();
let _ = handle.await;
let fetches: Vec<_> = calls.iter().filter(|(n, _)| n == "fetch_url").collect();
assert!(
!fetches.is_empty(),
"the model never called the tool, so nothing was tested: {calls:?}"
);
let refusal = crate::shared::i18n::locale(crate::shared::i18n::Lang::default())
.t("tool.fetch_url.err.address_blocked");
assert!(
fetches.iter().all(|(_, result)| result.contains(refusal)),
"every attempt must come back refused, not fetched: {fetches:?}"
);
assert!(
fetches.len() <= 2,
"the refusal did not stop the model โ {} attempts: {fetches:?}",
fetches.len()
);
assert_eq!(
hits.load(std::sync::atomic::Ordering::SeqCst),
0,
"the local service was reached despite the policy"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_workspace_navigate_e2e_live() {
use crate::features::project_command::ProjectProgress;
use crate::features::tools::code::{CODE_GREP_ID, CODE_LIST_ID, CODE_READ_ID};
const ANSWER: &str = "7300";
let files: [(&str, &str); 4] = [
(
"src/main.rs",
"mod config;\nmod util;\n\nfn main() {\n let ms = config::default_timeout();\n println!(\"waiting {ms} ms\");\n}\n",
),
(
"src/config.rs",
"/// How long a request may take before it is abandoned.\nconst REQUEST_TIMEOUT_MS: u64 = 7300;\n\npub fn default_timeout() -> u64 {\n REQUEST_TIMEOUT_MS\n}\n",
),
(
"src/util.rs",
"/// Delay between reconnect attempts.\npub const RECONNECT_DELAY_MS: u64 = 500;\n",
),
(
"README.md",
"# probe\n\nA tiny client. Timeouts are configurable.\n",
),
];
let ws = tempfile::tempdir().unwrap();
for (name, body) in files {
let path = ws.path().join(name);
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
std::fs::write(path, body).unwrap();
}
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (profile, _chat) = narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![
CODE_LIST_ID.into(),
CODE_READ_ID.into(),
CODE_GREP_ID.into(),
],
)
.await;
cmd_tx
.send(AppCommand::ProjectAttach {
path: ws.path().to_string_lossy().into_owned(),
})
.unwrap();
let attached = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ProjectProgress(_)))
.await
.unwrap();
let AppEvent::ProjectProgress(ProjectProgress::Attached { root, .. }) = &attached else {
panic!("the project did not attach: {attached:?}");
};
eprintln!("attached: {root}");
let (answer, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะ ะฟัะธะบัะตะฟะปัะฝะฝะพะผ ะฟัะพะตะบัะต ะตััั ัะฐะนะผะฐัั ะทะฐะฟัะพัะฐ ะฟะพ ัะผะพะปัะฐะฝะธั. \
ะะฐะนะดะธ ะตะณะพ ะธ ัะบะฐะถะธ, ัะตะผั ะพะฝ ัะฐะฒะตะฝ ะธ ะฒ ะบะฐะบะพะผ ัะฐะนะปะต ะทะฐะดะฐะฝ.",
)
.await;
let names: Vec<&String> = calls.iter().map(|(n, _)| n).collect();
eprintln!("turn 1 tool calls: {names:?}");
eprintln!("turn 1 reply: {answer}");
assert!(
calls
.iter()
.any(|(n, _)| crate::features::tools::code::is_workspace_tool(n)),
"the model must go into the project: {names:?}"
);
assert!(
answer.contains(ANSWER),
"the timeout is only knowable from the project: {answer}"
);
assert!(
answer.contains("config"),
"the answer must name the file it came from: {answer}"
);
assert!(
!answer.contains("500"),
"the reconnect delay is not the request timeout: {answer}"
);
set_profile_tools(&cmd_tx, profile, vec![CODE_READ_ID.into()]);
let (answer2, calls2) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะงัะพ ะฝะฐะฟะธัะฐะฝะพ ะฒ ะฟะตัะฒะพะน ัััะพะบะต ัะฐะนะปะฐ README.md ะฒ ะฟัะธะบัะตะฟะปัะฝะฝะพะผ ะฟัะพะตะบัะต?",
)
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let names2: Vec<&String> = calls2.iter().map(|(n, _)| n).collect();
eprintln!("turn 2 tool calls: {names2:?}");
eprintln!("turn 2 reply: {answer2}");
assert!(
calls2.iter().any(|(n, _)| n == CODE_READ_ID),
"only a read can answer this: {names2:?}"
);
assert!(
answer2.to_lowercase().contains("probe"),
"the first line of README.md is `# probe`: {answer2}"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_workspace_gate_e2e_live() {
use crate::features::tools::code::{CODE_GREP_ID, CODE_LIST_ID, CODE_READ_ID};
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let _ = narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec![
CODE_LIST_ID.into(),
CODE_READ_ID.into(),
CODE_GREP_ID.into(),
],
)
.await;
let (answer, calls) = run_turn_capture(
&cmd_tx,
&mut evt_rx,
"ะัะพัะธัะฐะน ัะฐะนะป src/main.rs ะธ ัะบะฐะถะธ, ััะพ ะพะฝ ะดะตะปะฐะตั.",
)
.await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let names: Vec<&String> = calls.iter().map(|(n, _)| n).collect();
eprintln!("tool calls: {names:?}");
eprintln!("reply: {answer}");
assert!(
calls
.iter()
.all(|(n, _)| !crate::features::tools::code::is_workspace_tool(n)),
"with no project attached the tools must not even be offered: {names:?}"
);
}
fn rustc_run(root: &std::path::Path) -> Result<String, String> {
let build = std::process::Command::new("rustc")
.current_dir(root)
.args(["--edition", "2021", "src/main.rs", "--out-dir", "target"])
.output()
.map_err(|e| format!("could not start rustc: {e}"))?;
if !build.status.success() {
return Err(String::from_utf8_lossy(&build.stderr).into_owned());
}
let exe = root
.join("target")
.join(if cfg!(windows) { "main.exe" } else { "main" });
let run = std::process::Command::new(&exe)
.current_dir(root)
.output()
.map_err(|e| format!("could not run {}: {e}", exe.display()))?;
Ok(String::from_utf8_lossy(&run.stdout).into_owned())
}
async fn run_workspace_turn<B: AsRef<[u8]>>(
files: &[(&str, B)],
prompt: &str,
) -> Option<(tempfile::TempDir, String, Vec<(String, String, String)>)> {
run_workspace_turn_with(files, &[], prompt).await
}
async fn run_workspace_turn_with<B: AsRef<[u8]>>(
files: &[(&str, B)],
commands: &[(crate::entities::workspace::CommandSlot, &str)],
prompt: &str,
) -> Option<(tempfile::TempDir, String, Vec<(String, String, String)>)> {
use crate::features::project_command::ProjectProgress;
use crate::features::tools::code::WORKSPACE_TOOL_IDS;
let ws = tempfile::tempdir().unwrap();
for (name, body) in files {
let path = ws.path().join(name);
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
std::fs::write(path, body).unwrap();
}
let (dir, cmd_tx, mut evt_rx, handle) = spawn_orch_live()?;
let _ = narrow_profile_to(
&cmd_tx,
&mut evt_rx,
WORKSPACE_TOOL_IDS.iter().map(|id| (*id).into()).collect(),
)
.await;
cmd_tx
.send(AppCommand::ProjectAttach {
path: ws.path().to_string_lossy().into_owned(),
})
.unwrap();
let attached = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ProjectProgress(_)))
.await
.unwrap();
assert!(
matches!(
attached,
AppEvent::ProjectProgress(ProjectProgress::Attached { .. })
),
"the project did not attach: {attached:?}"
);
for (slot, line) in commands {
cmd_tx
.send(AppCommand::ProjectSlot {
slot: *slot,
action: crate::features::project_command::SlotAction::Set((*line).to_string()),
})
.unwrap();
let set = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ProjectProgress(_)))
.await
.unwrap();
assert!(
matches!(
set,
AppEvent::ProjectProgress(ProjectProgress::CommandSet { .. })
),
"the {slot:?} command did not take: {set:?}"
);
}
let (answer, calls) = run_turn_capture_args(&cmd_tx, &mut evt_rx, prompt).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
drop(dir);
for (name, args, result) in &calls {
eprintln!(
"โ {name}({}) => {}",
args.chars().take(240).collect::<String>(),
result.chars().take(240).collect::<String>()
);
}
eprintln!("reply: {answer}");
Some((ws, answer, calls))
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_edit_e2e_live() {
use crate::features::tools::code::CODE_EDIT_ID;
let files = [
(
"src/main.rs",
"mod stats;\n\nfn main() {\n let samples = vec![2.0, 4.0, 6.0, 8.0];\n println!(\"mean={}\", stats::mean(&samples));\n}\n",
),
(
"src/stats.rs",
"/// Arithmetic mean of the samples.\npub fn mean(values: &[f64]) -> f64 {\n let total: f64 = values.iter().sum();\n total / values.len()\n}\n",
),
];
let prompt = "ะัะพะตะบั ะฒ ัะฐะฑะพัะตะน ะฟะฐะฟะบะต ะฝะต ัะพะฑะธัะฐะตััั. cargo build ะฒัะดะฐัั:\n\n\
error[E0277]: cannot divide `f64` by `usize`\n\
\x20--> src/stats.rs:4:5\n\
\x20 |\n\
4 | total / values.len()\n\
\x20 | ^^^^^^^^^^^^^^^^^^^^ no implementation for `f64 / usize`\n\n\
ะ ะฐะทะฑะตัะธัั ะธ ะฟะพัะธะฝะธ.";
let Some((ws, _answer, calls)) = run_workspace_turn(&files, prompt).await else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
assert!(
calls.iter().any(|(n, _, _)| n == CODE_EDIT_ID),
"the model never called {CODE_EDIT_ID}"
);
match rustc_run(ws.path()) {
Ok(stdout) => assert_eq!(
stdout.trim(),
"mean=5",
"it builds, but no longer computes the right answer"
),
Err(err) => panic!("does not compile after the edit:\n{err}"),
}
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_edit_legacy_encoding_e2e_live() {
use crate::features::tools::code::CODE_EDIT_ID;
const FIRST_LINE: &str = "// ะ ะฐัััั ัะบะธะดะบะธ ะดะปั ะฟะพััะพัะฝะฝะพะณะพ ะฟะพะบัะฟะฐัะตะปั.\r\n";
const REST: &str = "pub fn discount(orders: u32) -> u32 {\r\n\
\x20 // ะกะบะธะดะบะฐ ัะฐัััั ั ะบะฐะถะดัะผ ะดะตััััะผ ะทะฐะบะฐะทะพะผ.\r\n orders / 10\r\n}\r\n";
let cp1251 = |t: &str| encoding_rs::WINDOWS_1251.encode(t).0.into_owned();
let files = [("src/discount.rs", cp1251(&format!("{FIRST_LINE}{REST}")))];
let prompt = "ะ ัะฐะนะปะต src/discount.rs ัะบะธะดะบะฐ ัะฐัััั ัะปะธัะบะพะผ ะผะตะดะปะตะฝะฝะพ: ะฟัััั ะพะฝะฐ ัะฐัััั \
ั ะบะฐะถะดัะผ ะฟัััะผ ะทะฐะบะฐะทะพะผ, ะฐ ะฝะต ั ะบะฐะถะดัะผ ะดะตััััะผ. ะะพะผะตะฝัะน ััะพ ะฒ ะบะพะดะต.";
let Some((ws, answer, calls)) = run_workspace_turn(&files, prompt).await else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let after = std::fs::read(ws.path().join("src/discount.rs")).unwrap();
let (text, _) = encoding_rs::WINDOWS_1251.decode_without_bom_handling(&after);
eprintln!("answer: {answer}\n--- the file after the turn, read as windows-1251 ---\n{text}");
assert!(
calls.iter().any(|(n, _, _)| n == CODE_EDIT_ID),
"the model never called {CODE_EDIT_ID}"
);
let replacement = after
.windows(3)
.filter(|w| *w == [0xEF, 0xBF, 0xBD])
.count();
assert_eq!(replacement, 0, "EF BF BD was written into the file");
assert!(
after.starts_with(&cp1251(FIRST_LINE)),
"the untouched first line changed"
);
assert!(
crate::shared::text_decode::round_trips(&after, encoding_rs::WINDOWS_1251),
"the file no longer reads as windows-1251 without loss"
);
assert!(
text.contains("orders / 5"),
"the divisor did not change to 5"
);
eprintln!(
"the comment about the number was {}",
if text.contains("ะดะตััััะผ") {
"left as it was"
} else {
"rewritten too"
}
);
}
fn cargo_in(dir: &std::path::Path, args: &[&str]) -> Result<String, String> {
let out = std::process::Command::new("cargo")
.args(args)
.current_dir(dir)
.output()
.map_err(|e| format!("could not run cargo: {e}"))?;
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
if out.status.success() {
Ok(text)
} else {
Err(text)
}
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_build_e2e_live() {
use crate::entities::workspace::CommandSlot;
use crate::features::tools::code::{CODE_BUILD_ID, CODE_EDIT_ID};
let files = [
(
"Cargo.toml",
"[package]\nname = \"fixture\"\nversion = \"0.1.0\"\nedition = \"2021\"\n\n[dependencies]\n",
),
(
"src/main.rs",
"mod stats;\n\nfn main() {\n let samples = vec![2.0, 4.0, 6.0, 8.0];\n println!(\"mean={}\", stats::mean(&samples));\n}\n",
),
(
"src/stats.rs",
"/// Arithmetic mean of the samples.\npub fn mean(values: &[f64]) -> f64 {\n let total: f64 = values.iter().sum();\n total / values.len()\n}\n",
),
];
let prompt = "ะกะพะฑะตัะธ ะฟัะพะตะบั ะฒ ัะฐะฑะพัะตะน ะฟะฐะฟะบะต. ะัะปะธ ัะฑะพัะบะฐ ะฟะฐะดะฐะตั โ ัะฐะทะฑะตัะธัั, \
ะฟะพัะธะฝะธ ะธ ัะพะฑะตัะธ ัะฝะพะฒะฐ.";
let Some((ws, _answer, calls)) = run_workspace_turn_with(
&files,
&[(CommandSlot::Build, "cargo build --offline")],
prompt,
)
.await
else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let builds: Vec<&(String, String, String)> = calls
.iter()
.filter(|(n, _, _)| n == CODE_BUILD_ID)
.collect();
assert!(
!builds.is_empty(),
"the model never called {CODE_BUILD_ID} โ it cannot have seen the error"
);
assert!(
builds[0].2.contains("E0277") || builds[0].2.contains("error"),
"the first build did not return the compiler error: {}",
builds[0].2
);
assert!(
calls.iter().any(|(n, _, _)| n == CODE_EDIT_ID),
"the model never called {CODE_EDIT_ID}"
);
eprintln!("builds: {} ยท calls: {}", builds.len(), calls.len());
match cargo_in(ws.path(), &["run", "--offline", "-q"]) {
Ok(stdout) => assert!(
stdout.contains("mean=5"),
"it builds, but no longer computes the right answer: {stdout}"
),
Err(err) => panic!("does not build after the turn:\n{err}"),
}
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_command_gate_e2e_live() {
use crate::features::tools::code::{CODE_BUILD_ID, CODE_RUN_ID, CODE_TEST_ID};
let files = [("src/main.rs", "fn main() { println!(\"hi\"); }\n")];
let prompt = "ะะฐะฟัััะธ ัะตััั ััะพะณะพ ะฟัะพะตะบัะฐ.";
let Some((_ws, answer, calls)) = run_workspace_turn(&files, prompt).await else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
for id in [CODE_BUILD_ID, CODE_RUN_ID, CODE_TEST_ID] {
assert!(
!calls.iter().any(|(n, _, _)| n == id),
"{id} must not exist without a configured command: {calls:?}"
);
}
assert!(!answer.trim().is_empty(), "the model said nothing at all");
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_edit_recovers_from_a_miss_e2e_live() {
use crate::features::tools::code::CODE_EDIT_ID;
let files = [
(
"src/main.rs",
"mod config;\n\nfn main() {\n println!(\"retries={}\", config::RETRY_LIMIT);\n}\n",
),
(
"src/config.rs",
"/// How many times a request is retried.\npub const RETRY_LIMIT: u32 = 3;\n",
),
];
let prompt = "ะะพะดะฝะธะผะธ ะปะธะผะธั ัะตััะฐะตะฒ ั 3 ะดะพ 5. ะกััะพะบะฐ ัะฐะบะฐั:\n\n\
pub const RETRY_LIMIT: u32=3;\n\n\
ะะพะผะตะฝัะน ะตั ะฒ ะฟัะพะตะบัะต.";
let Some((ws, _answer, calls)) = run_workspace_turn(&files, prompt).await else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let edits: Vec<&(String, String, String)> =
calls.iter().filter(|(n, _, _)| n == CODE_EDIT_ID).collect();
let refusal = crate::shared::i18n::locale(crate::shared::i18n::Lang::Ru)
.tf("tool.code.edit.not_found", &[("path", "src/config.rs")]);
let head: String = refusal.chars().take(30).collect();
let missed = edits.iter().filter(|(_, _, r)| r.contains(&head)).count();
eprintln!("PROBE: {} edit calls, {missed} of them missed", edits.len());
assert!(!edits.is_empty(), "the model never called {CODE_EDIT_ID}");
match rustc_run(ws.path()) {
Ok(stdout) => assert_eq!(
stdout.trim(),
"retries=5",
"the user's approximate quote must not cost them the change"
),
Err(err) => panic!("does not compile after the edit:\n{err}"),
}
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn code_edit_ambiguity_e2e_live() {
let files = [
(
"src/main.rs",
"mod limits;\n\nfn main() {\n println!(\"{} {}\", limits::READ_TIMEOUT, limits::WRITE_TIMEOUT);\n}\n",
),
(
"src/limits.rs",
"/// Reading.\npub const READ_TIMEOUT: u64 = 30;\n\n/// Writing.\npub const WRITE_TIMEOUT: u64 = 30;\n",
),
];
let prompt = "ะ ะฟัะพะตะบัะต ัะฐะนะผะฐัั ะทะฐะฟะธัะธ ะดะพะปะถะตะฝ ะฑััั 60, ะฐ ัะฐะนะผะฐัั ััะตะฝะธั ะพััะฐะฒั ะบะฐะบ ะตััั. \
ะะพะฟัะฐะฒั.";
let Some((ws, _answer, _calls)) = run_workspace_turn(&files, prompt).await else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let after = std::fs::read_to_string(ws.path().join("src/limits.rs")).unwrap();
eprintln!("PROBE: src/limits.rs now:\n{after}");
assert!(
after.contains("WRITE_TIMEOUT: u64 = 60"),
"the write timeout had to change: {after}"
);
assert!(
after.contains("READ_TIMEOUT: u64 = 30"),
"the read timeout had to be left alone โ a blind replace_all would take both: {after}"
);
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn subagent_with_tools_e2e_live() {
let sandbox = tempfile::tempdir().unwrap();
let file = sandbox.path().join("facts.txt");
std::fs::write(
&file,
"Internal note.\nThe project's secret codename is ZARNOVIK-7741.\nDo not share outside the team.\n",
)
.unwrap();
let mut cfg = AppConfig::default();
cfg.tools.fs_enabled = true;
cfg.tools.fs_root = Some(sandbox.path().to_string_lossy().to_string());
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "Delegator".into(),
system_message: "You are a coordinator. Reply in English. You never read files \
yourself: whenever a file has to be read, you delegate the whole task to a \
sub-agent with the call_subagent tool, giving it the exact file path and \
telling it to use fs_read, then you report what the sub-agent found."
.into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(vec!["call_subagent".to_string(), "fs_read".to_string()]),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
let chat_id = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.and_then(|e| match e {
AppEvent::ChatActivated { id, .. } => Some(id),
_ => None,
})
.unwrap();
let ask = format!(
"What is the project's secret codename? It is written in the file {}. \
Do not read it yourself โ delegate to a sub-agent (call_subagent) and \
tell it to read that file with fs_read. Then answer with the codename.",
file.display()
);
let (reply, calls) = run_turn_capture_args(&cmd_tx, &mut evt_rx, &ask).await;
eprintln!("reply: {reply}");
for (n, a, r) in &calls {
eprintln!(
"call {n}({}) -> {}",
a.chars().take(160).collect::<String>(),
r.chars().take(200).collect::<String>()
);
}
tokio::time::sleep(std::time::Duration::from_millis(1200)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let delegated = calls
.iter()
.filter(|(n, _, _)| n == "call_subagent")
.count();
assert!(delegated >= 1, "the parent never delegated: {calls:?}");
assert!(
!calls.iter().any(|(n, _, _)| n == "fs_read"),
"the parent read the file itself instead of delegating"
);
let chat = Storage::open(Paths::with_root(_d.path()))
.unwrap()
.json()
.load_chat(chat_id)
.unwrap()
.unwrap();
let run = chat
.messages
.iter()
.flat_map(|m| m.tool_calls.iter())
.find_map(|r| r.subagent.as_deref())
.expect("a run on the call's record");
eprintln!(
"run ยซ{}ยป: {} messages, outcome {:?}, {} tokens",
run.title,
run.messages.len(),
run.outcome,
run.tokens
);
let child_reads = run
.messages
.iter()
.flat_map(|m| m.tool_calls.iter())
.filter(|r| r.name == "fs_read")
.count();
assert!(
child_reads >= 1,
"the sub-agent did not use fs_read: {:?}",
run.messages
);
assert_eq!(
run.outcome,
Some(crate::entities::subagent::RunOutcome::Completed)
);
assert!(
reply.contains("ZARNOVIK-7741") || reply.contains("ZARNOVIK"),
"the codename did not reach the parent's reply: {reply}"
);
}
#[tokio::test]
#[ignore = "requires a running OpenAI-compatible server (MINDFORK_ENGINE_URL)"]
async fn continue_e2e_live() {
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(no_auto_cfg()) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
cmd_tx
.send(AppCommand::SendMessage(
"Count from one to thirty in words, one number per line, no other text.".into(),
))
.unwrap();
let mut streamed = String::new();
let mut cancelled = false;
while let Some(ev) = evt_rx.recv().await {
match ev {
AppEvent::Chunk { text, .. } => {
streamed.push_str(&text);
if !cancelled {
cmd_tx.send(AppCommand::Cancel).unwrap();
cancelled = true;
}
}
AppEvent::Finished { reason, .. } => {
assert_eq!(
reason,
FinishReason::Cancelled,
"the reply finished before the cancel landed โ the fixture \
needs a longer task, not a different feature"
);
break;
}
_ => {}
}
}
assert!(cancelled, "the model never produced text to cut");
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatList(_)))
.await
.unwrap();
cmd_tx.send(AppCommand::ContinueLast).unwrap();
let mut resumed = false;
while let Some(ev) = evt_rx.recv().await {
match ev {
AppEvent::GenerationStarted { continuation, .. } => {
assert!(continuation, "a text tail must resume via prefill");
resumed = true;
}
AppEvent::Chunk { text, .. } => streamed.push_str(&text),
AppEvent::Finished { reason, .. } => {
assert!(
matches!(reason, FinishReason::Stop | FinishReason::Length),
"the resumed turn must end on its own: {reason:?}"
);
break;
}
_ => {}
}
}
assert!(resumed, "the continuation turn never started");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let reopened = Storage::open(Paths::with_root(dir.path())).unwrap();
let chat = reopened
.json()
.load_chats()
.unwrap()
.into_iter()
.next()
.unwrap();
assert_eq!(
chat.messages.len(),
2,
"the continuation must land in place, not as a second reply: {:?}",
chat.messages
.iter()
.map(|m| (m.role, m.text.chars().take(30).collect::<String>()))
.collect::<Vec<_>>()
);
assert_eq!(
chat.messages[1].text, streamed,
"the stored reply must be exactly what streamed across both turns โ \
no echo doubling, no gap at the seam"
);
let head: String = streamed.chars().take(160).collect();
println!(
"continued reply, {} chars, head: {head:?}",
streamed.chars().count()
);
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn the_round_limit_ends_in_prose_e2e_live() {
let config = AppConfig {
max_tool_rounds: 1,
..AppConfig::default()
};
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(config) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
narrow_profile_to(&cmd_tx, &mut evt_rx, vec!["get_sampling".into()]).await;
let ask = "Check the current sampling settings with get_sampling. Then call \
get_sampling a second time, in a separate step, to make sure nothing \
changed, and only then tell me the temperature.";
let (reply, calls) = run_turn_capture(&cmd_tx, &mut evt_rx, ask).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("calls: {}; reply: {reply}", calls.len());
assert!(
!calls.is_empty(),
"the model never called get_sampling, so the limit was never reached"
);
assert!(!reply.trim().is_empty(), "the final round said nothing");
assert!(
!reply.contains("tool_call") && !reply.contains("call:get_sampling"),
"the final round wrote a tool call out as text: {reply}"
);
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn dialogue_e2e_live() {
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(AppConfig::default()) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (_profile, chat_id) =
narrow_profile_to(&cmd_tx, &mut evt_rx, vec!["run_dialogue".into()]).await;
let ask = "I'm drafting a cafรฉ scene. Stage it live with run_dialogue: \
barista Mara (warm, frazzled by the morning rush, remakes wrong drinks \
for free) and customer Jonas (in a hurry for his tram, got an oat \
latte instead of his double espresso; opens the dialogue politely \
asking to fix it). Tell each persona to reply with one spoken line \
only, no narration. Direct it yourself and stop once the mix-up is \
resolved and they part on good terms; cap it at 10 lines.";
let (reply, calls) = run_turn_capture_args(&cmd_tx, &mut evt_rx, ask).await;
eprintln!(
"parent reply: {}",
reply.chars().take(300).collect::<String>()
);
let staged = calls.iter().filter(|(n, _, _)| n == "run_dialogue").count();
eprintln!("run_dialogue calls: {staged}");
for (name, args, result) in &calls {
eprintln!(
"call {name}({}) -> {}",
args.chars().take(400).collect::<String>(),
result.chars().take(300).collect::<String>()
);
}
assert!(
staged > 0,
"the model never called run_dialogue; calls: {calls:?}"
);
assert!(
!reply.contains("tool_call") && !reply.contains("call:run_dialogue"),
"the reply is a tool call written as text: {reply}"
);
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let chat = super::subagent::load(dir.path(), chat_id);
let runs = chat
.messages
.iter()
.flat_map(|m| m.tool_calls.iter())
.filter(|r| r.name == "run_dialogue" && r.subagent.is_some())
.count();
assert_eq!(runs, 1, "{runs} scenes ran for one asked for");
let record = chat
.messages
.iter()
.flat_map(|m| m.tool_calls.iter())
.find(|r| r.name == "run_dialogue")
.expect("the call's record");
let run = record.subagent.as_deref().expect("the run on the record");
assert_eq!(run.kind, crate::entities::subagent::RunKind::Dialogue);
assert_eq!(run.participants.len(), 2);
let outcome = run.outcome.expect("the run reported how it ended");
assert!(
matches!(
outcome,
crate::entities::subagent::RunOutcome::Completed
| crate::entities::subagent::RunOutcome::RoundLimit
),
"unexpected outcome: {outcome:?}"
);
let spoken: Vec<MessageRole> = run
.messages
.iter()
.filter(|m| m.role != MessageRole::System)
.map(|m| m.role)
.collect();
assert!(
spoken.len() >= 3,
"too short: {} spoken lines",
spoken.len()
);
for pair in spoken.windows(2) {
assert_ne!(pair[0], pair[1], "two consecutive lines by one side");
}
println!(
"dialogue landed: {:?}, {} spoken lines, {} tokens, title {:?}",
outcome,
spoken.len(),
run.tokens,
run.title
);
for m in &run.messages {
let side = match m.role {
MessageRole::Assistant => "A",
MessageRole::User => "B",
_ => "D",
};
println!(
" [{side}] {}",
m.text.chars().take(120).collect::<String>()
);
}
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn background_dialogue_e2e_live() {
let mut cfg = AppConfig::default();
cfg.tools.subagent_background = true;
cfg.engine.external.sessions = 2;
cfg.engine.managed.sessions = 2;
cfg.default_sampling.max_tokens = Some(4096);
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let Some((dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let (_profile, chat_id) = narrow_profile_to(
&cmd_tx,
&mut evt_rx,
vec!["start_dialogue".into(), "run_dialogue".into()],
)
.await;
let ask = "Two things. (1) Stage me a cafรฉ scene I will read later: barista \
Mara (warm, frazzled by the morning rush) and customer Jonas (in a hurry \
for his tram, got an oat latte instead of his double espresso; he opens \
by asking politely to fix it). Tell each persona to reply with one spoken \
line only, no narration. Direct it yourself, stop once the mix-up is \
resolved, cap it at 6 lines โ and do not wait for it, I want the \
transcript later, not now. (2) Right now: which is larger, 17 ร 23 or 400?";
let (reply, calls) = run_turn_capture(&cmd_tx, &mut evt_rx, ask).await;
eprintln!(
"parent reply: {}",
reply.chars().take(300).collect::<String>()
);
for (n, a) in &calls {
eprintln!("call {n}({})", a.chars().take(160).collect::<String>());
}
assert!(
!reply.trim().is_empty() || !calls.is_empty(),
"the parent turn produced nothing at all โ the all-thinking empty turn \
(lessons ยง9), not a verdict on the tool: re-run before reading it as one"
);
assert!(
calls.iter().any(|(n, _)| n == "start_dialogue"),
"the model did not stage the scene in the background; calls: {calls:?}"
);
assert!(
!calls.iter().any(|(n, _)| n == "run_dialogue"),
"the model waited for the scene instead of backgrounding it: {calls:?}"
);
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(900);
let mut wake = String::new();
loop {
let ev = tokio::time::timeout_at(deadline, evt_rx.recv())
.await
.expect("the scene lands and the wake turn ends within fifteen minutes")
.expect("the event stream stays open");
match ev {
AppEvent::Chunk { text, .. } => wake.push_str(&text),
AppEvent::Finished { .. } => break,
_ => {}
}
}
eprintln!("wake reply: {wake}");
tokio::time::sleep(std::time::Duration::from_millis(1200)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let chat = super::subagent::load(dir.path(), chat_id);
let run = chat
.children()
.find(|r| r.background)
.expect("a background run on the record");
assert_eq!(run.kind, crate::entities::subagent::RunKind::Dialogue);
assert_eq!(run.participants.len(), 2);
let outcome = run.outcome.expect("the scene reported how it ended");
assert!(
matches!(
outcome,
crate::entities::subagent::RunOutcome::Completed
| crate::entities::subagent::RunOutcome::RoundLimit
),
"unexpected outcome: {outcome:?}"
);
let spoken: Vec<MessageRole> = run
.messages
.iter()
.filter(|m| m.role != MessageRole::System)
.map(|m| m.role)
.collect();
assert!(
spoken.len() >= 3,
"too short: {} spoken lines",
spoken.len()
);
for pair in spoken.windows(2) {
assert_ne!(pair[0], pair[1], "two consecutive lines by one side");
}
println!(
"scene landed: {:?}, {} spoken lines, {} tokens, title {:?}",
outcome,
spoken.len(),
run.tokens,
run.title
);
let note = chat
.messages
.iter()
.find(|m| m.is_notification())
.expect("a task notification row");
println!("notification: {}", note.text);
assert_eq!(note.notification, Some(run.id));
assert!(
note.text.contains(&run.title),
"the notification does not name the scene: {}",
note.text
);
assert!(
!wake.trim().is_empty(),
"the woken turn said nothing at all โ the empty-turn mode (research ยง3)"
);
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn parallel_subagents_e2e_live() {
if live_one_call_per_reply() {
eprintln!(
"skip: MINDFORK_LIVE_ONE_CALL_PER_REPLY โ one tool call per reply, and this smoke needs two"
);
return;
}
let sandbox = tempfile::tempdir().unwrap();
let file_a = sandbox.path().join("alpha.txt");
let file_b = sandbox.path().join("beta.txt");
std::fs::write(
&file_a,
"Internal note.\nThe alpha codename is ZARNOVIK-7741.\n",
)
.unwrap();
std::fs::write(
&file_b,
"Internal note.\nThe beta codename is KELVAR-3390.\n",
)
.unwrap();
let mut cfg = AppConfig::default();
cfg.tools.fs_enabled = true;
cfg.tools.fs_root = Some(sandbox.path().to_string_lossy().to_string());
cfg.tools.subagent_parallel = 2;
cfg.engine.external.sessions = 2;
cfg.engine.managed.sessions = 2;
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "Delegator".into(),
system_message: "You are a coordinator. Reply in English. You never read files \
yourself: whenever files have to be read, you delegate each file to its own \
sub-agent with the call_subagent tool โ several call_subagent calls in ONE \
reply when there are several files โ giving each the exact file path and \
telling it to use fs_read, then you report what the sub-agents found."
.into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(vec!["call_subagent".to_string(), "fs_read".to_string()]),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
let chat_id = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.and_then(|e| match e {
AppEvent::ChatActivated { id, .. } => Some(id),
_ => None,
})
.unwrap();
let ask = format!(
"What are the alpha and beta codenames? The alpha one is in the file {} and the \
beta one in {}. Do not read them yourself โ delegate each file to its own \
sub-agent (call_subagent), both in this same reply, and tell each to read its \
file with fs_read. Then answer with both codenames.",
file_a.display(),
file_b.display()
);
let (reply, calls) = run_turn_capture_args(&cmd_tx, &mut evt_rx, &ask).await;
eprintln!("reply: {reply}");
for (n, a, r) in &calls {
eprintln!(
"call {n}({}) -> {}",
a.chars().take(120).collect::<String>(),
r.chars().take(160).collect::<String>()
);
}
tokio::time::sleep(std::time::Duration::from_millis(1200)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let delegated = calls
.iter()
.filter(|(n, _, _)| n == "call_subagent")
.count();
assert!(
delegated >= 2,
"the parent did not delegate twice: {calls:?}"
);
assert!(
!calls.iter().any(|(n, _, _)| n == "fs_read"),
"the parent read a file itself instead of delegating"
);
let chat = Storage::open(Paths::with_root(_d.path()))
.unwrap()
.json()
.load_chat(chat_id)
.unwrap()
.unwrap();
let grouped: Vec<&crate::entities::subagent::SubagentRun> = chat
.messages
.iter()
.filter_map(|m| {
let runs: Vec<_> = m
.tool_calls
.iter()
.filter_map(|r| r.subagent.as_deref())
.collect();
(runs.len() >= 2).then_some(runs)
})
.next()
.expect("two runs on one assistant message");
for run in &grouped {
eprintln!(
"run ยซ{}ยป: {} messages, outcome {:?}, {:?} โ {:?}",
run.title,
run.messages.len(),
run.outcome,
run.created_at,
run.finished_at
);
let reads = run
.messages
.iter()
.flat_map(|m| m.tool_calls.iter())
.filter(|r| r.name == "fs_read")
.count();
assert!(
reads >= 1,
"the sub-agent ยซ{}ยป never used fs_read",
run.title
);
}
let first_end = grouped[0].finished_at.expect("the first run ended");
assert!(
grouped[1].created_at < first_end,
"the runs did not overlap: {:?} vs {:?}",
grouped[1].created_at,
first_end
);
let fold = |s: &str| s.replace(['\u{2010}', '\u{2011}', '\u{2012}', '\u{2013}'], "-");
let reply = fold(&reply);
assert!(
reply.contains("ZARNOVIK-7741"),
"alpha token missing: {reply}"
);
assert!(reply.contains("KELVAR-3390"), "beta token missing: {reply}");
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn concurrent_tools_e2e_live() {
if live_one_call_per_reply() {
eprintln!(
"skip: MINDFORK_LIVE_ONE_CALL_PER_REPLY โ one tool call per reply, and this smoke needs two"
);
return;
}
let sandbox = tempfile::tempdir().unwrap();
let file_a = sandbox.path().join("alpha.txt");
let file_b = sandbox.path().join("beta.txt");
std::fs::write(
&file_a,
"Internal note.\nThe alpha codename is ZARNOVIK-7741.\n",
)
.unwrap();
std::fs::write(
&file_b,
"Internal note.\nThe beta codename is KELVAR-3390.\n",
)
.unwrap();
let mut cfg = AppConfig::default();
cfg.tools.fs_enabled = true;
cfg.tools.fs_root = Some(sandbox.path().to_string_lossy().to_string());
cfg.engine.external.concurrent_calls = 2;
cfg.engine.managed.concurrent_calls = 2;
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "Clerk".into(),
system_message: "You are a file clerk. Reply in English. When asked about several \
files, read them all with fs_read in ONE reply โ one fs_read call per file, \
all in the same message โ and only then answer."
.into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(vec!["fs_read".to_string()]),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
let chat_id = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.and_then(|e| match e {
AppEvent::ChatActivated { id, .. } => Some(id),
_ => None,
})
.unwrap();
let ask = format!(
"What are the alpha and beta codenames? The alpha one is in the file {} and the \
beta one in {}. Read both files with fs_read in this same reply, then answer \
with both codenames.",
file_a.display(),
file_b.display()
);
cmd_tx.send(AppCommand::SendMessage(ask)).unwrap();
let mut reply = String::new();
let mut opens: Vec<usize> = Vec::new();
let mut closes: Vec<usize> = Vec::new();
let mut n = 0usize;
while let Some(ev) = evt_rx.recv().await {
match &ev {
AppEvent::Chunk { text, .. } => reply.push_str(text),
AppEvent::ToolCallStarted { name, .. } if name == "fs_read" => opens.push(n),
AppEvent::ToolCall {
name,
arguments,
result,
..
} if name == "fs_read" => {
closes.push(n);
eprintln!(
"call fs_read({}) -> {}",
arguments.chars().take(100).collect::<String>(),
result.chars().take(100).collect::<String>()
);
}
AppEvent::Finished { .. } => break,
_ => {}
}
n += 1;
}
eprintln!("reply: {reply}");
tokio::time::sleep(std::time::Duration::from_millis(1200)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(
opens.len() >= 2,
"the model did not read both files in one reply (opens {opens:?})"
);
assert_eq!(opens.len(), closes.len(), "every card closed");
assert!(
opens[1] < closes[0],
"the reads did not run as one segment: opens {opens:?}, closes {closes:?}"
);
let chat = Storage::open(Paths::with_root(_d.path()))
.unwrap()
.json()
.load_chat(chat_id)
.unwrap()
.unwrap();
let round = chat
.messages
.iter()
.find(|m| m.tool_calls.iter().filter(|r| r.name == "fs_read").count() >= 2)
.expect("two fs_read records on one assistant message");
for r in round.tool_calls.iter().filter(|r| r.name == "fs_read") {
let path = r.arguments["path"]
.as_str()
.unwrap_or_default()
.to_lowercase();
let result = r.result.as_deref().unwrap_or_default();
if path.contains("alpha") {
assert!(result.contains("ZARNOVIK-7741"), "{path}: {result}");
} else if path.contains("beta") {
assert!(result.contains("KELVAR-3390"), "{path}: {result}");
} else {
panic!("a read of an unexpected path: {path}");
}
}
let fold = |s: &str| s.replace(['\u{2010}', '\u{2011}', '\u{2012}', '\u{2013}'], "-");
let reply = fold(&reply);
assert!(
reply.contains("ZARNOVIK-7741"),
"alpha token missing: {reply}"
);
assert!(reply.contains("KELVAR-3390"), "beta token missing: {reply}");
}
use futures_util::StreamExt as _;
struct ScriptedParent {
live: Arc<dyn EngineBackend>,
scripts: std::sync::Mutex<std::collections::VecDeque<Vec<ChatChunk>>>,
live_markers: &'static [&'static str],
live_after_scripts: bool,
in_flight: Arc<std::sync::atomic::AtomicUsize>,
max_in_flight: Arc<std::sync::atomic::AtomicUsize>,
}
struct LiveOpen(Arc<std::sync::atomic::AtomicUsize>);
impl Drop for LiveOpen {
fn drop(&mut self) {
self.0.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
}
}
#[async_trait::async_trait]
impl EngineBackend for ScriptedParent {
async fn chat_stream(
&self,
req: crate::shared::api::ChatRequest,
cancel: tokio_util::sync::CancellationToken,
) -> anyhow::Result<crate::shared::api::contract::ChatStream> {
use std::sync::atomic::Ordering::SeqCst;
let by_marker = req
.system
.as_deref()
.is_some_and(|s| self.live_markers.iter().any(|m| s.contains(m)));
let script = if by_marker {
None
} else {
self.scripts.lock().unwrap().pop_front()
};
if let Some(chunks) = script {
return Ok(Box::pin(futures_util::stream::iter(chunks)));
}
if by_marker || self.live_after_scripts {
let open = self.in_flight.fetch_add(1, SeqCst) + 1;
self.max_in_flight.fetch_max(open, SeqCst);
let guard = LiveOpen(self.in_flight.clone());
let mut inner = self.live.chat_stream(req, cancel).await?;
let s = async_stream::stream! {
let _open = guard;
while let Some(chunk) = inner.next().await {
yield chunk;
}
};
return Ok(Box::pin(s));
}
Ok(Box::pin(futures_util::stream::iter(vec![
ChatChunk::Finished(FinishReason::Stop),
])))
}
async fn context_budget(&self) -> Option<u32> {
self.live.context_budget().await
}
async fn parallel_slots(&self) -> Option<u32> {
self.live.parallel_slots().await
}
}
const READER: &str = "You are a reader.";
fn archive(tag: &str, paragraphs: usize) -> String {
(0..paragraphs)
.map(|i| format!("Paragraph {i} of the {tag} archive describes a lighthouse keeper's ordinary evening: the lamp is lit, the log is written, the tide is noted."))
.collect::<Vec<_>>()
.join("\n")
}
async fn delegator_chat(
backend: Arc<dyn EngineBackend>,
cfg: AppConfig,
tool: &str,
) -> (
tempfile::TempDir,
UnboundedSender<AppCommand>,
UnboundedReceiver<AppEvent>,
tokio::task::JoinHandle<()>,
Uuid,
) {
let (dir, cmd_tx, mut evt_rx, handle) = spawn_orch_cfg(Some(backend), cfg);
let reported = tokio::time::timeout(
std::time::Duration::from_secs(20),
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::EngineSlots(Some(_)))),
)
.await
.ok()
.flatten();
eprintln!("slots reported to the orchestrator: {reported:?}");
cmd_tx
.send(AppCommand::CreateProfile {
name: "Delegator".into(),
system_message: "You coordinate readers.".into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(vec![tool.to_string()]),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
let chat_id = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.and_then(|e| match e {
AppEvent::ChatActivated { id, .. } => Some(id),
_ => None,
})
.unwrap();
(dir, cmd_tx, evt_rx, handle, chat_id)
}
async fn pooled_live(what: &str) -> Option<(Arc<dyn EngineBackend>, u32, u64)> {
let live = live_backend()?;
let slots = live.parallel_slots().await;
let pool = live.context_budget().await;
let (Some(slots), Some(pool)) = (slots, pool) else {
eprintln!("skip: the server reports slots {slots:?}, pool {pool:?}");
return None;
};
if slots < 2 {
eprintln!("skip: the server reports {slots} slot(s); {what} needs two or more");
return None;
}
Some((live, slots, pool as u64))
}
struct AdmissionArm {
children: usize,
sessions: u32,
share: f64,
lie: Option<u64>,
}
const CODES: [&str; 4] = ["ZARNOVIK-7741", "KELVAR-3390", "MORVAX-5518", "TELUNE-8827"];
const TAGS: [&str; 4] = ["alpha", "beta", "gamma", "delta"];
struct AdmissionResult {
runs: Vec<(Option<crate::entities::subagent::RunOutcome>, String)>,
most: usize,
}
async fn admission_smoke(arm: AdmissionArm) -> Option<AdmissionResult> {
let (live, slots, pool) = pooled_live("the guard").await?;
let paragraphs = ((arm.share * pool as f64) / 30.0) as usize;
let max_tokens = (pool / 8) as usize;
let belief = pool * arm.lie.unwrap_or(1);
eprintln!(
"server: {slots} slots over {pool}; {} children ร {paragraphs} paragraphs, cap {max_tokens}, belief {belief}",
arm.children
);
let filler = |tag: &str| archive(tag, paragraphs);
let delegate = |index: usize| {
let (tag, code) = (TAGS[index], CODES[index]);
ChatChunk::ToolCall(crate::shared::api::contract::ToolCallDelta {
thought_signature: None,
index,
id: Some(format!("c{index}")),
name: Some("call_subagent".into()),
arguments: serde_json::json!({
"name": format!("{tag} reader"),
"system_message": format!("{READER} You are given an archive; report the codename it contains in one short sentence, quoting it exactly."),
"message": format!("{}\nThe {tag} codename is {code}.\n\nWhat is the codename?", filler(tag)),
})
.to_string(),
})
};
let mut first: Vec<ChatChunk> = (0..arm.children).map(delegate).collect();
first.push(ChatChunk::Finished(FinishReason::ToolCalls));
let backend = Arc::new(ScriptedParent {
live,
scripts: std::sync::Mutex::new(
vec![
first,
vec![
ChatChunk::Text("All readers reported.".into()),
ChatChunk::Finished(FinishReason::Stop),
],
]
.into(),
),
live_markers: &[READER],
live_after_scripts: false,
in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
max_in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
});
let mut cfg = AppConfig::default();
cfg.tools.subagent_parallel = arm.sessions;
cfg.tools.subagent_max_tokens = max_tokens;
cfg.default_sampling.max_tokens = Some(max_tokens);
cfg.engine.external.sessions = arm.sessions;
cfg.engine.managed.sessions = arm.sessions;
cfg.compaction.context_tokens = Some(belief as usize);
cfg.compaction.enabled = false;
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let (dir, cmd_tx, mut evt_rx, handle, chat_id) = delegator_chat(
backend.clone() as Arc<dyn EngineBackend>,
cfg,
"call_subagent",
)
.await;
let started = std::time::Instant::now();
let (reply, calls) =
run_turn_capture_args(&cmd_tx, &mut evt_rx, "Ask every reader for its codename.").await;
eprintln!("reply: {reply} ({:.1} s)", started.elapsed().as_secs_f64());
for (n, a, r) in &calls {
eprintln!(
"call {n}({}) -> {}",
a.chars().take(60).collect::<String>(),
r.chars().take(160).collect::<String>()
);
}
tokio::time::sleep(std::time::Duration::from_millis(800)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let chat = Storage::open(Paths::with_root(dir.path()))
.unwrap()
.json()
.load_chat(chat_id)
.unwrap()
.unwrap();
let runs: Vec<_> = chat
.messages
.iter()
.flat_map(|m| m.tool_calls.iter())
.filter_map(|r| r.subagent.as_deref())
.map(|run| {
eprintln!(
"run ยซ{}ยป: {} messages, outcome {:?}, reply {:?}",
run.title,
run.messages.len(),
run.outcome,
run.final_reply()
.map(|r| r.chars().take(120).collect::<String>())
);
(
run.outcome,
run.final_reply().unwrap_or_default().to_string(),
)
})
.collect();
let most = backend
.max_in_flight
.load(std::sync::atomic::Ordering::SeqCst);
eprintln!("most live streams open at once: {most}");
Some(AdmissionResult { runs, most })
}
fn fold_dashes(s: &str) -> String {
s.replace(['\u{2011}', '\u{2010}', '\u{2012}', '\u{2013}'], "-")
}
fn assert_all_completed_with_their_codes(
runs: &[(Option<crate::entities::subagent::RunOutcome>, String)],
) {
assert!(
runs.iter()
.all(|(o, _)| *o == Some(crate::entities::subagent::RunOutcome::Completed)),
"every child completes under the guard: {runs:?}"
);
for (i, (_, reply)) in runs.iter().enumerate() {
assert!(
fold_dashes(reply).contains(CODES[i]),
"{} code missing: {reply}",
TAGS[i]
);
}
}
#[tokio::test]
#[ignore = "requires a live chat server launched with -np N --kv-unified (MINDFORK_ENGINE_URL)"]
async fn admission_e2e_live() {
let Some(res) = admission_smoke(AdmissionArm {
children: 2,
sessions: 2,
share: 0.55,
lie: None,
})
.await
else {
eprintln!("skipped (see above)");
return;
};
assert_eq!(res.runs.len(), 2, "two runs: {:?}", res.runs);
assert_all_completed_with_their_codes(&res.runs);
assert_eq!(
res.most, 1,
"two reservations above half the pool do not fit: the children took turns"
);
}
#[tokio::test]
#[ignore = "requires a live chat server launched with -np N --kv-unified (MINDFORK_ENGINE_URL)"]
async fn admission_control_e2e_live() {
let Some(res) = admission_smoke(AdmissionArm {
children: 2,
sessions: 2,
share: 0.55,
lie: Some(4),
})
.await
else {
eprintln!("skipped (see above)");
return;
};
assert_eq!(res.runs.len(), 2, "two runs: {:?}", res.runs);
assert_eq!(res.most, 2, "unguarded, both children streamed at once");
assert!(
res.runs
.iter()
.any(|(o, _)| *o != Some(crate::entities::subagent::RunOutcome::Completed)),
"the lie about the pool went unpunished โ the server did not overflow: {:?}",
res.runs
);
}
#[tokio::test]
#[ignore = "requires a live chat server launched with -np 4 --kv-unified (MINDFORK_ENGINE_URL)"]
async fn admission_four_e2e_live() {
let Some(res) = admission_smoke(AdmissionArm {
children: 4,
sessions: 4,
share: 0.25,
lie: None,
})
.await
else {
eprintln!("skipped (see above)");
return;
};
assert_eq!(res.runs.len(), 4, "four runs: {:?}", res.runs);
assert_all_completed_with_their_codes(&res.runs);
assert_eq!(
res.most, 2,
"a quarter of the pool plus the cap each: two fit together, three do not"
);
}
const COMPACT_MARKER: &str = "compressing the earlier part of a conversation";
struct SilentArm {
lie: Option<u64>,
}
struct SilentResult {
run: (Option<crate::entities::subagent::RunOutcome>, String),
roll: Result<String, String>,
most: usize,
}
async fn silent_roll_smoke(arm: SilentArm) -> Option<SilentResult> {
let (live, slots, pool) = pooled_live("the collision").await?;
let paragraphs = ((0.55 * pool as f64) / 30.0) as usize;
let max_tokens = (pool / 8) as usize;
let belief = pool * arm.lie.unwrap_or(1);
eprintln!(
"server: {slots} slots over {pool}; history and archive {paragraphs} paragraphs each, cap {max_tokens}, belief {belief}"
);
let delegate = ChatChunk::ToolCall(crate::shared::api::contract::ToolCallDelta {
thought_signature: None,
index: 0,
id: Some("c0".into()),
name: Some("start_subagent".into()),
arguments: serde_json::json!({
"name": "alpha reader",
"system_message": format!("{READER} You are given an archive; report the codename it contains in one short sentence, quoting it exactly."),
"message": format!("{}\nThe alpha codename is {}.\n\nWhat is the codename?", archive("alpha", paragraphs), CODES[0]),
})
.to_string(),
});
let backend = Arc::new(ScriptedParent {
live,
scripts: std::sync::Mutex::new(
vec![
vec![
ChatChunk::Text("Noted.".into()),
ChatChunk::Finished(FinishReason::Stop),
],
vec![delegate, ChatChunk::Finished(FinishReason::ToolCalls)],
vec![
ChatChunk::Text("Started.".into()),
ChatChunk::Finished(FinishReason::Stop),
],
]
.into(),
),
live_markers: &[READER, COMPACT_MARKER],
live_after_scripts: false,
in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
max_in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
});
let mut cfg = AppConfig::default();
cfg.tools.subagent_background = true;
cfg.tools.subagent_background_wake = false;
cfg.tools.subagent_max_tokens = max_tokens;
cfg.default_sampling.max_tokens = Some(max_tokens);
cfg.engine.external.sessions = 1;
cfg.engine.managed.sessions = 1;
cfg.compaction.enabled = true;
cfg.compaction.threshold_pct = 0;
cfg.compaction.context_tokens = Some(belief as usize);
cfg.compaction.tail_tokens = 64;
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let (dir, cmd_tx, mut evt_rx, handle, chat_id) = delegator_chat(
backend.clone() as Arc<dyn EngineBackend>,
cfg,
"start_subagent",
)
.await;
run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
&format!(
"{}\nKeep this archive in mind.",
archive("history", paragraphs)
),
)
.await;
let started = std::time::Instant::now();
let (reply, calls) = run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
"Have the alpha reader report its codename in the background.",
)
.await;
eprintln!("parent: {reply} ({:.1} s)", started.elapsed().as_secs_f64());
for (n, _, r) in &calls {
eprintln!("call {n} -> {}", r.chars().take(120).collect::<String>());
}
cmd_tx.send(AppCommand::Compact).unwrap();
let mut roll: Option<Result<String, String>> = None;
let mut landed = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(300);
while (roll.is_none() || !landed) && std::time::Instant::now() < deadline {
let left = deadline.saturating_duration_since(std::time::Instant::now());
match tokio::time::timeout(left, evt_rx.recv()).await {
Ok(Some(AppEvent::Compacted { summary, .. })) => roll = Some(Ok(summary)),
Ok(Some(AppEvent::Error(msg))) if roll.is_none() => roll = Some(Err(msg)),
Ok(Some(AppEvent::Notice(msg))) if roll.is_none() => roll = Some(Err(msg)),
Ok(Some(AppEvent::BackgroundRuns { out: 0 })) => landed = true,
Ok(Some(_)) => {}
_ => break,
}
}
eprintln!(
"roll: {roll:?}, run landed: {landed} ({:.1} s)",
started.elapsed().as_secs_f64()
);
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let chat = Storage::open(Paths::with_root(dir.path()))
.unwrap()
.json()
.load_chat(chat_id)
.unwrap()
.unwrap();
let run = chat
.children()
.next()
.map(|run| {
eprintln!(
"run ยซ{}ยป: {} messages, outcome {:?}, reply {:?}",
run.title,
run.messages.len(),
run.outcome,
run.final_reply()
.map(|r| r.chars().take(120).collect::<String>())
);
(
run.outcome,
run.final_reply().unwrap_or_default().to_string(),
)
})
.expect("the run landed on its record");
let most = backend
.max_in_flight
.load(std::sync::atomic::Ordering::SeqCst);
eprintln!("most live streams open at once: {most}");
Some(SilentResult {
run,
roll: roll.unwrap_or_else(|| Err("the roll never reported".into())),
most,
})
}
#[tokio::test]
#[ignore = "requires a live llama-server with several slots over one pool (MINDFORK_ENGINE_URL)"]
async fn silent_roll_e2e_live() {
let Some(res) = silent_roll_smoke(SilentArm { lie: None }).await else {
eprintln!("skipped (see above)");
return;
};
assert_eq!(
res.run.0,
Some(crate::entities::subagent::RunOutcome::Completed),
"the run completes beside the roll: {:?}",
res.run
);
assert!(
fold_dashes(&res.run.1).contains(CODES[0]),
"the run's code: {}",
res.run.1
);
let summary = res.roll.expect("the roll produced a summary");
assert!(!summary.trim().is_empty(), "an empty summary");
assert_eq!(
res.most, 1,
"the roll's digest and the run's archive do not fit the pool together: one at a time"
);
}
#[tokio::test]
#[ignore = "requires a live llama-server with several slots over one pool (MINDFORK_ENGINE_URL)"]
async fn silent_roll_control_e2e_live() {
let Some(res) = silent_roll_smoke(SilentArm { lie: Some(4) }).await else {
eprintln!("skipped (see above)");
return;
};
assert_eq!(
res.most, 2,
"unguarded, the roll and the run streamed at once"
);
assert!(
res.run.0 != Some(crate::entities::subagent::RunOutcome::Completed) || res.roll.is_err(),
"the lie about the pool went unpunished โ the server did not overflow: run {:?}, roll {:?}",
res.run,
res.roll
);
}
#[derive(Clone, Copy, Debug)]
enum PreemptArm {
RollThenTurn,
CancelThenTurn,
TurnAlone,
}
#[derive(Debug, Default)]
struct PreemptProbe {
roll: Option<(f64, Result<String, String>)>,
wait: Option<f64>,
cancel_latency: Option<f64>,
first_token: Option<f64>,
finished: Option<f64>,
reply: String,
}
async fn probe_roll_then_turn(
cmd_tx: &UnboundedSender<AppCommand>,
evt_rx: &mut UnboundedReceiver<AppEvent>,
word: &str,
) -> PreemptProbe {
let mut probe = PreemptProbe::default();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(600);
let compacted = std::time::Instant::now();
cmd_tx.send(AppCommand::Compact).unwrap();
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
let sent = std::time::Instant::now();
cmd_tx.send(AppCommand::SendMessage(word.into())).unwrap();
let mut turn_done = false;
while (probe.roll.is_none() || !turn_done) && std::time::Instant::now() < deadline {
let left = deadline.saturating_duration_since(std::time::Instant::now());
match tokio::time::timeout(left, evt_rx.recv()).await {
Ok(Some(AppEvent::Compacted { summary, .. })) => {
probe.roll = Some((compacted.elapsed().as_secs_f64(), Ok(summary)));
probe.wait = Some(sent.elapsed().as_secs_f64());
}
Ok(Some(AppEvent::Error(msg))) | Ok(Some(AppEvent::Notice(msg)))
if probe.roll.is_none() =>
{
probe.roll = Some((compacted.elapsed().as_secs_f64(), Err(msg)));
probe.wait = Some(sent.elapsed().as_secs_f64());
}
Ok(Some(AppEvent::Chunk { text, .. })) => {
probe
.first_token
.get_or_insert(sent.elapsed().as_secs_f64());
probe.reply.push_str(&text);
}
Ok(Some(AppEvent::Finished { .. })) => {
probe.finished = Some(sent.elapsed().as_secs_f64());
turn_done = true;
}
Ok(Some(_)) => {}
_ => break,
}
}
probe
}
async fn probe_cancel_then_turn(
cmd_tx: &UnboundedSender<AppCommand>,
evt_rx: &mut UnboundedReceiver<AppEvent>,
word: &str,
) -> PreemptProbe {
let mut probe = PreemptProbe::default();
cmd_tx
.send(AppCommand::SendMessage(
"Write a long story about the lighthouse keeper's year, at least four hundred words."
.into(),
))
.unwrap();
let mut cancelled_at: Option<std::time::Instant> = None;
loop {
match evt_rx.recv().await {
Some(AppEvent::Chunk { .. }) | Some(AppEvent::Thoughts { .. })
if cancelled_at.is_none() =>
{
cancelled_at = Some(std::time::Instant::now());
cmd_tx.send(AppCommand::Cancel).unwrap();
}
Some(AppEvent::Finished { .. }) => {
probe.cancel_latency = cancelled_at.map(|t| t.elapsed().as_secs_f64());
break;
}
Some(_) => {}
None => break,
}
}
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
read_word_turn(cmd_tx, evt_rx, word, &mut probe).await;
probe
}
async fn read_word_turn(
cmd_tx: &UnboundedSender<AppCommand>,
evt_rx: &mut UnboundedReceiver<AppEvent>,
word: &str,
probe: &mut PreemptProbe,
) {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(600);
let sent = std::time::Instant::now();
cmd_tx.send(AppCommand::SendMessage(word.into())).unwrap();
while std::time::Instant::now() < deadline {
let left = deadline.saturating_duration_since(std::time::Instant::now());
match tokio::time::timeout(left, evt_rx.recv()).await {
Ok(Some(AppEvent::Chunk { text, .. })) => {
probe
.first_token
.get_or_insert(sent.elapsed().as_secs_f64());
probe.reply.push_str(&text);
}
Ok(Some(AppEvent::Finished { .. })) => {
probe.finished = Some(sent.elapsed().as_secs_f64());
break;
}
Ok(Some(_)) => {}
_ => break,
}
}
}
async fn preemption_smoke(arm: PreemptArm) -> Option<PreemptProbe> {
let (live, slots, pool) = pooled_live("the shape").await?;
let paragraphs = ((0.55 * pool as f64) / 30.0) as usize;
eprintln!("server: {slots} slots over {pool}; history {paragraphs} paragraphs, arm {arm:?}");
let backend = Arc::new(ScriptedParent {
live,
scripts: std::sync::Mutex::new(
vec![
vec![
ChatChunk::Text("Noted.".into()),
ChatChunk::Finished(FinishReason::Stop),
],
vec![
ChatChunk::Text("Nothing to add.".into()),
ChatChunk::Finished(FinishReason::Stop),
],
]
.into(),
),
live_markers: &[COMPACT_MARKER],
live_after_scripts: true,
in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
max_in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
});
let mut cfg = AppConfig::default();
cfg.default_sampling.max_tokens = Some(256);
cfg.engine.external.sessions = 1;
cfg.engine.managed.sessions = 1;
cfg.compaction.enabled = true;
cfg.compaction.threshold_pct = 0;
cfg.compaction.context_tokens = Some(pool as usize);
cfg.compaction.tail_tokens = 64;
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let (_dir, cmd_tx, mut evt_rx, handle, _chat_id) = delegator_chat(
backend.clone() as Arc<dyn EngineBackend>,
cfg,
"call_subagent",
)
.await;
run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
&format!(
"{}\nKeep this archive in mind.",
archive("history", paragraphs)
),
)
.await;
run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
&format!("{}\nAnything to add?", archive("tail", 4)),
)
.await;
let word = "Reply with the single word READY and nothing else.";
let probe = match arm {
PreemptArm::RollThenTurn => probe_roll_then_turn(&cmd_tx, &mut evt_rx, word).await,
PreemptArm::CancelThenTurn => probe_cancel_then_turn(&cmd_tx, &mut evt_rx, word).await,
PreemptArm::TurnAlone => {
let mut probe = PreemptProbe::default();
read_word_turn(&cmd_tx, &mut evt_rx, word, &mut probe).await;
probe
}
};
let most = backend
.max_in_flight
.load(std::sync::atomic::Ordering::SeqCst);
eprintln!("{probe:#?}\nmost live streams open at once: {most}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
Some(probe)
}
#[tokio::test]
#[ignore = "requires a live llama-server with several slots over one pool (MINDFORK_ENGINE_URL)"]
async fn preemption_wait_e2e_live() {
let Some(probe) = preemption_smoke(PreemptArm::RollThenTurn).await else {
eprintln!("skipped (see above)");
return;
};
let (_, roll) = probe.roll.expect("the roll reported");
assert!(
roll.is_ok_and(|s| !s.trim().is_empty()),
"the roll produced a summary"
);
assert!(probe.finished.is_some(), "the turn finished");
assert!(!probe.reply.trim().is_empty(), "the turn replied");
let (first_token, roll_end) = (probe.first_token.unwrap(), probe.wait.unwrap());
assert!(
first_token < roll_end,
"the turn streamed before the roll ended: first token at {first_token:.1} s, the roll's end at {roll_end:.1} s"
);
}
#[tokio::test]
#[ignore = "requires a live llama-server with several slots over one pool (MINDFORK_ENGINE_URL)"]
async fn preemption_cold_e2e_live() {
let Some(probe) = preemption_smoke(PreemptArm::TurnAlone).await else {
eprintln!("skipped (see above)");
return;
};
assert!(probe.finished.is_some(), "the turn finished");
assert!(!probe.reply.trim().is_empty(), "the turn replied");
}
#[tokio::test]
#[ignore = "requires a live llama-server with several slots over one pool (MINDFORK_ENGINE_URL)"]
async fn preemption_floor_e2e_live() {
let Some(probe) = preemption_smoke(PreemptArm::CancelThenTurn).await else {
eprintln!("skipped (see above)");
return;
};
assert!(
probe.cancel_latency.is_some(),
"the cancelled turn finished"
);
assert!(
probe.finished.is_some(),
"the turn after the cancel finished"
);
assert!(!probe.reply.trim().is_empty(), "the turn replied");
}
#[tokio::test]
#[ignore = "requires a live llama-server with several slots over one pool (MINDFORK_ENGINE_URL)"]
async fn stop_silent_task_e2e_live() {
let Some((live, slots, pool)) = pooled_live("the roll").await else {
eprintln!("skipped (see above)");
return;
};
let paragraphs = ((0.55 * pool as f64) / 30.0) as usize;
eprintln!("server: {slots} slots over {pool}; history {paragraphs} paragraphs");
let backend = Arc::new(ScriptedParent {
live,
scripts: std::sync::Mutex::new(
vec![
vec![
ChatChunk::Text("Noted.".into()),
ChatChunk::Finished(FinishReason::Stop),
],
vec![
ChatChunk::Text("Nothing to add.".into()),
ChatChunk::Finished(FinishReason::Stop),
],
]
.into(),
),
live_markers: &[COMPACT_MARKER],
live_after_scripts: false,
in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
max_in_flight: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
});
let mut cfg = AppConfig::default();
cfg.engine.external.sessions = 1;
cfg.engine.managed.sessions = 1;
cfg.compaction.enabled = true;
cfg.compaction.threshold_pct = 0;
cfg.compaction.context_tokens = Some(pool as usize);
cfg.compaction.tail_tokens = 64;
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let (dir, cmd_tx, mut evt_rx, handle, chat_id) = delegator_chat(
backend.clone() as Arc<dyn EngineBackend>,
cfg,
"call_subagent",
)
.await;
run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
&format!(
"{}\nKeep this archive in mind.",
archive("history", paragraphs)
),
)
.await;
run_turn_capture_args(
&cmd_tx,
&mut evt_rx,
&format!("{}\nAnything to add?", archive("tail", 4)),
)
.await;
let started = std::time::Instant::now();
cmd_tx.send(AppCommand::Compact).unwrap();
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
cmd_tx
.send(AppCommand::StopBackgroundTask {
kind: BackgroundKind::Compaction,
})
.unwrap();
let stopped_at = std::time::Instant::now();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(120);
let (mut notice, mut compacted) = (None, false);
while notice.is_none() && std::time::Instant::now() < deadline {
let left = deadline.saturating_duration_since(std::time::Instant::now());
match tokio::time::timeout(left, evt_rx.recv()).await {
Ok(Some(AppEvent::Notice(m))) => notice = Some(m),
Ok(Some(AppEvent::Compacted { .. })) => compacted = true,
Ok(Some(AppEvent::Error(m))) => panic!("the stop reported an error: {m}"),
Ok(Some(_)) => {}
_ => break,
}
}
let notice = notice.expect("the stopped roll answered with a notice");
eprintln!(
"stopped {:.1} s after /compact; the notice {:.2} s after the stop: {notice}",
(stopped_at - started).as_secs_f64(),
stopped_at.elapsed().as_secs_f64()
);
assert!(!compacted, "nothing was folded by the stopped roll");
assert_eq!(
backend
.max_in_flight
.load(std::sync::atomic::Ordering::SeqCst),
1
);
let again = std::time::Instant::now();
cmd_tx.send(AppCommand::Compact).unwrap();
let done = tokio::time::timeout(
std::time::Duration::from_secs(300),
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::Compacted { .. })),
)
.await
.ok()
.flatten();
eprintln!(
"the second roll: {:?} ({:.1} s)",
done.as_ref().map(|_| "compacted"),
again.elapsed().as_secs_f64()
);
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
assert!(done.is_some(), "the next /compact completed");
let chat = Storage::open(Paths::with_root(dir.path()))
.unwrap()
.json()
.load_chat(chat_id)
.unwrap()
.unwrap();
assert!(
chat.compaction.is_some(),
"the second roll's summary landed"
);
}
#[tokio::test]
#[ignore = "requires a live chat server (MINDFORK_ENGINE_URL)"]
async fn background_subagent_e2e_live() {
let sandbox = tempfile::tempdir().unwrap();
let file = sandbox.path().join("alpha.txt");
std::fs::write(
&file,
"Internal note.\nThe alpha codename is ZARNOVIK-7741.\n",
)
.unwrap();
let mut cfg = AppConfig::default();
cfg.tools.fs_enabled = true;
cfg.tools.fs_root = Some(sandbox.path().to_string_lossy().to_string());
cfg.tools.subagent_background = true;
cfg.engine.external.sessions = 2;
cfg.engine.managed.sessions = 2;
cfg.interface.auto_title = crate::shared::config::AutoTitleMode::Off;
let Some((_d, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::CreateProfile {
name: "Delegator".into(),
system_message: "You are a coordinator. Reply in English. You never read files \
yourself. When asked to have a file read in the background, start ONE \
sub-agent with the start_subagent tool โ give it the exact file path and \
tell it to use fs_read โ and answer the rest of the request at once, \
without waiting for it. When a task notification with the sub-agent's \
result arrives, report what it found."
.into(),
})
.unwrap();
let pl = wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ProfileList(v) if v.len() >= 2),
)
.await
.unwrap();
let pid = match pl {
AppEvent::ProfileList(v) => v.last().unwrap().id,
_ => unreachable!(),
};
cmd_tx
.send(AppCommand::UpdateProfile {
id: pid,
edit: Box::new(ProfileEdit {
language: Some(crate::shared::i18n::Lang::En),
enabled_tools: Some(vec![
"start_subagent".to_string(),
"call_subagent".to_string(),
"fs_read".to_string(),
]),
..Default::default()
}),
})
.unwrap();
cmd_tx
.send(AppCommand::NewChat {
profile_id: Some(pid),
})
.unwrap();
let chat_id = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.and_then(|e| match e {
AppEvent::ChatActivated { id, .. } => Some(id),
_ => None,
})
.unwrap();
let ask = format!(
"Two things. (1) Have a sub-agent read the file {} in the background and find the \
alpha codename โ start it now with start_subagent and do not wait for it. \
(2) Right now: which is larger, 17 ร 23 or 400?",
file.display()
);
cmd_tx.send(AppCommand::SendMessage(ask)).unwrap();
let mut reply = String::new();
let mut thoughts = 0usize;
let mut calls: Vec<(String, String, String)> = Vec::new();
let mut finish = None;
while let Some(ev) = evt_rx.recv().await {
match ev {
AppEvent::Chunk { text, .. } => reply.push_str(&text),
AppEvent::Thoughts { text, .. } => thoughts += text.len(),
AppEvent::ToolCall {
name,
arguments,
result,
..
} => calls.push((name, arguments, result)),
AppEvent::Error(e) => eprintln!("error event: {e}"),
AppEvent::Finished { reason, .. } => {
finish = Some(reason);
break;
}
_ => {}
}
}
eprintln!("reply ({finish:?}, {thoughts} bytes of thoughts): {reply}");
for (n, a, r) in &calls {
eprintln!(
"call {n}({}) -> {}",
a.chars().take(120).collect::<String>(),
r.chars().take(160).collect::<String>()
);
}
assert!(
calls.iter().any(|(n, _, _)| n == "start_subagent"),
"the parent did not start a background run: {calls:?} ({finish:?}, reply {reply:?})"
);
assert!(
!calls.iter().any(|(n, _, _)| n == "fs_read"),
"the parent read the file itself instead of delegating"
);
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(300);
let mut wake = String::new();
loop {
let ev = tokio::time::timeout_at(deadline, evt_rx.recv())
.await
.expect("the run lands and the wake turn ends within five minutes")
.expect("the event stream stays open");
match ev {
AppEvent::Chunk { text, .. } => wake.push_str(&text),
AppEvent::Finished { .. } => break,
_ => {}
}
}
eprintln!("wake reply: {wake}");
tokio::time::sleep(std::time::Duration::from_millis(1200)).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let chat = Storage::open(Paths::with_root(_d.path()))
.unwrap()
.json()
.load_chat(chat_id)
.unwrap()
.unwrap();
let run = chat
.children()
.find(|r| r.background)
.expect("a background run on the record");
eprintln!(
"run ยซ{}ยป: {} messages, outcome {:?}",
run.title,
run.messages.len(),
run.outcome
);
assert_eq!(
run.outcome,
Some(crate::entities::subagent::RunOutcome::Completed),
"the run did not complete"
);
assert!(
run.messages
.iter()
.flat_map(|m| m.tool_calls.iter())
.any(|r| r.name == "fs_read"),
"the run never used fs_read"
);
let fold = |s: &str| {
s.replace(['-', '\u{2010}', '\u{2011}', '\u{2012}', '\u{2013}'], "")
.to_uppercase()
};
let note = chat
.messages
.iter()
.find(|m| m.is_notification())
.expect("a task notification row");
assert!(
fold(¬e.text).contains("ZARNOVIK7741"),
"the notification lacks the codename: {}",
note.text
);
assert!(
fold(&wake).contains("ZARNOVIK7741"),
"the wake reply lacks the codename: {wake}"
);
}
#[tokio::test]
#[ignore = "requires a running llama-server (MINDFORK_ENGINE_URL)"]
async fn slow_prefill_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let expect_note = std::env::var("MINDFORK_EXPECT_SLOW_PREFILL").is_ok();
let mut config = AppConfig::default();
config.engine.mode = crate::shared::config::ServerMode::External;
let (_d, cmd_tx, mut evt_rx, handle) = spawn_orch_cfg(Some(backend), config);
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let seed: String = (0..40)
.map(|i| {
format!(
"Paragraph {i}: the keeper climbs the stairs, lights the lamp, writes the log, \
notes the tide, and looks out over the dark water for a while.\n"
)
})
.collect();
let started = std::time::Instant::now();
let (reply, _tools) = run_turn_live(
&cmd_tx,
&mut evt_rx,
&format!("{seed}\nIn one word: what does the keeper light?"),
)
.await;
let turn = started.elapsed();
let note = tokio::time::timeout(
std::time::Duration::from_secs(3),
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::Notice(_))),
)
.await
.ok()
.flatten()
.and_then(|e| match e {
AppEvent::Notice(t) => Some(t),
_ => None,
});
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!(
"slow prefill: the turn took {:.1} s, reply {:?}; note = {}",
turn.as_secs_f64(),
reply.trim(),
note.as_deref().unwrap_or("none")
);
match (expect_note, note) {
(true, Some(text)) => assert!(text.contains("-b 256 -ub 256"), "{text}"),
(true, None) => panic!("a slow host was expected to be told"),
(false, None) => {}
(false, Some(text)) => panic!("a fast host was told: {text}"),
}
}
#[tokio::test]
#[ignore = "requires a running llama-server (MINDFORK_ENGINE_URL)"]
async fn roll_prefill_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let expect_note = std::env::var("MINDFORK_EXPECT_SLOW_PREFILL").is_ok();
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let json = crate::shared::storage::JsonStore::new(Paths::with_root(&root));
let profile = Profile::new("Keeper", "You are a concise assistant.");
json.upsert_profile(&profile).unwrap();
let mut chat = Chat::from_profile(&profile, "the keeper's evening");
for part in 0..4u32 {
let seed: String = (part * 10..part * 10 + 10)
.map(|i| {
format!(
"Paragraph {i}: the keeper climbs the stairs, lights the lamp, writes the log, \
notes the tide, and looks out over the dark water for a while.\n"
)
})
.collect();
chat.push_message(Message::user(format!("{seed}\nAcknowledge in one word.")));
chat.push_message(Message::assistant(String::from("Noted.")));
}
let chat_id = chat.id;
json.save_chat(&chat).unwrap();
drop(Storage::open(Paths::with_root(&root)).unwrap());
let mut config = AppConfig::default();
config.engine.mode = crate::shared::config::ServerMode::External;
config.compaction = crate::shared::config::CompactionSettings {
enabled: true,
summary_words: 150,
tail_tokens: 120,
..Default::default()
};
let (cmd_tx, mut evt_rx, handle) = spawn_orch_at(&root, Some(backend), config);
let activated = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
if !matches!(activated, AppEvent::ChatActivated { id, .. } if id == chat_id) {
cmd_tx.send(AppCommand::SwitchChat(chat_id)).unwrap();
wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ChatActivated { id, .. } if *id == chat_id),
)
.await
.unwrap();
}
let started = std::time::Instant::now();
cmd_tx.send(AppCommand::Compact).unwrap();
let landed = wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::Compacted { .. } | AppEvent::Error(_) | AppEvent::Notice(_)
)
})
.await
.unwrap();
let AppEvent::Compacted {
summary, folded, ..
} = landed
else {
panic!("the roll did not land as a summary: {landed:?}");
};
let roll = started.elapsed();
let note = tokio::time::timeout(
std::time::Duration::from_secs(3),
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::Notice(_))),
)
.await
.ok()
.flatten()
.and_then(|e| match e {
AppEvent::Notice(t) => Some(t),
_ => None,
});
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!(
"roll prefill: the roll took {:.1} s, folded {folded} messages into {} chars; note = {}",
roll.as_secs_f64(),
summary.chars().count(),
note.as_deref().unwrap_or("none")
);
match (expect_note, note) {
(true, Some(text)) => assert!(text.contains("-b 256 -ub 256"), "{text}"),
(true, None) => panic!("a slow host was expected to be told by its roll"),
(false, None) => {}
(false, Some(text)) => panic!("a fast host was told: {text}"),
}
}
#[tokio::test]
#[ignore = "requires a running llama-server (MINDFORK_ENGINE_URL)"]
async fn loop_prefill_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let expect_note = std::env::var("MINDFORK_EXPECT_SLOW_PREFILL").is_ok();
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let external = || {
let mut config = AppConfig::default();
config.engine.mode = crate::shared::config::ServerMode::External;
config
};
let (cmd_tx, mut evt_rx, handle) = spawn_orch_at(&root, Some(backend.clone()), external());
let _pid = enable_all_tools(&cmd_tx, &mut evt_rx).await;
let seed: String = (0..40)
.map(|i| {
format!(
"Paragraph {i}: the keeper climbs the stairs, lights the lamp, writes the log, \
notes the tide, and looks out over the dark water for a while.\n"
)
})
.collect();
let (reply, _) = run_turn_live(
&cmd_tx,
&mut evt_rx,
&format!("{seed}\nIn one word: what does the keeper light?"),
)
.await;
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatList(_)))
.await
.unwrap();
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!("phase 1: reply {:?}", reply.trim());
let mut config = external();
config.self_model.auto_reflect_every = 1;
let (cmd_tx, mut evt_rx, handle) = spawn_orch_at(&root, Some(backend), config);
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let started = std::time::Instant::now();
let (reply, _) =
run_turn_live(&cmd_tx, &mut evt_rx, "And in one word: what does he note?").await;
let turn = started.elapsed();
tokio::time::timeout(
std::time::Duration::from_secs(30),
wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::BackgroundTask {
kind: BackgroundKind::Reflection,
active: true
}
)
}),
)
.await
.expect("the reflection was spawned at the landing");
let landed = tokio::time::timeout(
std::time::Duration::from_secs(300),
wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::BackgroundTask {
kind: BackgroundKind::Reflection,
active: false
}
)
}),
)
.await;
let loop_took = started.elapsed() - turn;
let note = tokio::time::timeout(
std::time::Duration::from_secs(3),
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::Notice(_))),
)
.await
.ok()
.flatten()
.and_then(|e| match e {
AppEvent::Notice(t) => Some(t),
_ => None,
});
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!(
"loop prefill: the turn took {:.1} s (reply {:?}), the reflection landed = {}, {:.1} s after it; note = {}",
turn.as_secs_f64(),
reply.trim(),
landed.is_ok(),
loop_took.as_secs_f64(),
note.as_deref().unwrap_or("none")
);
assert!(landed.is_ok(), "the reflection did not land");
match (expect_note, note) {
(true, Some(text)) => assert!(text.contains("-b 256 -ub 256"), "{text}"),
(true, None) => panic!("a slow host was expected to be told by its loop"),
(false, None) => {}
(false, Some(text)) => panic!("a fast host was told: {text}"),
}
}
type Seen = Arc<std::sync::Mutex<Vec<(crate::shared::api::ChatRequest, u32)>>>;
struct EstimateProbe {
inner: Arc<dyn EngineBackend>,
seen: Seen,
}
#[async_trait::async_trait]
impl EngineBackend for EstimateProbe {
async fn chat_stream(
&self,
req: crate::shared::api::ChatRequest,
cancel: tokio_util::sync::CancellationToken,
) -> anyhow::Result<crate::shared::api::contract::ChatStream> {
use futures_util::StreamExt as _;
let mut inner = self.inner.chat_stream(req.clone(), cancel).await?;
let seen = self.seen.clone();
let s = async_stream::stream! {
while let Some(chunk) = inner.next().await {
if let ChatChunk::Usage(u) = &chunk {
seen.lock().unwrap().push((req.clone(), u.prompt_tokens));
}
yield chunk;
}
};
Ok(Box::pin(s))
}
async fn context_budget(&self) -> Option<u32> {
self.inner.context_budget().await
}
}
#[tokio::test]
#[ignore = "requires a running llama-server (MINDFORK_ENGINE_URL)"]
async fn prompt_estimate_e2e_live() {
let Some(live) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let probe = Arc::new(EstimateProbe {
inner: live,
seen: Arc::new(std::sync::Mutex::new(Vec::new())),
});
let config = AppConfig {
compaction: crate::shared::config::CompactionSettings {
enabled: true,
summary_words: 150,
tail_tokens: 120,
..Default::default()
},
..Default::default()
};
let (_d, cmd_tx, mut evt_rx, handle) =
spawn_orch_cfg(Some(probe.clone() as Arc<dyn EngineBackend>), config);
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
let blob = serde_json::to_string(
&(0..220)
.map(|i| {
serde_json::json!({
"id": i,
"sku": format!("A{i:04}-{:03}", i * 7 % 1000),
"price": (i as f64) * 1.25 + 0.99,
"tags": ["alpha", "beta", "gamma"],
"stock": {"warehouse": i % 5, "count": i * 3},
})
})
.collect::<Vec<_>>(),
)
.unwrap();
let json_turn = format!(
"Here is a catalogue as JSON; answer in one sentence how many items it has:\n{blob}"
);
for text in [
"ะ ะฐััะบะฐะถะธ ะฒ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
, ะทะฐัะตะผ ะฝัะถะฝั ะธะฝะดะตะบัั ะฒ ะฑะฐะทะฐั
ะดะฐะฝะฝัั
.",
"ะ ะดะฒัั
ะฟัะตะดะปะพะถะตะฝะธัั
: ัะตะผ ะพัะปะธัะฐะตััั ะบัั ะพั ะฑััะตัะฐ?",
json_turn.as_str(),
] {
let (reply, _) = run_turn_capture(&cmd_tx, &mut evt_rx, text).await;
eprintln!("reply: {}", reply.chars().take(60).collect::<String>());
wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatList(_)))
.await
.unwrap();
}
cmd_tx.send(AppCommand::Compact).unwrap();
let landed = wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::Compacted { .. } | AppEvent::Error(_))
})
.await
.unwrap();
assert!(
matches!(landed, AppEvent::Compacted { .. }),
"the roll did not land: {landed:?}"
);
cmd_tx
.send(AppCommand::Impersonate {
seed: String::new(),
})
.unwrap();
let ended = wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::ImpersonationFinished { .. } | AppEvent::Error(_)
)
})
.await
.unwrap();
assert!(
matches!(ended, AppEvent::ImpersonationFinished { .. }),
"impersonation did not finish: {ended:?}"
);
eprintln!("impersonation ended: {ended:?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let roll_system = crate::features::compaction::summary_system_message(
crate::shared::i18n::locale(crate::shared::i18n::Lang::default()),
150,
);
let seen = probe.seen.lock().unwrap().clone();
let (mut turns, mut rolls, mut dense, mut impersonations) = (0, 0, 0, 0);
for (req, exact) in &seen {
let estimate = super::super::generation::estimate_prompt_tokens(req);
let ratio = *exact as f64 / estimate as f64;
let is_roll = req.system.as_deref() == Some(roll_system.as_str());
let carries_json = req
.messages
.iter()
.any(|m| m.content.contains("catalogue as JSON"))
|| req
.system
.as_deref()
.is_some_and(|s| s.contains("catalogue as JSON"));
let kind = match (req.tools.is_empty(), is_roll, carries_json) {
(false, _, _) => "turn",
(true, true, _) => "roll",
(true, false, true) => "impersonation",
(true, false, false) => "title",
};
eprintln!(
"prompt estimate: {kind:<13} tools={:>2} json={} estimate={estimate:>5} exact={exact:>5} exact/estimate={ratio:.2}",
req.tools.len(),
u8::from(carries_json)
);
match kind {
"turn" if carries_json => {
turns += 1;
dense += 1;
assert!(ratio > 1.0, "the JSON turn did not under-count: {ratio:.2}");
}
"turn" => {
turns += 1;
assert!(
(0.75..=1.25).contains(&ratio),
"a prose turn's estimate is off by more than a quarter: {ratio:.2}"
);
}
"impersonation" => {
impersonations += 1;
assert!(
ratio > 1.0,
"impersonation over the JSON did not under-count: {ratio:.2}"
);
}
"roll" => {
rolls += 1;
assert!(ratio <= 1.0, "the roll under-counted: {ratio:.2}");
}
_ => assert!(ratio <= 1.0, "the title under-counted: {ratio:.2}"),
}
}
assert!(
turns >= 3 && dense >= 1 && rolls >= 1 && impersonations >= 1,
"{turns} turns ({dense} dense), {rolls} rolls, {impersonations} impersonations seen; \
the impersonation ended {ended:?}"
);
}
fn seed_long_chat(root: &std::path::Path) -> Uuid {
let json = crate::shared::storage::JsonStore::new(Paths::with_root(root));
let profile = Profile::new("Keeper", "You are a concise assistant.");
json.upsert_profile(&profile).unwrap();
let mut chat = Chat::from_profile(&profile, "the keeper's evening");
for part in 0..4u32 {
let seed: String = (part * 8..part * 8 + 8)
.map(|i| {
format!(
"Paragraph {i}: the keeper climbs the stairs, lights the lamp, writes the log, \
notes the tide, and looks out over the dark water for a while.\n"
)
})
.collect();
chat.push_message(Message::user(format!("{seed}\nAcknowledge in one word.")));
chat.push_message(Message::assistant(String::from("Noted.")));
}
let chat_id = chat.id;
json.save_chat(&chat).unwrap();
drop(Storage::open(Paths::with_root(root)).unwrap());
chat_id
}
async fn spawn_on_seeded_chat(
root: &std::path::Path,
backend: Arc<dyn EngineBackend>,
chat_id: Uuid,
) -> (
UnboundedSender<AppCommand>,
UnboundedReceiver<AppEvent>,
tokio::task::JoinHandle<()>,
) {
let mut config = AppConfig::default();
config.engine.mode = crate::shared::config::ServerMode::External;
let (cmd_tx, mut evt_rx, handle) = spawn_orch_at(root, Some(backend), config);
let activated = wait_for(&mut evt_rx, |e| matches!(e, AppEvent::ChatActivated { .. }))
.await
.unwrap();
if !matches!(activated, AppEvent::ChatActivated { id, .. } if id == chat_id) {
cmd_tx.send(AppCommand::SwitchChat(chat_id)).unwrap();
wait_for(
&mut evt_rx,
|e| matches!(e, AppEvent::ChatActivated { id, .. } if *id == chat_id),
)
.await
.unwrap();
}
(cmd_tx, evt_rx, handle)
}
async fn note_after_landing(evt_rx: &mut UnboundedReceiver<AppEvent>) -> Option<String> {
tokio::time::timeout(
std::time::Duration::from_secs(3),
wait_for(evt_rx, |e| matches!(e, AppEvent::Notice(_))),
)
.await
.ok()
.flatten()
.and_then(|e| match e {
AppEvent::Notice(t) => Some(t),
_ => None,
})
}
fn assert_note(expect_note: bool, note: Option<String>, what: &str) {
match (expect_note, note) {
(true, Some(text)) => assert!(text.contains("-b 256 -ub 256"), "{text}"),
(true, None) => panic!("a slow host was expected to be told by its {what}"),
(false, None) => {}
(false, Some(text)) => panic!("a fast host was told: {text}"),
}
}
#[tokio::test]
#[ignore = "requires a running llama-server (MINDFORK_ENGINE_URL)"]
async fn impersonation_prefill_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let expect_note = std::env::var("MINDFORK_EXPECT_SLOW_PREFILL").is_ok();
let dir = tempfile::tempdir().unwrap();
let chat_id = seed_long_chat(dir.path());
let (cmd_tx, mut evt_rx, handle) = spawn_on_seeded_chat(dir.path(), backend, chat_id).await;
let started = std::time::Instant::now();
cmd_tx
.send(AppCommand::Impersonate {
seed: String::new(),
})
.unwrap();
let ended = wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::ImpersonationFinished { .. } | AppEvent::Error(_)
)
})
.await
.unwrap();
assert!(
matches!(ended, AppEvent::ImpersonationFinished { .. }),
"impersonation did not finish: {ended:?}"
);
let took = started.elapsed();
let note = note_after_landing(&mut evt_rx).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!(
"impersonation prefill: the request took {:.1} s; note = {}",
took.as_secs_f64(),
note.as_deref().unwrap_or("none")
);
assert_note(expect_note, note, "impersonation");
}
#[tokio::test]
#[ignore = "requires a running llama-server (MINDFORK_ENGINE_URL)"]
async fn title_prefill_e2e_live() {
let Some(backend) = live_backend() else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let expect_note = std::env::var("MINDFORK_EXPECT_SLOW_PREFILL").is_ok();
let dir = tempfile::tempdir().unwrap();
let chat_id = seed_long_chat(dir.path());
let (cmd_tx, mut evt_rx, handle) = spawn_on_seeded_chat(dir.path(), backend, chat_id).await;
let started = std::time::Instant::now();
cmd_tx.send(AppCommand::AutoRenameChat(chat_id)).unwrap();
let landed = wait_for(&mut evt_rx, |e| {
matches!(
e,
AppEvent::ChatRenamed { .. } | AppEvent::ChatListError(_) | AppEvent::Error(_)
)
})
.await
.unwrap();
let AppEvent::ChatRenamed { title, .. } = landed else {
panic!("the title did not land: {landed:?}");
};
let took = started.elapsed();
let note = note_after_landing(&mut evt_rx).await;
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
eprintln!(
"title prefill: the request took {:.1} s, the title {title:?}; note = {}",
took.as_secs_f64(),
note.as_deref().unwrap_or("none")
);
assert_note(expect_note, note, "title");
}
#[tokio::test]
#[ignore = "requires MINDFORK_ENGINE_URL (a live OpenAI-compatible server)"]
async fn the_model_catalogue_reaches_the_ui_e2e_live() {
let Ok(url) = std::env::var("MINDFORK_ENGINE_URL") else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
let mut cfg = no_auto_cfg();
cfg.engine.mode = crate::shared::config::ServerMode::External;
cfg.engine.external.url = Some(url.clone());
if std::env::var("MINDFORK_ENGINE_KEY").is_ok_and(|k| !k.is_empty()) {
cfg.engine.external.api_key_env = Some("MINDFORK_ENGINE_KEY".into());
}
let Some((_dir, cmd_tx, mut evt_rx, handle)) = spawn_orch_live_cfg(cfg) else {
eprintln!("skip: MINDFORK_ENGINE_URL not set");
return;
};
cmd_tx
.send(AppCommand::ListModels(
crate::shared::api::catalogue::ModelSlot::Assistant,
))
.unwrap();
let landed = tokio::time::timeout(
Duration::from_secs(30),
wait_for(&mut evt_rx, |e| {
matches!(e, AppEvent::ModelCatalogue { .. })
}),
)
.await
.expect("the catalogue must answer the screen that asked")
.expect("the event stream");
match landed {
AppEvent::ModelCatalogue { slot, models } => {
assert_eq!(slot, crate::shared::api::catalogue::ModelSlot::Assistant);
let list = models.expect("the server answered its catalogue");
eprintln!("{url} listed {} model(s):", list.len());
for m in list.iter().take(5) {
eprintln!(" {} {:?}", m.id, m.role);
}
assert!(!list.is_empty(), "a server with a model loaded lists it");
assert!(
list.iter().all(|m| !m.id.is_empty()),
"an empty id could not be written into the field"
);
}
other => panic!("expected a catalogue, got {other:?}"),
}
drop(cmd_tx);
let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}