From e2c9638f27333436ca6dc5e342ab4ff17b25ca1f Mon Sep 17 00:00:00 2001 From: Eureka Date: Sat, 7 Nov 2026 20:52:30 +0000 Subject: [PATCH] mind-body: a delivered chat image reaches her session as an image block ChatDelivery gains tuple index 11, `image: Option` ([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 `/[.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 --- Cargo.lock | 1 + crates/mind-body/Cargo.toml | 1 + crates/mind-body/src/address/tests.rs | 1 + crates/mind-body/src/chat_client.rs | 18 +- crates/mind-body/src/chat_client/tests.rs | 60 ++++- crates/mind-body/src/chat_image.rs | 146 ++++++++++++ crates/mind-body/src/chat_image/tests.rs | 223 ++++++++++++++++++ crates/mind-body/src/chat_rudp.rs | 4 +- crates/mind-body/src/heartbeat/tests.rs | 1 + crates/mind-body/src/lib.rs | 1 + crates/mind-body/src/persona.rs | 61 ++++- crates/mind-body/src/persona/tests.rs | 132 ++++++++++- crates/mind-body/src/self_run.rs | 2 + crates/mind-body/src/self_run/session.rs | 24 +- crates/mind-body/src/self_run/tests.rs | 29 +++ crates/mind-body/tests/chat_rudp.rs | 28 ++- ...cult.persona.chat_pending.v2.image.msgpack | Bin 0 -> 399 bytes ...t.persona.chat_pending.v2.no-image.msgpack | Bin 0 -> 185 bytes crates/mind-body/tests/public_surface.txt | 8 +- 19 files changed, 712 insertions(+), 28 deletions(-) create mode 100644 crates/mind-body/src/chat_image.rs create mode 100644 crates/mind-body/src/chat_image/tests.rs create mode 100644 crates/mind-body/tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack create mode 100644 crates/mind-body/tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.no-image.msgpack diff --git a/Cargo.lock b/Cargo.lock index 7946b40..a3cfb26 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/crates/mind-body/Cargo.toml b/crates/mind-body/Cargo.toml index fcfb9d4..a8fdd9e 100644 --- a/crates/mind-body/Cargo.toml +++ b/crates/mind-body/Cargo.toml @@ -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 diff --git a/crates/mind-body/src/address/tests.rs b/crates/mind-body/src/address/tests.rs index 0159686..64ae998 100644 --- a/crates/mind-body/src/address/tests.rs +++ b/crates/mind-body/src/address/tests.rs @@ -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, diff --git a/crates/mind-body/src/chat_client.rs b/crates/mind-body/src/chat_client.rs index 83e04c7..662232c 100644 --- a/crates/mind-body/src/chat_client.rs +++ b/crates/mind-body/src/chat_client.rs @@ -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, /// 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, +} + +/// An image the mouth fetched and stored, by reference: never its bytes. The path is relative to the Persona's image +/// store, `/[.]`, 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)] diff --git a/crates/mind-body/src/chat_client/tests.rs b/crates/mind-body/src/chat_client/tests.rs index aa3ce22..34f1c00 100644 --- a/crates/mind-body/src/chat_client/tests.rs +++ b/crates/mind-body/src/chat_client/tests.rs @@ -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) -> 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, + } + + 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(); +} diff --git a/crates/mind-body/src/chat_image.rs b/crates/mind-body/src/chat_image.rs new file mode 100644 index 0000000..85fd217 --- /dev/null +++ b/crates/mind-body/src/chat_image.rs @@ -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, `//[.]`. 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 { + 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::>()[..] 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; diff --git a/crates/mind-body/src/chat_image/tests.rs b/crates/mind-body/src/chat_image/tests.rs new file mode 100644 index 0000000..17f1fe8 --- /dev/null +++ b/crates/mind-body/src/chat_image/tests.rs @@ -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 { + 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""); + 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)); +} diff --git a/crates/mind-body/src/chat_rudp.rs b/crates/mind-body/src/chat_rudp.rs index e4b673b..a15c09f 100644 --- a/crates/mind-body/src/chat_rudp.rs +++ b/crates/mind-body/src/chat_rudp.rs @@ -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> { 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)?; diff --git a/crates/mind-body/src/heartbeat/tests.rs b/crates/mind-body/src/heartbeat/tests.rs index 33767b2..a00947e 100644 --- a/crates/mind-body/src/heartbeat/tests.rs +++ b/crates/mind-body/src/heartbeat/tests.rs @@ -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, } } diff --git a/crates/mind-body/src/lib.rs b/crates/mind-body/src/lib.rs index a2166da..01dd16a 100644 --- a/crates/mind-body/src/lib.rs +++ b/crates/mind-body/src/lib.rs @@ -36,6 +36,7 @@ mod address; pub mod chat_client; +mod chat_image; pub mod chat_rudp; pub mod cli; pub mod connector; diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index 1b8b9c4..69835a6 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -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, + /// 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, } #[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, replies: Vec, + /// The images its attachment lines say are attached, in their order. + images: Vec, } /// 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, /// Her in-force replies read and not yet told, oldest first. waiting_replies: Vec, + /// The images of the line last written, until `drive` takes them into its user message. + images: Vec, /// The replies the CLI took in since the last result: that result is what answers them. echoed_replies: Vec, /// 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, Rendered { input, shown, told, replies }: Rendered, refusals: u32) -> String { + fn write(&mut self, batch: Vec, 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 SessionInbox for PersonaInbox<'_, C, S> Ok(self.write(batch, line, 0)) } + fn images(&mut self) -> Vec { + 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) { + 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(); diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index 25c24ee..1316beb 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/crates/mind-body/src/persona/tests.rs @@ -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::>()); + 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::>()); + 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); +} diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index f33628a..58b63f5 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -376,6 +376,8 @@ pub fn read_usage(line: &str) -> Usage { .and_then(|(_, rest)| rest.split_once("")) .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, } } diff --git a/crates/mind-body/src/self_run/session.rs b/crates/mind-body/src/self_run/session.rs index d7179a8..03b8c4f 100644 --- a/crates/mind-body/src/self_run/session.rs +++ b/crates/mind-body/src/self_run/session.rs @@ -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; 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 { + 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) -> String { + if images.is_empty() { + return json!({"type": "user", "message": {"role": "user", "content": text}}).to_string(); + } + let blocks: Vec = 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, inbox: &mut dyn SessionInbox, run_cap_s: u64, out: &mut dyn Write) -> Result { - 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, 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(); diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 18a6a86..a2819b0 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -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); +} diff --git a/crates/mind-body/tests/chat_rudp.rs b/crates/mind-body/tests/chat_rudp.rs index 7a5318a..8825ea3 100644 --- a/crates/mind-body/tests/chat_rudp.rs +++ b/crates/mind-body/tests/chat_rudp.rs @@ -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::(&document.payload).unwrap().persona) => Some(answer(PENDING, "eureka", PENDING_BYTES)), + PENDING_REQUEST if for_eureka(&chat_client::decode::(&document.payload).unwrap().persona) => Some(answer(pending_kind, "eureka", pending_bytes)), SEND_REQUEST if for_eureka(&chat_client::decode::(&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::(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(¤t, 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"); +} diff --git a/crates/mind-body/tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack new file mode 100644 index 0000000000000000000000000000000000000000..ef5f8a19b43cd51d0387b42d66d82fd4b419b017 GIT binary patch literal 399 zcmbQVEVZ;KH9K+Q9QJL>`30#(i6!|(R_TeksmY}|C3^Wq=^GUjlaf(H7OSKhZgR*? zElETXUg8TBS(A~PlasFiG4wnG1LG{+hK&ppCvS0p2%+d$o|;&clbKosRkv1R5>{oK u9P;wZQ0+QA literal 0 HcmV?d00001 diff --git a/crates/mind-body/tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.no-image.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.no-image.msgpack new file mode 100644 index 0000000000000000000000000000000000000000..ee79d3cc77e19cda546dfd93b2efe9f395a4470d GIT binary patch literal 185 zcmbQVEVZ;KH9K+Q9QJL>`30#(i6!|(R_TeksmY}|C3^Wq=^GUjlaf(H7OSKhZgR*? zElETXUg8TBS(A~PlasFiG4wnG1LG{+hK&ppCvS0p2%+d$o|;&clbKosRkv1R5>{oK N9P;wZQ0+Q=000qUN}&J% literal 0 HcmV?d00001 diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 84c1995..2b4966c 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -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]