mind-body: a delivered chat image reaches her session as an image block

ChatDelivery gains tuple index 11, `image: Option<ChatImage>` ([path, mime,
size, sha256], a reference into her per-room image store, never bytes); the
pending kind is gamecult.persona.chat_pending.v2, and v1 answers and records
still decode with no image. chat_image::load is the one place a reference
becomes bytes: exact `<hex sha256 of room id>/<sha256>[.ext]` path with no
traversal or symlink, regular file, 5 MiB declared and actual, declared size
and sha256, png/jpeg/gif/webp magic bytes matching the declared mime. The
persona line carries the image as a base64 image block after its text and an
`attachment role=organ` line saying attached or not shown and why. The
stream-json echo of a line with blocks still counts as our echo. The store is
MIND_PERSONA_IMAGE_STORE; unset, every image is not shown.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Eureka 2026-11-07 20:52:30 +00:00
commit e2c9638f27
19 changed files with 712 additions and 28 deletions

1
Cargo.lock generated
View file

@ -1146,6 +1146,7 @@ name = "mind-body"
version = "0.1.0"
dependencies = [
"anyhow",
"base64",
"chrono",
"cultcache-rs 0.2.0",
"cultnet-rs 0.1.0",

View file

@ -32,6 +32,7 @@ publish.workspace = true
# `uuid`'s v5 makes a Self run's session id derivable from the run's id.
[dependencies]
anyhow = "1"
base64 = "0.22"
chrono = { version = "0.4.44", features = ["serde"] }
cultcache-rs.workspace = true
cultnet-rs.workspace = true

View file

@ -9,6 +9,7 @@ fn said(text: &str, mentions: &[&str]) -> ChatDelivery {
room_id: "!r:chat".into(),
room_alias: "swarm".into(),
is_direct: false,
image: None,
sender: "@op:chat".into(),
sender_display: "Op".into(),
at_unix_ms: 0,

View file

@ -18,7 +18,9 @@ use serde::{Deserialize, Serialize, de::DeserializeOwned};
/// Document kinds, as the mouth names them.
pub const PENDING_REQUEST: &str = "gamecult.persona.chat_pending_request.v1";
pub const PENDING: &str = "gamecult.persona.chat_pending.v1";
pub const PENDING: &str = "gamecult.persona.chat_pending.v2";
/// The pending answer of a mouth that has no image field yet: its deliveries decode with `image: None`.
pub const PENDING_V1: &str = "gamecult.persona.chat_pending.v1";
pub const ACK: &str = "gamecult.persona.chat_ack.v1";
pub const SEND_REQUEST: &str = "gamecult.persona.chat_send_request.v2";
pub const SEND_RECEIPT: &str = "gamecult.persona.chat_send_receipt.v2";
@ -48,6 +50,20 @@ pub struct ChatDelivery {
pub mentions: Vec<String>,
/// Whether the room is a direct message.
pub is_direct: bool,
/// The attachment the mouth stored for an image event. Tuple index 11; a v1 record ends at 10 and reads as none.
#[serde(default)]
pub image: Option<ChatImage>,
}
/// An image the mouth fetched and stored, by reference: never its bytes. The path is relative to the Persona's image
/// store, `<hex sha256 of the room id>/<hex sha256 of the bytes>[.<ext>]`, and every field is re-checked against the
/// file before it reaches a model (`chat_image`).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChatImage {
pub path: String,
pub mime: String,
pub size: u64,
pub sha256: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

View file

@ -7,7 +7,9 @@ use super::*;
const REFUSAL_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_refusal.v1.msgpack");
const ACK_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack");
const PENDING_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack");
const PENDING_V1_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack");
const PENDING_NO_IMAGE_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.no-image.msgpack");
const PENDING_IMAGE_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack");
const PENDING_REQUEST_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack");
const SENT_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack");
const ROOM_MOVED_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.room-moved.msgpack");
@ -27,9 +29,8 @@ fn the_pending_request_and_ack_are_the_mouths_bytes() {
golden(ACK_BYTES, &ChatAck { persona: "eureka".into(), through_seq: 7 });
}
#[test]
fn a_pending_answer_is_the_mouths_bytes_with_its_context() {
let delivery = ChatDelivery {
fn delivery(image: Option<ChatImage>) -> ChatDelivery {
ChatDelivery {
seq: 7,
room_alias: "#operator:gamecult.org".into(),
room_id: "!abc:gamecult.org".into(),
@ -41,8 +42,46 @@ fn a_pending_answer_is_the_mouths_bytes_with_its_context() {
context: vec![ContextLine { sender: "@eureka:gamecult.org".into(), text: "earlier".into(), at_unix_ms: 1_761_699_999_000 }],
mentions: vec!["@eureka:gamecult.org".into(), "@nova:gamecult.org".into()],
is_direct: true,
};
golden(PENDING_BYTES, &ChatPending { persona: "eureka".into(), deliveries: vec![delivery] });
image,
}
}
fn image() -> ChatImage {
let sha256 = "ab".repeat(32);
ChatImage { path: format!("{}/{sha256}.png", "cd".repeat(32)), mime: "image/png".into(), size: 2048, sha256 }
}
#[test]
fn a_v1_pending_answer_still_decodes_with_no_image() {
let pending: ChatPending = decode(PENDING_V1_BYTES).unwrap();
assert_eq!(pending, ChatPending { persona: "eureka".into(), deliveries: vec![delivery(None)] });
assert_eq!(PENDING_V1_BYTES[0] & 0xf0, 0x90, "a positional tuple");
}
#[test]
fn a_v2_pending_answer_is_the_golden_bytes_with_the_image_at_tuple_index_11() {
golden(PENDING_NO_IMAGE_BYTES, &ChatPending { persona: "eureka".into(), deliveries: vec![delivery(None)] });
golden(PENDING_IMAGE_BYTES, &ChatPending { persona: "eureka".into(), deliveries: vec![delivery(Some(image()))] });
// Index 11 of the delivery tuple, by position: 12 elements, the last an array of [path, mime, size, sha256].
let value = shape::delivery_tuple(PENDING_IMAGE_BYTES);
assert_eq!(value, shape::Probe { delivery_len: 12, image_len: Some(4) });
assert_eq!(shape::delivery_tuple(PENDING_NO_IMAGE_BYTES), shape::Probe { delivery_len: 12, image_len: None });
}
/// Reads the tuple's shape from the raw bytes, independent of the serde types under test.
mod shape {
#[derive(Debug, PartialEq, Eq)]
pub struct Probe {
pub delivery_len: usize,
pub image_len: Option<usize>,
}
pub fn delivery_tuple(bytes: &[u8]) -> Probe {
let value: serde_json::Value = rmp_serde::from_slice(bytes).unwrap();
let delivery = &value[1][0];
let array = delivery.as_array().expect("a delivery is a tuple");
Probe { delivery_len: array.len(), image_len: array[11].as_array().map(Vec::len) }
}
}
#[test]
@ -59,3 +98,12 @@ fn a_send_and_its_receipt_are_the_mouths_bytes() {
golden(SENT_BYTES, &ChatSendReceipt { request_id: "req-1".into(), event_id: Some("$sent".into()), refusal: None, newer: vec![] });
golden(ROOM_MOVED_BYTES, &ChatSendReceipt { request_id: "req-1".into(), event_id: None, refusal: Some(Refusal::RoomMoved), newer: vec!["$a".into(), "$b".into()] });
}
/// Writes the v2 fixtures; run once with `WRITE_CHAT_FIXTURES=1` when the wire changes.
#[test]
#[ignore = "regenerates the golden fixtures"]
fn write_the_v2_pending_fixtures() {
let dir = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fixtures/persona-chat-v3/");
std::fs::write(format!("{dir}gamecult.persona.chat_pending.v2.no-image.msgpack"), encode(&ChatPending { persona: "eureka".into(), deliveries: vec![delivery(None)] }).unwrap()).unwrap();
std::fs::write(format!("{dir}gamecult.persona.chat_pending.v2.image.msgpack"), encode(&ChatPending { persona: "eureka".into(), deliveries: vec![delivery(Some(image()))] }).unwrap()).unwrap();
}

View file

@ -0,0 +1,146 @@
//! A stored chat image, loaded for a model: the one place an image reference from the mouth becomes bytes.
//!
//! The mouth (Bifrost) fetches an image event's media with the Persona's own credential and stores it under that
//! Persona's image store, `<store>/<hex sha256 of the room id>/<hex sha256 of the bytes>[.<ext>]`. The delivery carries
//! a `ChatImage` reference, never bytes. Nothing in it is trusted here: the path is checked to be that exact shape, to
//! name this delivery's room and to resolve to itself with every link followed (so no `..`, no absolute path and no
//! symlink gets out); the file's length is held to `MAX_BYTES` and to the declared size, its sha256 to the declared one,
//! and its magic bytes to png, jpeg, gif or webp and to the declared mime. Only then is it base64. A reference that
//! fails any of it is a `Skip`, a fixed reason the Persona is told in words, and the bytes are never in a log, an
//! error or a message.
use std::fs::File;
use std::io::Read;
use std::path::{Component, Path};
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use sha2::{Digest, Sha256};
use crate::chat_client::ChatImage;
/// The most a stored image may weigh, declared and actual.
pub(crate) const MAX_BYTES: u64 = 5 * 1024 * 1024;
/// The most images one line to a session carries; the rest are skipped.
pub(crate) const MAX_PER_LINE: usize = 4;
/// An image ready for a model's user message.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct Image {
pub(crate) media_type: &'static str,
pub(crate) base64: String,
}
/// Why an image is not shown. Fixed words: nothing the sender or the mouth wrote reaches a Persona through them.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Skip {
NoStore,
BadReference,
OutsideStore,
Missing,
TooLarge,
SizeMismatch,
HashMismatch,
NotAnImage,
MimeMismatch,
TooMany,
}
impl Skip {
pub(crate) fn reason(self) -> &'static str {
match self {
Self::NoStore => "this Persona has no image store",
Self::BadReference => "the image reference is malformed",
Self::OutsideStore => "the image reference leaves this room's store",
Self::Missing => "the stored image cannot be read",
Self::TooLarge => "the image is over 5 MiB",
Self::SizeMismatch => "the stored image is not the declared size",
Self::HashMismatch => "the stored image is not the declared bytes",
Self::NotAnImage => "the stored file is not a png, jpeg, gif or webp",
Self::MimeMismatch => "the stored image is not the declared type",
Self::TooMany => "too many images in one line",
}
}
}
/// Loads `image`, which was delivered for `room_id`, from `store`.
pub(crate) fn load(store: Option<&Path>, room_id: &str, image: &ChatImage) -> Result<Image, Skip> {
let store = store.ok_or(Skip::NoStore)?;
if image.size > MAX_BYTES {
return Err(Skip::TooLarge);
}
let relative = Path::new(&image.path);
let room = room_dir(room_id);
let [Component::Normal(room_part), Component::Normal(file_part)] = relative.components().collect::<Vec<_>>()[..] else {
return Err(Skip::BadReference);
};
let (Some(room_part), Some(file_part)) = (room_part.to_str(), file_part.to_str()) else {
return Err(Skip::BadReference);
};
if relative.as_os_str() != format!("{room_part}/{file_part}").as_str() || !is_stored_name(file_part, &image.sha256) {
return Err(Skip::BadReference);
}
if room_part != room {
return Err(Skip::OutsideStore);
}
let root = store.canonicalize().map_err(|_| Skip::Missing)?;
let wanted = root.join(room_part).join(file_part);
// Every link followed, the file must still be exactly where the reference says: a symlink anywhere on it moves it.
if wanted.canonicalize().map_err(|_| Skip::Missing)? != wanted {
return Err(Skip::OutsideStore);
}
// Checked before the open: opening a FIFO would wait for a writer.
let meta = std::fs::symlink_metadata(&wanted).map_err(|_| Skip::Missing)?;
if !meta.is_file() {
return Err(Skip::Missing);
}
if meta.len() > MAX_BYTES {
return Err(Skip::TooLarge);
}
let mut bytes = Vec::new();
// A file that grew since the check reads as another size than the declared one.
File::open(&wanted).map_err(|_| Skip::Missing)?.take(MAX_BYTES + 1).read_to_end(&mut bytes).map_err(|_| Skip::Missing)?;
if bytes.len() as u64 != image.size {
return Err(Skip::SizeMismatch);
}
if hex(&Sha256::digest(&bytes)) != image.sha256 {
return Err(Skip::HashMismatch);
}
let media_type = sniff(&bytes).ok_or(Skip::NotAnImage)?;
if image.mime != media_type {
return Err(Skip::MimeMismatch);
}
Ok(Image { media_type, base64: STANDARD.encode(&bytes) })
}
/// The media type the first bytes say, among the four a model takes.
fn sniff(bytes: &[u8]) -> Option<&'static str> {
if bytes.starts_with(b"\x89PNG\r\n\x1a\n") {
Some("image/png")
} else if bytes.starts_with(&[0xff, 0xd8, 0xff]) {
Some("image/jpeg")
} else if bytes.starts_with(b"GIF87a") || bytes.starts_with(b"GIF89a") {
Some("image/gif")
} else if bytes.len() >= 12 && &bytes[..4] == b"RIFF" && &bytes[8..12] == b"WEBP" {
Some("image/webp")
} else {
None
}
}
/// The file's name: the bytes' sha256, then at most a short alphanumeric extension.
fn is_stored_name(name: &str, sha256: &str) -> bool {
name.strip_prefix(sha256).is_some_and(|rest| rest.is_empty() || rest.strip_prefix('.').is_some_and(|ext| (1..=5).contains(&ext.len()) && ext.bytes().all(|byte| byte.is_ascii_alphanumeric())))
}
/// The name of a room's directory in the store: the hex sha256 of its Matrix room id.
pub(crate) fn room_dir(room_id: &str) -> String {
hex(&Sha256::digest(room_id.as_bytes()))
}
pub(crate) fn hex(bytes: &[u8]) -> String {
bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}
#[cfg(test)]
mod tests;

View file

@ -0,0 +1,223 @@
//! The image guards, each against the one input it exists for. Every test builds a real store on disk; a guard
//! removed from `load` lets exactly its test's input through.
use std::os::unix::fs::symlink;
use super::*;
const ROOM: &str = "!room:gamecult.test";
const PNG: &[u8] = b"\x89PNG\r\n\x1a\nnot a real picture, only its magic bytes and a tail";
struct Store {
dir: tempfile::TempDir,
}
impl Store {
fn new() -> Self {
Self { dir: tempfile::tempdir().unwrap() }
}
fn root(&self) -> &Path {
self.dir.path()
}
/// Stores `bytes` the way the mouth does, in `room`'s directory, and names them honestly.
fn put(&self, room: &str, bytes: &[u8]) -> ChatImage {
let sha256 = hex(&Sha256::digest(bytes));
let room = room_dir(room);
std::fs::create_dir_all(self.root().join(&room)).unwrap();
std::fs::write(self.root().join(&room).join(&sha256), bytes).unwrap();
ChatImage { path: format!("{room}/{sha256}"), mime: sniff(bytes).unwrap_or("image/png").into(), size: bytes.len() as u64, sha256 }
}
fn load(&self, image: &ChatImage) -> Result<Image, Skip> {
load(Some(self.root()), ROOM, image)
}
}
#[test]
fn a_stored_image_loads_as_its_exact_bytes_in_base64() {
let store = Store::new();
let image = store.load(&store.put(ROOM, PNG)).unwrap();
assert_eq!(image.media_type, "image/png");
assert_eq!(STANDARD.decode(&image.base64).unwrap(), PNG);
}
#[test]
fn the_four_formats_load_as_their_own_media_type() {
let store = Store::new();
for (bytes, expected) in [
(&b"\xff\xd8\xff\xe0 jpeg tail"[..], "image/jpeg"),
(&b"GIF89a gif tail"[..], "image/gif"),
(&b"GIF87a gif tail"[..], "image/gif"),
(&b"RIFF\x10\0\0\0WEBPVP8 tail"[..], "image/webp"),
] {
assert_eq!(store.load(&store.put(ROOM, bytes)).unwrap().media_type, expected);
}
}
#[test]
fn a_file_without_an_images_magic_bytes_is_refused_whatever_it_claims() {
let store = Store::new();
let mut image = store.put(ROOM, b"<html><script>alert(1)</script></html>");
image.mime = "image/png".into();
assert_eq!(store.load(&image), Err(Skip::NotAnImage));
// RIFF without WEBP is a wav, not an image.
assert_eq!(store.load(&store.put(ROOM, b"RIFF\x10\0\0\0WAVEfmt ")), Err(Skip::NotAnImage));
}
#[test]
fn a_declared_type_that_is_not_the_sniffed_one_is_refused() {
let store = Store::new();
let mut image = store.put(ROOM, PNG);
image.mime = "image/jpeg".into();
assert_eq!(store.load(&image), Err(Skip::MimeMismatch));
}
#[test]
fn an_absolute_path_is_refused_even_when_it_names_the_real_file() {
let store = Store::new();
let mut image = store.put(ROOM, PNG);
image.path = store.root().join(&image.path).to_string_lossy().into_owned();
assert_eq!(store.load(&image), Err(Skip::BadReference));
}
#[test]
fn traversal_is_refused_and_a_file_outside_the_store_is_never_read() {
let store = Store::new();
let outside = tempfile::tempdir().unwrap();
let sha256 = hex(&Sha256::digest(PNG));
std::fs::write(outside.path().join(&sha256), PNG).unwrap();
let real = store.put(ROOM, PNG);
let room = room_dir(ROOM);
let up = format!("../{}", outside.path().file_name().unwrap().to_string_lossy());
for path in [
format!("{room}/../{room}/{sha256}"),
format!("{room}/./{sha256}"),
format!("../{room}/{sha256}"),
format!("{up}/{sha256}"),
format!("{room}/{up}/{sha256}"),
format!("{room}/{sha256}/"),
format!("{room}//{sha256}"),
format!("{room}/sub/{sha256}"),
format!("{room}\\{sha256}"),
format!("{room}/{sha256}\0"),
sha256.clone(),
String::new(),
] {
assert_eq!(store.load(&ChatImage { path: path.clone(), ..real.clone() }), Err(Skip::BadReference), "{path:?}");
}
}
#[test]
fn another_rooms_image_is_not_this_rooms() {
let store = Store::new();
let image = store.put("!other:gamecult.test", PNG);
assert_eq!(store.load(&image), Err(Skip::OutsideStore));
}
#[test]
fn a_symlink_on_the_way_is_refused_though_its_target_is_a_good_image() {
let store = Store::new();
let elsewhere = tempfile::tempdir().unwrap();
let sha256 = hex(&Sha256::digest(PNG));
std::fs::write(elsewhere.path().join("target"), PNG).unwrap();
let room = room_dir(ROOM);
std::fs::create_dir_all(store.root().join(&room)).unwrap();
symlink(elsewhere.path().join("target"), store.root().join(&room).join(&sha256)).unwrap();
let image = ChatImage { path: format!("{room}/{sha256}"), mime: "image/png".into(), size: PNG.len() as u64, sha256: sha256.clone() };
assert_eq!(store.load(&image), Err(Skip::OutsideStore), "a symlinked file");
let linked = Store::new();
symlink(elsewhere.path(), linked.root().join(&room)).unwrap();
std::fs::write(elsewhere.path().join(&sha256), PNG).unwrap();
assert_eq!(linked.load(&image), Err(Skip::OutsideStore), "a symlinked room directory");
}
#[test]
fn a_name_that_is_not_the_declared_hash_is_refused() {
let store = Store::new();
let real = store.put(ROOM, PNG);
let other = "0".repeat(64);
for name in [other.clone(), format!("{}.png.exe", real.sha256), format!("{}.", real.sha256), format!("{}.toolong", real.sha256), format!("x{}", real.sha256)] {
let image = ChatImage { path: format!("{}/{name}", room_dir(ROOM)), ..real.clone() };
assert_eq!(store.load(&image), Err(Skip::BadReference), "{name}");
}
for sha in ["short", &real.sha256.to_uppercase()] {
assert_eq!(store.load(&ChatImage { sha256: sha.into(), ..real.clone() }), Err(Skip::BadReference), "{sha}");
}
}
#[test]
fn an_extension_is_allowed_after_the_hash() {
let store = Store::new();
let real = store.put(ROOM, PNG);
let named = format!("{}.png", real.sha256);
std::fs::rename(store.root().join(&real.path), store.root().join(room_dir(ROOM)).join(&named)).unwrap();
assert!(store.load(&ChatImage { path: format!("{}/{named}", room_dir(ROOM)), ..real }).is_ok());
}
#[test]
fn over_five_mib_is_refused_declared_or_actual_and_before_it_is_read() {
let store = Store::new();
let mut big = PNG.to_vec();
big.resize(MAX_BYTES as usize + 1, 0);
let honest = store.put(ROOM, &big);
assert_eq!(store.load(&honest), Err(Skip::TooLarge), "a file over the cap, honestly declared");
let small = store.put(ROOM, PNG);
assert_eq!(store.load(&ChatImage { size: MAX_BYTES + 1, ..small.clone() }), Err(Skip::TooLarge), "a declaration over the cap");
let lying = ChatImage { size: 10, ..honest };
assert_eq!(store.load(&lying), Err(Skip::TooLarge), "an oversize file declared small is refused on its length, unread");
let mut exact = PNG.to_vec();
exact.resize(MAX_BYTES as usize, 0);
assert!(store.load(&store.put(ROOM, &exact)).is_ok(), "exactly the cap passes");
}
#[test]
fn the_declared_size_and_hash_must_be_the_files() {
let store = Store::new();
let image = store.put(ROOM, PNG);
assert_eq!(store.load(&ChatImage { size: image.size + 1, ..image.clone() }), Err(Skip::SizeMismatch));
// Same length, other bytes: swapped after the mouth stored it.
let mut swapped = PNG.to_vec();
*swapped.last_mut().unwrap() ^= 1;
std::fs::write(store.root().join(&image.path), &swapped).unwrap();
assert_eq!(store.load(&image), Err(Skip::HashMismatch));
}
#[test]
fn no_store_and_no_file_are_skips_not_panics() {
let store = Store::new();
let image = store.put(ROOM, PNG);
assert_eq!(load(None, ROOM, &image), Err(Skip::NoStore));
std::fs::remove_file(store.root().join(&image.path)).unwrap();
assert_eq!(store.load(&image), Err(Skip::Missing));
assert_eq!(load(Some(&store.root().join("absent")), ROOM, &image), Err(Skip::Missing));
}
#[test]
fn a_directory_where_the_file_should_be_is_refused() {
let store = Store::new();
let sha256 = hex(&Sha256::digest(PNG));
let room = room_dir(ROOM);
std::fs::create_dir_all(store.root().join(&room).join(&sha256)).unwrap();
let image = ChatImage { path: format!("{room}/{sha256}"), mime: "image/png".into(), size: PNG.len() as u64, sha256 };
assert_eq!(store.load(&image), Err(Skip::Missing));
}
#[test]
fn a_fifo_where_the_file_should_be_is_refused_without_opening_it() {
let store = Store::new();
let sha256 = hex(&Sha256::digest(PNG));
let room = room_dir(ROOM);
std::fs::create_dir_all(store.root().join(&room)).unwrap();
assert!(std::process::Command::new("mkfifo").arg(store.root().join(&room).join(&sha256)).status().unwrap().success());
let image = ChatImage { path: format!("{room}/{sha256}"), mime: "image/png".into(), size: PNG.len() as u64, sha256 };
let (tx, rx) = std::sync::mpsc::channel();
let root = store.root().to_path_buf();
std::thread::spawn(move || tx.send(load(Some(&root), ROOM, &image)).ok());
assert_eq!(rx.recv_timeout(std::time::Duration::from_secs(5)).expect("load returned without waiting on the FIFO"), Err(Skip::Missing));
}

View file

@ -29,7 +29,7 @@ use serde::de::DeserializeOwned;
use sha2::{Digest, Sha256};
use crate::chat_client::{
ACK, Chat, ChatAck, ChatDelivery, ChatPending, ChatPendingRequest, ChatRefusal, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, REFUSAL, SEND_RECEIPT,
ACK, Chat, ChatAck, ChatDelivery, ChatPending, ChatPendingRequest, ChatRefusal, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_V1, REFUSAL, SEND_RECEIPT,
SEND_REQUEST, decode, encode,
};
@ -160,7 +160,7 @@ impl RudpChat {
impl Chat for RudpChat {
fn pending(&self, after_seq: u64) -> Result<Vec<ChatDelivery>> {
let request = ChatPendingRequest { persona: self.persona.clone(), after_seq };
let kinds: &[&str] = &[PENDING, REFUSAL];
let kinds: &[&str] = &[PENDING, PENDING_V1, REFUSAL];
let (kind, payload) = self.exchange(PENDING_REQUEST, &self.persona, &request, Some((kinds, self.persona.as_str())))?.ok_or_else(|| anyhow!("no answer payload"))?;
if kind == REFUSAL {
let refusal: ChatRefusal = decode(&payload)?;

View file

@ -152,6 +152,7 @@ fn delivery(seq: u64, sender: &str, text: &str, mentions: &[&str]) -> ChatDelive
context: vec![],
mentions: mentions.iter().map(|id| (*id).to_string()).collect(),
is_direct: false,
image: None,
}
}

View file

@ -36,6 +36,7 @@
mod address;
pub mod chat_client;
mod chat_image;
pub mod chat_rudp;
pub mod cli;
pub mod connector;

View file

@ -48,6 +48,11 @@
//! step per rank
//! (ruling `eureka-body:ruling:unaddressed-tell-stagger-ruled`), so the Minds read each other before they answer.
//!
//! A delivery may carry an image the mouth stored (`ChatDelivery::image`, a reference, never bytes). `attachment_lines` loads it from her
//! image store (`MIND_PERSONA_IMAGE_STORE`) through `chat_image::load`, which re-checks path, size, hash and magic bytes, and the line
//! to her session carries it as a base64 image block after its text, with one `attachment role=organ` line saying whether it is
//! attached or not shown and why.
//!
//! The reply is decoded, never scanned: text that is not one `PersonaReply` is a refused
//! result. Chat, persona state and trace are ports; the chat transport is `chat_rudp::RudpChat`,
//! wired by the binary. Inside her sandbox her session has a shell and the web (`--tools`);
@ -78,6 +83,7 @@ use serde_json::{Value, json};
use uuid::Uuid;
use crate::address::{Addressee, Name, addresses, localpart, project_names};
use crate::chat_image::{self, Image, MAX_PER_LINE};
use crate::chat_client::{Chat, ChatDelivery, ChatSendRequest, ContextLine, Refusal, SendBasis};
use crate::connector::render_plan;
use crate::control::{ControlSource, Effective, HeldReason, derive};
@ -131,6 +137,8 @@ pub const ENV_RUN_CAP: &str = "MIND_PERSONA_RUN_CAP_S";
pub const ENV_CONFIG_SEED: &str = "MIND_PERSONA_CONFIG_SEED";
/// Colon-separated absolute paths of the doctrine files whose text leads every process's instructions, in order.
pub const ENV_DOCTRINE: &str = "MIND_PERSONA_DOCTRINE";
/// The absolute path of her image store (`Config::image_store`).
pub const ENV_IMAGE_STORE: &str = "MIND_PERSONA_IMAGE_STORE";
/// The most doctrine text, in bytes, one process's instructions may carry.
pub const DOCTRINE_MAX_BYTES: usize = 96 * 1024;
/// How many of her latest runs a turn's input lists.
@ -185,6 +193,8 @@ pub struct Config {
pub model: String,
/// Doctrine files, read at every process start and put ahead of her persona, in this order. Empty is legal.
pub doctrine: Vec<PathBuf>,
/// Her image store: where the mouth keeps the images it fetched for her, one directory per room. None shows no image.
pub image_store: Option<PathBuf>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@ -235,7 +245,7 @@ impl Config {
if items.iter().any(String::is_empty) { Err(anyhow!("{usage}")) } else { Ok(items) }
};
let rooms = list(rooms)?;
let config = Self { rooms, operators: list(operators)?, minds: list(minds)?, own_sender: own.ok_or_else(|| anyhow!("{usage}"))?, model: String::new(), doctrine: Vec::new() };
let config = Self { rooms, operators: list(operators)?, minds: list(minds)?, own_sender: own.ok_or_else(|| anyhow!("{usage}"))?, model: String::new(), doctrine: Vec::new(), image_store: None };
Ok((config, rest))
}
@ -262,6 +272,12 @@ impl Config {
bail!("{ENV_DOCTRINE} names a path that is not absolute");
}
}
if let Some(path) = lookup(ENV_IMAGE_STORE).filter(|value| !value.is_empty()) {
if !Path::new(&path).is_absolute() {
bail!("{ENV_IMAGE_STORE} is not an absolute path");
}
config.image_store = Some(PathBuf::from(path));
}
Ok((config, rest))
}
@ -1110,6 +1126,8 @@ struct Rendered {
shown: Shown,
told: Vec<LineKey>,
replies: Vec<HeardReply>,
/// The images its attachment lines say are attached, in their order.
images: Vec<Image>,
}
/// One result of her Self session, as her mind holds it: what the reply line shows and what resolving it needs.
@ -1196,6 +1214,8 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> {
told_replies: BTreeSet<String>,
/// Her in-force replies read and not yet told, oldest first.
waiting_replies: Vec<HeardReply>,
/// The images of the line last written, until `drive` takes them into its user message.
images: Vec<Image>,
/// The replies the CLI took in since the last result: that result is what answers them.
echoed_replies: Vec<HeardReply>,
/// Replies she answered whose resolution the mind has not yet shown took, with the follow-up her answer started: retried at each
@ -1255,6 +1275,7 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> {
first_seen: BTreeMap::new(),
told_replies: BTreeSet::new(),
waiting_replies: Vec::new(),
images: Vec::new(),
echoed_replies: Vec::new(),
answered_replies: Vec::new(),
reply_read_due_ms: 0,
@ -1297,8 +1318,9 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> {
let people = people_she_may_name(self.config, deliveries, &earlier, &shown);
let started_self_today = started_self_today(organs.mind, &instance, organs.clock.now()).unwrap_or(true);
let (heard, replies) = self.take_reply_lines();
let input = render_input(self.config, started_self_today, &format!("{heard}{}", message_lines(self.config, &self.addressee, &self.addressing, deliveries, &earlier)), note, &history, &goals.text, &wear, &recall::render(&recalled), &format!("{}\n\n{tempo}", self.plan), &people);
Ok(Rendered { input, shown: Shown { people, open: goals.open }, told: lines_told(deliveries, &earlier), replies })
let (attached, images) = attachment_lines(self.config, deliveries);
let input = render_input(self.config, started_self_today, &format!("{heard}{}{attached}", message_lines(self.config, &self.addressee, &self.addressing, deliveries, &earlier)), note, &history, &goals.text, &wear, &recall::render(&recalled), &format!("{}\n\n{tempo}", self.plan), &people);
Ok(Rendered { input, shown: Shown { people, open: goals.open }, told: lines_told(deliveries, &earlier), replies, images })
}
/// A later line: the messages, a refused batch's note before them, the organ's lines (her Self's replies, the notices of her repos), and
@ -1306,13 +1328,15 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> {
/// The organ's lines come first: they are not messages and answer to no delivery.
fn tell_line(&self, deliveries: &[ChatDelivery], note: Option<&str>, notices: &str) -> Rendered {
let earlier = self.earlier(deliveries);
let lines = format!("{notices}{}", message_lines(self.config, &self.addressee, &self.addressing, deliveries, &earlier));
let (attached, images) = attachment_lines(self.config, deliveries);
let lines = format!("{notices}{}{attached}", message_lines(self.config, &self.addressee, &self.addressing, deliveries, &earlier));
let input = note.map_or_else(|| lines.clone(), |note| format!("{note}\n\n{lines}"));
Rendered { input, shown: Shown { people: people_she_may_name(self.config, deliveries, &earlier, &[]), open: Vec::new() }, told: lines_told(deliveries, &earlier), replies: Vec::new() }
Rendered { input, shown: Shown { people: people_she_may_name(self.config, deliveries, &earlier, &[]), open: Vec::new() }, told: lines_told(deliveries, &earlier), replies: Vec::new(), images }
}
/// Notes what a rendered line carries, as it goes to the child.
fn write(&mut self, batch: Vec<ChatDelivery>, Rendered { input, shown, told, replies }: Rendered, refusals: u32) -> String {
fn write(&mut self, batch: Vec<ChatDelivery>, Rendered { input, shown, told, replies, images }: Rendered, refusals: u32) -> String {
self.images = images;
if let Some(through) = batch.iter().map(|delivery| delivery.seq).max() {
self.told_through = self.told_through.max(through);
}
@ -1526,6 +1550,10 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
Ok(self.write(batch, line, 0))
}
fn images(&mut self) -> Vec<Image> {
std::mem::take(&mut self.images)
}
fn flush_text(&mut self, why: Why) -> String {
if self.written.is_none() {
self.written = Some(Written { batch: Vec::new(), shown: Shown::default(), refusals: 0, revision: false, replies: Vec::new() });
@ -3148,6 +3176,27 @@ fn render_input(config: &Config, started_self_today: bool, messages: &str, note:
out
}
/// One line per delivery that carries an image, after the messages: the organ's word whether it is attached to the line (the
/// next image block, in order) or not shown, and why. Every reason is the organ's fixed words, never the sender's. The images
/// are loaded here and nowhere else (`chat_image::load`).
fn attachment_lines(config: &Config, deliveries: &[ChatDelivery]) -> (String, Vec<Image>) {
let mut out = String::new();
let mut images = Vec::new();
for delivery in deliveries {
let Some(reference) = &delivery.image else { continue };
let loaded = if images.len() >= MAX_PER_LINE { Err(chat_image::Skip::TooMany) } else { chat_image::load(config.image_store.as_deref(), &delivery.room_id, reference) };
let status = match loaded {
Ok(image) => {
images.push(image);
format!("status=attached image={}", images.len())
}
Err(skip) => format!("status=not_shown reason={}", quoted(skip.reason())),
};
out.push_str(&format!("attachment role=organ room={} event={} at={} {status}\n", quoted(&delivery.room_alias), quoted(&delivery.event_id), at_text(delivery.at_unix_ms)));
}
(out, images)
}
/// A batch as framed lines: each message, after the `earlier` lines it brings.
fn message_lines(config: &Config, own: &Addressee, addressing: &[Addressee], deliveries: &[ChatDelivery], earlier: &[Vec<&ContextLine>]) -> String {
let mut out = String::new();

View file

@ -20,7 +20,8 @@ use serde_json::json;
use tempfile::TempDir;
use super::*;
use crate::chat_client::{ChatSendReceipt, ContextLine, Refusal};
use crate::chat_client::{ChatImage, ChatSendReceipt, ContextLine, Refusal};
use base64::Engine;
use eureka_pipeline::OrgRepo;
use crate::control::{ControlWriter, FileSource, SwarmWriter};
use crate::lease_client::{LeaseRefusalKind, TokenFd};
@ -156,6 +157,8 @@ struct Told {
instructions: String,
input: String,
output_schema_json: String,
/// The image blocks of the line, as (media type, base64 data).
images: Vec<(String, String)>,
}
/// The spawner: every child it starts shares these, so the tests see across processes.
@ -235,9 +238,19 @@ impl Child for Replying {
return;
}
let written: serde_json::Value = serde_json::from_str(line).unwrap();
let input = written["message"]["content"].as_str().unwrap().to_string();
let content = &written["message"]["content"];
let (input, images) = match content.as_str() {
Some(text) => (text.to_string(), Vec::new()),
None => {
let blocks = content.as_array().expect("content is text or blocks");
assert_eq!(blocks[0]["type"], "text", "the text block comes first");
let images = blocks[1..].iter().map(|block| (block["source"]["media_type"].as_str().unwrap().to_string(), block["source"]["data"].as_str().unwrap().to_string())).collect();
assert!(blocks[1..].iter().all(|block| block["type"] == "image" && block["source"]["type"] == "base64"));
(blocks[0]["text"].as_str().unwrap().to_string(), images)
}
};
self.log.borrow_mut().push("turn".into());
self.requests.borrow_mut().push(Told { instructions: self.instructions.clone(), input, output_schema_json: self.schema.clone() });
self.requests.borrow_mut().push(Told { instructions: self.instructions.clone(), input, output_schema_json: self.schema.clone(), images });
if let Some(during) = &*self.during.borrow() {
during();
}
@ -346,6 +359,7 @@ fn config() -> Config {
own_sender: "@eureka:chat".into(),
model: "claude-test-model".into(),
doctrine: vec![],
image_store: None,
}
}
@ -375,6 +389,7 @@ fn delivery_of(seq: u64, room: &str, sender: &str, text: &str, event_id: &str) -
context: vec![],
mentions: vec![],
is_direct: false,
image: None,
}
}
@ -1618,7 +1633,7 @@ fn the_persona_flags_come_from_the_environment_when_the_command_line_lacks_them(
let (config, rest) = Config::from_args_and_env(&args("--instance eureka"), env, "usage").unwrap();
assert_eq!(
config,
Config { rooms: vec!["swarm".into(), "usage".into()], operators: vec!["@op:chat".into(), "@op2:chat".into()], minds: vec!["@libby:chat".into(), "@aqua:chat".into()], own_sender: "@eureka:chat".into(), model: "claude-test-model".into(), doctrine: vec![] }
Config { rooms: vec!["swarm".into(), "usage".into()], operators: vec!["@op:chat".into(), "@op2:chat".into()], minds: vec!["@libby:chat".into(), "@aqua:chat".into()], own_sender: "@eureka:chat".into(), model: "claude-test-model".into(), doctrine: vec![], image_store: None }
);
assert_eq!(rest, ["--instance", "eureka"]);
// A flag wins over its variable, one at a time.
@ -6907,3 +6922,112 @@ fn her_own_row_is_in_the_addressing_exactly_once_whatever_the_table_holds() {
assert_eq!(ids(vec![crate::heartbeat::Addressing { minds: vec![eureka.clone()] }]), ["@eureka:chat"]);
assert_eq!(ids(vec![crate::heartbeat::Addressing { minds: vec![libby, eureka, aqua] }]), ["@libby:chat", "@eureka:chat", "@aqua:chat"], "a table with her keeps its own order and row");
}
/// An image stored the way the mouth stores it, for `room_id` under `store`: returns the reference.
fn stored_image(store: &std::path::Path, room_id: &str, bytes: &[u8]) -> ChatImage {
use sha2::{Digest, Sha256};
let sha256 = crate::chat_image::hex(&Sha256::digest(bytes));
let room = crate::chat_image::room_dir(room_id);
std::fs::create_dir_all(store.join(&room)).unwrap();
std::fs::write(store.join(&room).join(&sha256), bytes).unwrap();
ChatImage { path: format!("{room}/{sha256}"), mime: "image/png".into(), size: bytes.len() as u64, sha256 }
}
const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\nlast-bytes";
fn with_image(delivery: ChatDelivery, image: ChatImage) -> ChatDelivery {
ChatDelivery { image: Some(image), ..delivery }
}
#[test]
fn a_delivered_image_reaches_the_model_as_an_image_block_beside_its_text() {
let store = tempfile::tempdir().unwrap();
let sent = with_image(delivery(1, "swarm", "@libby:chat", "[image filename=\"s.png\"]"), stored_image(store.path(), "!swarm:chat", PNG_BYTES));
let rig = Rig::new(vec![ChatDelivery { room_id: "!swarm:chat".into(), ..sent }], &[says(&[("swarm", "seen")])]);
rig.clock.step.set(0);
assert_eq!(failure_of(rig.go_with(&Config { image_store: Some(store.path().to_path_buf()), ..config() })), None);
let requests = rig.model.requests.borrow();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].images, vec![("image/png".to_string(), base64::engine::general_purpose::STANDARD.encode(PNG_BYTES))]);
assert!(requests[0].input.contains("attachment role=organ room=\"swarm\" ") && requests[0].input.contains("status=attached image=1"), "{}", requests[0].input);
}
#[test]
fn an_image_that_does_not_resolve_is_one_attachment_line_with_its_reason_and_no_block() {
let store = tempfile::tempdir().unwrap();
let elsewhere = tempfile::tempdir().unwrap();
// Stored for another room: the delivery's room has no such image.
let other = stored_image(elsewhere.path(), "!another:chat", PNG_BYTES);
for (config, reference, reason) in [
(Config { image_store: None, ..config() }, stored_image(store.path(), "!swarm:chat", PNG_BYTES), "this Persona has no image store"),
(Config { image_store: Some(store.path().to_path_buf()), ..config() }, other, "the image reference leaves this room's store"),
] {
let rig = Rig::new(vec![ChatDelivery { room_id: "!swarm:chat".into(), ..with_image(delivery(1, "swarm", "@libby:chat", "look"), reference) }], &[says(&[("swarm", "ok")])]);
rig.clock.step.set(0);
assert_eq!(failure_of(rig.go_with(&config)), None);
let requests = rig.model.requests.borrow();
assert!(requests[0].images.is_empty());
assert!(requests[0].input.contains(&format!("status=not_shown reason=\"{reason}\"")), "{}", requests[0].input);
}
}
#[test]
fn a_message_with_no_image_has_no_attachment_line_and_a_plain_string_line() {
let rig = Rig::new(vec![delivery(1, "swarm", "@libby:chat", "just words")], &[says(&[("swarm", "ok")])]);
rig.clock.step.set(0);
assert_eq!(failure_of(rig.go()), None);
let requests = rig.model.requests.borrow();
assert!(requests[0].images.is_empty() && !requests[0].input.contains("attachment role=organ"));
}
#[test]
fn the_attachment_lines_cap_at_four_images_and_say_so() {
let store = tempfile::tempdir().unwrap();
let make = |at: u64| ChatDelivery { room_id: "!swarm:chat".into(), ..with_image(delivery(at, "swarm", "@libby:chat", "pic"), stored_image(store.path(), "!swarm:chat", &[PNG_BYTES, &at.to_be_bytes()].concat())) };
let config = Config { image_store: Some(store.path().to_path_buf()), ..config() };
let (lines, images) = attachment_lines(&config, &(1..=6).map(make).collect::<Vec<_>>());
assert_eq!(images.len(), 4);
assert_eq!(lines.matches("status=attached").count(), 4);
assert_eq!(lines.matches("reason=\"too many images in one line\"").count(), 2, "{lines}");
}
#[test]
fn the_image_store_comes_from_the_environment_and_must_be_absolute() {
let args = ["--rooms", "swarm", "--operator", "@op:chat", "--minds", "@eureka:chat", "--self-sender", "@eureka:chat"].map(String::from).to_vec();
let lookup = |store: &'static str| move |name: &str| match name {
ENV_MODEL => Some("m".to_string()),
ENV_IMAGE_STORE => Some(store.to_string()),
_ => None,
};
let (config, _) = Config::from_args_and_env(&args, lookup("/srv/images"), "usage").unwrap();
assert_eq!(config.image_store, Some(PathBuf::from("/srv/images")));
assert!(Config::from_args_and_env(&args, lookup("images"), "usage").is_err());
let (config, _) = Config::from_args_and_env(&args, lookup(""), "usage").unwrap();
assert_eq!(config.image_store, None);
}
#[test]
fn an_images_line_is_the_only_line_that_carries_it() {
let store = tempfile::tempdir().unwrap();
let first = ChatDelivery { room_id: "!swarm:chat".into(), ..with_image(delivery(1, "operator", "@op:chat", "look"), stored_image(store.path(), "!swarm:chat", PNG_BYTES)) };
let rig = Rig::new(vec![first], &[says(&[("operator", "a")]), says(&[("operator", "b")])]);
tell_one_more(&rig);
assert_eq!(failure_of(rig.go_with(&Config { image_store: Some(store.path().to_path_buf()), ..config() })), None);
let requests = rig.model.requests.borrow();
assert_eq!(requests.len(), 2, "{:?}", requests.iter().map(|told| &told.input).collect::<Vec<_>>());
assert_eq!(requests[0].images.len(), 1);
assert!(requests[1].images.is_empty(), "the next line does not repeat the image");
}
#[test]
fn a_flush_after_an_image_line_carries_no_image() {
let store = tempfile::tempdir().unwrap();
let first = ChatDelivery { room_id: "!operator:chat".into(), ..with_image(operator_says(1, "look"), stored_image(store.path(), "!operator:chat", PNG_BYTES)) };
let rig = Rig::new(vec![first], &[says(&[("operator", "x")]), says(&[])]);
rig.cap.set(u64::try_from(ROTATE_MARGIN_S).unwrap());
assert_eq!(failure_of(rig.go_with(&Config { image_store: Some(store.path().to_path_buf()), ..config() })), None);
let requests = rig.model.requests.borrow();
let flush = requests.iter().find(|told| told.input.contains("This session ends soon")).expect("a flush line");
assert!(flush.images.is_empty(), "the flush does not repeat the image");
assert_eq!(requests.iter().filter(|told| !told.images.is_empty()).count(), 1);
}

View file

@ -376,6 +376,8 @@ pub fn read_usage(line: &str) -> Usage {
.and_then(|(_, rest)| rest.split_once("</task-id>"))
.map_or(Usage::Other, |(id, _)| Usage::NotificationTaken { id: id.to_string() }),
Some(_) => Usage::Echo,
// A line with images replays as blocks: a text block first is still our line.
None if event.get("message").and_then(|message| message.get("content")).and_then(Value::as_array).and_then(|blocks| blocks.first()).and_then(|block| block.get("type")).and_then(Value::as_str) == Some("text") => Usage::Echo,
None => Usage::Other,
}
}

View file

@ -11,10 +11,11 @@ use std::time::Duration as StdDuration;
use anyhow::Result;
use chrono::Duration as ChronoDuration;
use serde_json::json;
use serde_json::{Value, json};
use super::decide::{Action, Event, State, Why, decide, next_deadline};
use super::{Child, Read, SelfRunOutcome, Usage, read_usage, write_result, writes_hour_cache};
use crate::chat_image::Image;
use crate::launch::Clock;
/// What one poll of an inbox found.
@ -40,6 +41,10 @@ pub(crate) trait SessionInbox {
/// The text of what waits, which is thereby told. `switch`: the session is idle and the text is another conversation's, the oldest waiting; else it is the served conversation's.
fn tell(&mut self, switch: bool) -> Result<String>;
fn flush_text(&mut self, why: Why) -> String;
/// The images of the line `start_line` or `tell` just returned, taken once: they ride beside its text in the same user message.
fn images(&mut self) -> Vec<Image> {
Vec::new()
}
/// Every line of the child's stream, before it is read.
fn line(&mut self, _line: &str) {}
/// The CLI took one of our lines into a request.
@ -50,10 +55,21 @@ pub(crate) trait SessionInbox {
}
}
/// One user message for the CLI's stream-json input. With no image its content is the text, as ever; with images it is a text block
/// then one base64 image block each.
pub(super) fn user_line(text: &str, images: Vec<Image>) -> String {
if images.is_empty() {
return json!({"type": "user", "message": {"role": "user", "content": text}}).to_string();
}
let blocks: Vec<Value> = std::iter::once(json!({"type": "text", "text": text}))
.chain(images.into_iter().map(|image| json!({"type": "image", "source": {"type": "base64", "media_type": image.media_type, "data": image.base64}})))
.collect();
json!({"type": "user", "message": {"role": "user", "content": blocks}}).to_string()
}
/// The loop: facts in, one clock read per fact, actions out. The one writer into the session.
pub(crate) fn drive(clock: &dyn Clock, mut child: Box<dyn Child>, inbox: &mut dyn SessionInbox, run_cap_s: u64, out: &mut dyn Write) -> Result<SelfRunOutcome> {
let user_line = |text: &str| json!({"type": "user", "message": {"role": "user", "content": text}}).to_string();
child.write_line(&user_line(&inbox.start_line()?));
child.write_line(&user_line(&inbox.start_line()?, inbox.images()));
let poll_every = ChronoDuration::seconds(inbox.poll_every_s());
let longest = StdDuration::from_secs(u64::try_from(inbox.poll_every_s()).unwrap_or(1));
@ -128,7 +144,7 @@ pub(crate) fn drive(clock: &dyn Clock, mut child: Box<dyn Child>, inbox: &mut dy
if let Some(why) = *flush {
parts.push(inbox.flush_text(why));
}
child.write_line(&user_line(&parts.join("\n\n")));
child.write_line(&user_line(&parts.join("\n\n"), inbox.images()));
}
Action::CloseStdin => {
child.close_input();

View file

@ -1911,3 +1911,32 @@ mod sandbox {
assert!(!seen.contains("canary") && !seen.contains("UNLISTED"), "the child's environment is the allow-list: {seen}");
}
}
#[test]
fn a_user_line_with_images_is_a_text_block_then_one_base64_image_block_each_and_without_them_is_the_bare_text() {
use super::session::user_line;
use crate::chat_image::Image;
let bare: serde_json::Value = serde_json::from_str(&user_line("hello", vec![])).unwrap();
assert_eq!(bare, json!({"type": "user", "message": {"role": "user", "content": "hello"}}));
let images = vec![Image { media_type: "image/png", base64: "QUJD".into() }, Image { media_type: "image/jpeg", base64: "REVG".into() }];
let with: serde_json::Value = serde_json::from_str(&user_line("hello", images)).unwrap();
assert_eq!(
with,
json!({"type": "user", "message": {"role": "user", "content": [
{"type": "text", "text": "hello"},
{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "QUJD"}},
{"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": "REVG"}},
]}})
);
}
#[test]
fn a_replayed_line_with_images_is_still_our_echo_and_a_tool_result_is_not() {
let replayed = json!({"type": "user", "isReplay": true, "message": {"role": "user", "content": [
{"type": "text", "text": "hello"},
{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "QUJD"}},
]}}).to_string();
assert_eq!(read_usage(&replayed), Usage::Echo);
let tool_result = json!({"type": "user", "isReplay": true, "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t", "content": "x"}]}}).to_string();
assert_eq!(read_usage(&tool_result), Usage::Other);
}

View file

@ -13,11 +13,12 @@ use std::thread::JoinHandle;
use std::time::Duration;
use cultnet_rs::{CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions};
use mind_body::chat_client::{self, ACK, Chat, ChatAck, ChatPending, ChatPendingRequest, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, SEND_RECEIPT, SEND_REQUEST};
use mind_body::chat_client::{self, ACK, Chat, ChatAck, ChatPending, ChatPendingRequest, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_V1, SEND_RECEIPT, SEND_REQUEST};
use mind_body::chat_rudp::{RudpChat, connection_id};
const ACK_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack");
const PENDING_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack");
const PENDING_V1_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack");
const PENDING_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack");
const PENDING_REQUEST_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack");
const SEND_RECEIPT_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack");
const SEND_REQUEST_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.nil-basis.msgpack");
@ -65,8 +66,17 @@ fn mouth(key: &str) -> Mouth {
mouth_at(key, "127.0.0.1:0")
}
/// A mouth that has no image field yet: it answers a pending request in the v1 kind.
fn legacy_mouth(key: &str) -> Mouth {
serve(key, "127.0.0.1:0", PENDING_V1, PENDING_V1_BYTES)
}
/// The same mouth on a chosen address: a restart binds the address the old one held.
fn mouth_at(key: &str, bind: &str) -> Mouth {
serve(key, bind, PENDING, PENDING_BYTES)
}
fn serve(key: &str, bind: &str, pending_kind: &'static str, pending_bytes: &'static [u8]) -> Mouth {
let socket = UdpSocket::bind(bind).unwrap();
socket.set_read_timeout(Some(Duration::from_millis(20))).unwrap();
let address = socket.local_addr().unwrap();
@ -84,7 +94,7 @@ fn mouth_at(key: &str, bind: &str) -> Mouth {
log.lock().unwrap().push((document.schema_id.clone(), document.payload.clone()));
let for_eureka = |persona: &str| persona == "eureka";
let reply = match document.schema_id.as_str() {
PENDING_REQUEST if for_eureka(&chat_client::decode::<ChatPendingRequest>(&document.payload).unwrap().persona) => Some(answer(PENDING, "eureka", PENDING_BYTES)),
PENDING_REQUEST if for_eureka(&chat_client::decode::<ChatPendingRequest>(&document.payload).unwrap().persona) => Some(answer(pending_kind, "eureka", pending_bytes)),
SEND_REQUEST if for_eureka(&chat_client::decode::<ChatSendRequest>(&document.payload).unwrap().persona) => Some(answer(SEND_RECEIPT, &document.record_key, SEND_RECEIPT_BYTES)),
_ => None,
};
@ -177,3 +187,15 @@ fn a_send_that_timed_out_is_not_repeated_by_the_client() {
let sends: Vec<_> = mouth.seen.lock().unwrap().iter().filter(|(kind, _)| kind == SEND_REQUEST).map(|(_, payload)| chat_client::decode::<ChatSendRequest>(payload).unwrap().request_id).collect();
assert_eq!(sends, ["r-1"], "the send reached the mouth once; a repeat is the caller's, under the same request id");
}
#[test]
fn a_v1_mouths_pending_answer_still_reads_with_no_image_and_a_v2_one_carries_it() {
let legacy = legacy_mouth(KEY);
let old = client(&legacy, KEY, "eureka", Duration::from_millis(1500)).pending(0).unwrap();
assert_eq!(old.len(), 1);
assert!(old[0].image.is_none());
let current = mouth(KEY);
let new = client(&current, KEY, "eureka", Duration::from_millis(1500)).pending(0).unwrap();
assert_eq!(new[0].image.as_ref().map(|image| (image.mime.as_str(), image.size)), Some(("image/png", 2048)));
assert_eq!(new[0].text, old[0].text, "the same message, with its attachment");
}

View file

@ -18,6 +18,7 @@ mind_body::chat_client: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub e
mind_body::chat_client: const ACK : & str
mind_body::chat_client: const PENDING : & str
mind_body::chat_client: const PENDING_REQUEST : & str
mind_body::chat_client: const PENDING_V1 : & str
mind_body::chat_client: const REFUSAL : & str
mind_body::chat_client: const SEND_RECEIPT : & str
mind_body::chat_client: const SEND_REQUEST : & str
@ -29,7 +30,8 @@ mind_body::chat_client: impl Serialize for Refusal
mind_body::chat_client: impl< 'de > Deserialize < 'de > for Refusal
mind_body::chat_client: pub trait Chat { fn pending (& self , after_seq : u64) -> Result < Vec < ChatDelivery > > ; fn send (& self , request : & ChatSendRequest) -> Result < ChatSendReceipt > ; fn ack (& self , through_seq : u64) -> Result < () > ; }
mind_body::chat_client: struct ChatAck # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, through_seq: u64]
mind_body::chat_client: struct ChatDelivery # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[seq: u64, room_alias: String, room_id: String, event_id: String, sender: String, sender_display: String, text: String, at_unix_ms: u64, context: Vec < ContextLine >, mentions: Vec < String >, is_direct: bool]
mind_body::chat_client: struct ChatDelivery # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[seq: u64, room_alias: String, room_id: String, event_id: String, sender: String, sender_display: String, text: String, at_unix_ms: u64, context: Vec < ContextLine >, mentions: Vec < String >, is_direct: bool, image: Option < ChatImage >]
mind_body::chat_client: struct ChatImage # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[path: String, mime: String, size: u64, sha256: String]
mind_body::chat_client: struct ChatPending # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, deliveries: Vec < ChatDelivery >]
mind_body::chat_client: struct ChatPendingRequest # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, after_seq: u64]
mind_body::chat_client: struct ChatRefusal # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, refusal: Refusal]
@ -37,6 +39,7 @@ mind_body::chat_client: struct ChatSendReceipt # [derive (Clone , Debug , Partia
mind_body::chat_client: struct ChatSendRequest # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[request_id: String, persona: String, room_alias: String, text: String, basis: Option < SendBasis >]
mind_body::chat_client: struct ContextLine # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[sender: String, text: String, at_unix_ms: u64]
mind_body::chat_client: struct SendBasis # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[seen_event_id: Option < String >, since_unix_ms: u64]
mind_body::chat_image: impl Skip
mind_body::chat_rudp: fn connection_id (key_file_text : & str) -> u32
mind_body::chat_rudp: impl Chat for RudpChat
mind_body::chat_rudp: impl Drop for RudpChat
@ -160,6 +163,7 @@ mind_body::persona: const AGENT : & str
mind_body::persona: const DOCTRINE_MAX_BYTES : usize
mind_body::persona: const ENV_CONFIG_SEED : & str
mind_body::persona: const ENV_DOCTRINE : & str
mind_body::persona: const ENV_IMAGE_STORE : & str
mind_body::persona: const ENV_MINDS : & str
mind_body::persona: const ENV_MODEL : & str
mind_body::persona: const ENV_OPERATOR : & str
@ -192,7 +196,7 @@ mind_body::persona: impl< C : ControlSource , S : ControlSource > SessionInbox f
mind_body::persona: pub trait PersonaState { fn instructions (& self) -> Result < String > ; }
mind_body::persona: pub trait Trace { fn record (& self , trace : & TurnTrace) -> Result < () > ; }
mind_body::persona: pub use model_connector :: PlanQuota ;
mind_body::persona: struct Config # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[rooms: Vec < String >, operators: Vec < String >, minds: Vec < String >, own_sender: String, model: String, doctrine: Vec < PathBuf >]
mind_body::persona: struct Config # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[rooms: Vec < String >, operators: Vec < String >, minds: Vec < String >, own_sender: String, model: String, doctrine: Vec < PathBuf >, image_store: Option < PathBuf >]
mind_body::persona: struct FileTrace pub-fields[]
mind_body::persona: struct HuginnPersona pub-fields[]
mind_body::persona: struct Organs< 'a > pub-fields[mind: & 'a dyn MindPort, launcher: & 'a dyn Launcher, clock: & 'a dyn Clock, host: & 'a str, chat: & 'a dyn Chat, persona: & 'a dyn PersonaState, trace: & 'a dyn Trace, heartbeat: Option < & 'a dyn HeartbeatView >, lender: & 'a dyn Lender, spawner: & 'a dyn Spawner, run_cap_s: u64, sandbox: & 'a Path, scratch: & 'a Path]