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))
}
#[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 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:?}"
);
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 evt_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 evt_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:?}");
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
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"))
);
}
}
#[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();
let first = first.to_lowercase();
assert!(
first.contains("blue"),
"the model must see the background colour, got: {first}"
);
assert!(
first.contains("square"),
"the model must see the centred shape, got: {first}"
);
assert!(
second.to_lowercase().contains("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!(
first.to_lowercase().contains("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();
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");
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!(
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 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();
let reply = reply.to_lowercase();
assert!(
reply.contains("green"),
"the downloaded image's background colour, got: {reply}"
);
assert!(
reply.contains("square"),
"the downloaded image's centred shape, got: {reply}"
);
let control = control.to_lowercase();
assert!(
!(control.contains("green") && control.contains("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 (url, hits) = crate::shared::net::stub::counting_stub();
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 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(&cmd_tx, &mut evt_rx, ask).await;
eprintln!(
"parent reply: {}",
reply.chars().take(300).collect::<String>()
);
assert!(
calls.iter().any(|(n, _)| n == "run_dialogue"),
"the model never called run_dialogue; calls: {calls:?}"
);
cmd_tx.send(AppCommand::Quit).unwrap();
handle.await.unwrap();
let chat = super::subagent::load(dir.path(), chat_id);
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() {
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() {
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:?}"
);
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"
);
}
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());
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;
}