diff --git a/crates/rhema-claw/Cargo.toml b/crates/rhema-claw/Cargo.toml index 002520f..4cb9549 100644 --- a/crates/rhema-claw/Cargo.toml +++ b/crates/rhema-claw/Cargo.toml @@ -73,6 +73,9 @@ tokio-stream = { workspace = true } rusqlite = { version = "0.40", features = ["bundled"] } # RFC3339 parsing for schedule_task's `at` time = { version = "0.3", default-features = false, features = ["parsing"] } +# Inbound chat images ride base64; the dispatch decodes them into the +# attachment store +base64 = "0.22" [dev-dependencies] tempfile = "3" diff --git a/crates/rhema-claw/src/daemon.rs b/crates/rhema-claw/src/daemon.rs index e9dfc91..d5455bf 100644 --- a/crates/rhema-claw/src/daemon.rs +++ b/crates/rhema-claw/src/daemon.rs @@ -352,6 +352,7 @@ impl Daemon { bootstrap.clone(), hub.clone(), notifications.clone(), + attachments.clone(), config.max_turn_steps, extensions.clone(), config.transcript_enabled(), diff --git a/crates/rhema-claw/src/http.rs b/crates/rhema-claw/src/http.rs index 90d16c9..d81a7d3 100644 --- a/crates/rhema-claw/src/http.rs +++ b/crates/rhema-claw/src/http.rs @@ -20,6 +20,13 @@ //! still a pass; a pathological backlog stops at //! [`MAX_DRAIN_PASSES`] and leaves the rest to the daemon's tick, which //! drains the interactive dispatcher as its backstop. +//! +//! A chat message may carry images. The route checks the budget — +//! [`MAX_INBOUND_IMAGE_BYTES`] decoded per image, [`MAX_INBOUND_IMAGES`] +//! a message — and the dispatch decodes the bodies into the attachment +//! store: the stream replays ids and media types, never megabytes, and +//! a client fetches the bytes through the attachments route it already +//! has. use std::collections::BTreeSet; use std::sync::{Arc, Mutex, PoisonError}; @@ -35,6 +42,7 @@ use axum::routing::put; use axum::routing::{delete, get, post}; use axum::{Router, middleware}; use rhema_autonomous_agent::Driver; +use rhema_core::ImageMediaType; use rhema_rsd::{ DispatchError, Dispatcher, IngestResult, KeyRecord, KeyState, ResourceKey, SqliteStore, Store, }; @@ -49,6 +57,7 @@ use tokio_stream::wrappers::ReceiverStream; use crate::backends::BackendRegistry; use crate::bootstrap::Bootstrap; use crate::hub::{EventHub, TurnEvent}; +use crate::interactive::WireImage; use crate::notifications::NotificationStore; use crate::task_store::TaskStore; use crate::watch_store::{ClawStore, validate_watch}; @@ -235,7 +244,10 @@ impl ClawState { pub fn router(state: ClawState) -> Router { let v1 = Router::new() .route("/sessions", get(sessions).post(sessions_create)) - .route("/sessions/{key}/messages", post(messages)) + .route( + "/sessions/{key}/messages", + post(messages).layer(axum::extract::DefaultBodyLimit::max(MAX_MESSAGE_BODY_BYTES)), + ) .route("/sessions/{key}/events", get(events)) .route("/bootstrap", get(bootstrap_status)) .route("/watches", get(watches_list).post(watches_add)) @@ -508,11 +520,68 @@ async fn bootstrap_status( .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string())) } -/// `{"text": "..."}` — one chat message. +/// The decoded budget for one inbound chat image: 5 MiB decoded per +/// image, eight images a message ([`MAX_INBOUND_IMAGES`]) — enough for +/// screenshots, small enough that one message cannot bury the +/// attachment store or a provider's inline limit. +pub const MAX_INBOUND_IMAGE_BYTES: usize = 5 * 1024 * 1024; + +/// How many images one chat message may carry. +pub const MAX_INBOUND_IMAGES: usize = 8; + +/// The messages route's request-body ceiling, derived from the image +/// budget: [`MAX_INBOUND_IMAGES`] images at [`MAX_INBOUND_IMAGE_BYTES`] +/// decoded each, base64-expanded 4:3, plus room for the text. Axum's +/// default 2 MiB would refuse what the budget allows before this +/// route's own check could answer. +pub const MAX_MESSAGE_BODY_BYTES: usize = + MAX_INBOUND_IMAGE_BYTES * 4 / 3 * MAX_INBOUND_IMAGES + 1024 * 1024; + +/// `{"text": "..."}` — one chat message, plus the images it carries. +/// A body without `images` is the old shape, unchanged. #[derive(Deserialize)] pub struct MessageBody { /// The message text. pub text: String, + /// The message's inline images, each the full MIME type plus bare + /// base64. The route checks the budget, then the dispatch copies + /// the bytes into the attachment store — the stream carries ids, + /// not megabytes. + #[serde(default)] + pub images: Vec, +} + +/// The inbound-image budget: a media type that parses as an image +/// format, a non-empty base64 body (the decoder, not this check, is +/// the base64 judge), decoded size within [`MAX_INBOUND_IMAGE_BYTES`], +/// at most [`MAX_INBOUND_IMAGES`] a message. The OpenAI shim runs the +/// same check over the same budget. +pub(crate) fn validate_images(images: &[WireImage]) -> Result<(), String> { + if images.len() > MAX_INBOUND_IMAGES { + return Err(format!( + "at most {MAX_INBOUND_IMAGES} images a message, got {}", + images.len() + )); + } + for image in images { + if ImageMediaType::parse(&image.media_type).is_none() { + return Err(format!( + "unsupported image media type {:?}", + image.media_type + )); + } + if image.data.trim().is_empty() { + return Err("image data must not be empty".to_owned()); + } + // Bare base64: four characters decode to three bytes, so the + // size check needs no decode. + if image.data.len() * 3 / 4 > MAX_INBOUND_IMAGE_BYTES { + return Err(format!( + "image too large: the decoded budget is {MAX_INBOUND_IMAGE_BYTES} bytes" + )); + } + } + Ok(()) } async fn messages( @@ -534,15 +603,19 @@ async fn messages( ) .into_response(); } + if let Err(reason) = validate_images(&body.images) { + return (StatusCode::BAD_REQUEST, Json(json!({"error": reason}))).into_response(); + } let key = ResourceKey::from(key); // No subscribe-before-ingest here: the hub records every turn // whether or not anyone is watching, so a chat surface connecting // later replays the whole exchange. The reply itself is consumed on // the event stream, not here. - if let Err(err) = state - .dispatcher - .ingest(crate::interactive::user_event(key.as_str(), &body.text)) - { + if let Err(err) = state.dispatcher.ingest(crate::interactive::user_event( + key.as_str(), + &body.text, + &body.images, + )) { return ( StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()})), diff --git a/crates/rhema-claw/src/hub.rs b/crates/rhema-claw/src/hub.rs index c59a2da..55e4840 100644 --- a/crates/rhema-claw/src/hub.rs +++ b/crates/rhema-claw/src/hub.rs @@ -44,18 +44,37 @@ pub const TURN_CHANNEL_CAPACITY: usize = 1024; /// thinking traces and tool rows are deltas, and they add up. pub const RECORD_CAPACITY: usize = 8192; +/// One image a chat message carried, as the transcript records it: the +/// attachment store's id and the media type the bytes carry. The +/// transcript streams ids, never megabytes — a client fetches the +/// bytes through `GET /v1/attachments/{id}`, the route the agent's own +/// attachments already serve. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct UserImage { + /// The attachment store's id — the fetch route's address. + pub id: String, + /// The full MIME type (`image/png`). + pub media_type: String, +} + /// One event of a turn, as the wire sees it. Serialized with an internal /// `type` tag, snake-cased: `{"type":"text_delta","text":"..."}`. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum TurnEvent { - /// The turn's input: the text (a batch folds into one) the turn - /// answers, recorded by the handler when the dispatch starts. It - /// exists for the replay — a chat surface shows its own messages - /// as it sends them, and skips the live echo. + /// The turn's input: the text (a batch folds into one) and the + /// images the message batch carried, recorded by the handler when + /// the dispatch starts. It exists for the replay — a chat surface + /// shows its own messages as it sends them, and skips the live + /// echo. UserMessage { /// The input text, verbatim. text: String, + /// The batch's images, in arrival order, as the attachment + /// store holds them. `default` because rows recorded before + /// images rode the transcript carry no field. + #[serde(default)] + images: Vec, }, /// A piece of the model's reasoning trace, in stream order. ThinkingDelta { @@ -316,8 +335,12 @@ mod tests { r#"{"type":"text_delta","text":"hi"}"# ); assert_eq!( - serde_json::to_string(&TurnEvent::UserMessage { text: "hi".into() }).unwrap(), - r#"{"type":"user_message","text":"hi"}"# + serde_json::to_string(&TurnEvent::UserMessage { + text: "hi".into(), + images: Vec::new(), + }) + .unwrap(), + r#"{"type":"user_message","text":"hi","images":[]}"# ); assert_eq!( serde_json::to_string(&TurnEvent::ToolResult { @@ -340,6 +363,21 @@ mod tests { ); } + /// Rows recorded before images rode the transcript carry no + /// `images` field; they replay as plain questions. + #[test] + fn a_user_message_row_from_before_images_loads_without_them() { + let event: TurnEvent = + serde_json::from_str(r#"{"type":"user_message","text":"hello"}"#).unwrap(); + assert_eq!( + event, + TurnEvent::UserMessage { + text: "hello".into(), + images: Vec::new() + } + ); + } + #[test] fn the_kind_is_the_wire_type_tag() { assert_eq!( @@ -347,7 +385,11 @@ mod tests { "text_delta" ); assert_eq!( - TurnEvent::UserMessage { text: "x".into() }.kind(), + TurnEvent::UserMessage { + text: "x".into(), + images: Vec::new() + } + .kind(), "user_message" ); assert_eq!( @@ -389,6 +431,7 @@ mod tests { &key, TurnEvent::UserMessage { text: "hello".into(), + images: Vec::new(), }, ); hub.emit(&key, text("hi")); @@ -398,7 +441,8 @@ mod tests { history, vec![ TurnEvent::UserMessage { - text: "hello".into() + text: "hello".into(), + images: Vec::new() }, text("hi"), ], @@ -523,6 +567,7 @@ mod tests { &key, TurnEvent::UserMessage { text: "earlier".into(), + images: Vec::new(), }, ); let (history, mut receiver) = hub.subscribe_with_history(&key); diff --git a/crates/rhema-claw/src/interactive.rs b/crates/rhema-claw/src/interactive.rs index ae0e391..3a95100 100644 --- a/crates/rhema-claw/src/interactive.rs +++ b/crates/rhema-claw/src/interactive.rs @@ -17,7 +17,10 @@ //! verbatim, and the handler's only classification is "none". One //! dispatch drains the inbox and folds every drained text into ONE user //! input (the spec §3 batching rule), so a burst of quick messages costs -//! one reasoning step, not one per message. +//! one reasoning step, not one per message. A message may carry images: +//! the batch's images copy into the attachment store, the transcript +//! records their ids, and the turn's input carries the inline image +//! blocks the providers take — the checkpoint holds what the model saw. //! //! **The delivery rule:** `interactive:` keys reply into the live chat //! surface (the event hub's stream). Task, phone, and calendar keys @@ -57,8 +60,8 @@ use std::time::{SystemTime, UNIX_EPOCH}; use rhema_agent_tools::{AllowAll, ExpandPastes, ShellTools, revive_dangling_calls}; use rhema_autonomous_agent::{Allowlist, Handler, ResolverChain, SessionContext, async_trait}; use rhema_core::{ - Agent, ChatParams, Content, Event as ModelEvent, EventSink, Message, ModelProvider, Role, - Session, ToolContext, ToolResolver, run_turn, + Agent, ChatParams, Content, Event as ModelEvent, EventSink, Image, ImageMediaType, Message, + ModelProvider, Role, Session, ToolContext, ToolResolver, run_turn, }; use rhema_memory::{MemoryResolver, MemoryStore, Scope}; use rhema_rsd::{Dispatcher, Event, KeyFn, Outcome, ResourceKey}; @@ -66,7 +69,7 @@ use serde_json::{Value, json}; use crate::backends::BackendRegistry; use crate::bootstrap::{Bootstrap, DEFAULT_SYSTEM_MD, render_system}; -use crate::hub::{EventHub, TurnEvent}; +use crate::hub::{EventHub, TurnEvent, UserImage}; use crate::notifications::NotificationStore; /// The key prefixes of the resident context (see the module docs): @@ -133,10 +136,24 @@ pub fn register_user(dispatcher: &Dispatcher) { dispatcher.register_source(SOURCE, passthrough_key_fn()); } +/// One inbound image on a chat message's payload: the full MIME type +/// and the bare base64 body — no `data:` prefix, exactly the shape the +/// provider APIs take inline. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct WireImage { + /// The full MIME type (`image/png`), parseable by + /// [`ImageMediaType::parse`]. + pub media_type: String, + /// The bare base64 body. + pub data: String, +} + /// The event one chat message becomes. The id is unique per message (a /// wall-clock millisecond plus a process counter), so a redelivery -/// never dedups a different message away. -pub fn user_event(key: &str, text: &str) -> Event { +/// never dedups a different message away. The payload carries the +/// message's images when it has any, and no `images` key when it has +/// none — old payloads stay byte-identical. +pub fn user_event(key: &str, text: &str, images: &[WireImage]) -> Event { static MESSAGE_SEQ: AtomicU64 = AtomicU64::new(0); let now = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -148,11 +165,26 @@ pub fn user_event(key: &str, text: &str) -> Event { MESSAGE_SEQ.fetch_add(1, Ordering::Relaxed) ), source: SOURCE.to_owned(), - payload: json!({ "key": key, "text": text }), + payload: if images.is_empty() { + json!({ "key": key, "text": text }) + } else { + json!({ "key": key, "text": text, "images": images }) + }, received_at: now.as_millis() as u64, } } +/// Decodes an inbound chat image's base64 body: the standard alphabet +/// with padding, the shape every browser and fetch helper mints. +/// `None` when it does not decode — a junk entry rides no further than +/// the fold, and the rest of the batch is unaffected. +fn decode_image_data(data: &str) -> Option> { + use base64::Engine as _; + base64::engine::general_purpose::STANDARD + .decode(data.trim()) + .ok() +} + /// The resident agent: a provider, a model, one shared workspace, the /// bootstrap store its identity files live in, the watch table its /// tools manage, the notification table it reaches devices through, and @@ -169,6 +201,10 @@ pub struct InteractiveHandler { /// The identity files: the system prompt renders from them per /// dispatch, and the `bootstrap_save` resolver writes them. bootstrap: Arc, + /// The attachment store: inbound chat images copy into it, so the + /// transcript carries ids and a client fetches the bytes over the + /// attachments route it already has. + attachments: Arc, bootstrap_resolver: Arc, hub: Arc, /// The notification table: the `notify` tool writes it, and @@ -197,7 +233,7 @@ impl InteractiveHandler { /// shell (approval v1-permissive — the shell is the point of a /// resident coding agent — installed once, since resolution is /// synchronous within a turn), memory, and `bootstrap_save`. - #[expect(clippy::too_many_arguments, reason = "nine parts, each load-bearing")] + #[expect(clippy::too_many_arguments, reason = "ten parts, each load-bearing")] pub fn new( backends: Arc, workspace: impl Into, @@ -205,6 +241,7 @@ impl InteractiveHandler { bootstrap: Arc, hub: Arc, notifications: Arc, + attachments: Arc, max_turn_steps: u32, extensions: Arc, transcript: bool, @@ -223,6 +260,7 @@ impl InteractiveHandler { bootstrap_resolver, hub, notifications, + attachments, max_turn_steps, extensions, transcript_mirror: transcript.then(|| { @@ -311,20 +349,68 @@ impl InteractiveHandler { .poll_inbox() .await .map_err(|err| format!("inbox poll for {key} failed: {err}"))?; - let input = items - .iter() - .filter_map(|item| { - item.event - .payload - .get("text") - .or_else(|| item.event.payload.get("prompt")) - .and_then(Value::as_str) - }) - .collect::>() - .join("\n\n"); + // Texts fold in arrival order, and so do the images, across + // the whole batch whichever events carried them. + let mut texts = Vec::new(); + let mut images = Vec::new(); + for item in &items { + let payload = &item.event.payload; + if let Some(text) = payload + .get("text") + .or_else(|| payload.get("prompt")) + .and_then(Value::as_str) + { + texts.push(text); + } + if let Some(carried) = payload.get("images").and_then(Value::as_array) { + images.extend( + carried.iter().filter_map(|image| { + serde_json::from_value::(image.clone()).ok() + }), + ); + } + } + let input = texts.join("\n\n"); if input.trim().is_empty() { return Err(format!("dispatch for {key} carried no text")); } + // Inbound images: the bytes copy into the attachment store and + // the transcript records ids — a replay streams the same small + // cards every other event does, never the megabytes again. A + // payload the messages route never validated (the generic + // ingest door) loses its junk entries here: an unparseable + // media type or an undecodable body rides nothing, and the + // rest of the batch is unaffected. + let mut transcript_images = Vec::new(); + let mut core_images = Vec::new(); + for (n, image) in images.iter().enumerate() { + let Some(media_type) = ImageMediaType::parse(&image.media_type) else { + continue; + }; + let Some(bytes) = decode_image_data(&image.data) else { + continue; + }; + let extension = match media_type { + ImageMediaType::Png => "png", + // `jpg`, not `jpeg`: the shorter form is the + // conventional file name, and the store's content type + // guess knows both. + ImageMediaType::Jpeg => "jpg", + ImageMediaType::Gif => "gif", + ImageMediaType::Webp => "webp", + }; + let stored = self + .attachments + .put(&format!("image-{}.{extension}", n + 1), &bytes) + .map_err(|err| format!("image store for {key} failed: {err}"))?; + transcript_images.push(UserImage { + id: stored.id, + media_type: media_type.as_str().to_owned(), + }); + // The request's base64 verbatim: the session checkpoint + // carries what the model saw. + core_images.push(Image::new(media_type, image.data.clone())); + } // The transcript's question, recorded before anything else can // stream: a replay shows what was asked, not only what was // answered. Raw input — the notification-contract framing below @@ -333,6 +419,7 @@ impl InteractiveHandler { &ctx.key, TurnEvent::UserMessage { text: input.clone(), + images: transcript_images, }, ); // A non-chat resident key has no subscriber to read the reply: @@ -385,7 +472,7 @@ impl InteractiveHandler { let output = run_turn( &mut agent, &mut session, - Message::text(Role::User, input), + Message::text_with_images(Role::User, input, core_images), &mut sink, &tools, self.max_turn_steps, @@ -753,6 +840,30 @@ mod tests { }) .expect("a user text") } + + /// The last request's user-turn images, as (media type, base64) + /// pairs in block order — what the model was handed inline. + fn last_user_images(&self) -> Vec<(String, String)> { + let requests = self.requests.lock().unwrap(); + let last = requests.last().expect("a request was made"); + last.messages + .iter() + .rev() + .find(|message| message.role == CoreRole::User) + .map(|message| { + message + .content + .iter() + .filter_map(|content| match content { + Content::Image(image) => { + Some((image.media_type.as_str().to_owned(), image.data.clone())) + } + _ => None, + }) + .collect() + }) + .expect("a user turn") + } } #[async_trait] @@ -797,6 +908,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub, notifications, + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, on, @@ -844,6 +956,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -864,6 +977,7 @@ mod tests { .ingest(user_event( key.as_str(), "review this:\u{FFFC}fn main() {}\nfn other() {}\u{FFFC}", + &[], )) .expect("ingests"); driver.run_once().await.expect("the run happens"); @@ -875,7 +989,7 @@ mod tests { ); let (history, _) = hub.subscribe_with_history(&key); let saw_input = history.into_iter().find_map(|event| match event { - TurnEvent::UserMessage { text } => Some(text), + TurnEvent::UserMessage { text, .. } => Some(text), _ => None, }); assert!( @@ -932,6 +1046,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -1005,8 +1120,8 @@ mod tests { #[test] fn user_events_carry_the_key_and_unique_ids() { - let first = user_event("interactive:web", "hello"); - let second = user_event("interactive:web", "again"); + let first = user_event("interactive:web", "hello", &[]); + let second = user_event("interactive:web", "again", &[]); assert_eq!(first.source, "user"); assert_eq!(first.id, first.id, "stable within itself"); assert_ne!(first.id, second.id, "messages never share a dedup id"); @@ -1016,6 +1131,43 @@ mod tests { ); } + /// An image-carrying message names its images on the payload, in + /// order; a plain message carries no `images` key at all. + #[test] + fn user_events_carry_images_only_when_the_message_has_any() { + let bare = user_event("interactive:web", "plain", &[]); + assert_eq!( + bare.payload, + json!({"key": "interactive:web", "text": "plain"}), + "no images, no key" + ); + let carried = user_event( + "interactive:web", + "look", + &[ + WireImage { + media_type: "image/png".to_owned(), + data: "cG5n".to_owned(), + }, + WireImage { + media_type: "image/webp".to_owned(), + data: "d2VicA==".to_owned(), + }, + ], + ); + assert_eq!( + carried.payload, + json!({ + "key": "interactive:web", + "text": "look", + "images": [ + {"media_type": "image/png", "data": "cG5n"}, + {"media_type": "image/webp", "data": "d2VicA=="}, + ], + }) + ); + } + #[test] fn the_passthrough_key_fn_reads_the_payload_verbatim() { let dispatcher = Dispatcher::new( @@ -1024,7 +1176,7 @@ mod tests { Arc::new(VolatileStore::new()), ); register_user(&dispatcher); - let event = user_event("interactive:web", "hello"); + let event = user_event("interactive:web", "hello", &[]); // Ingesting resolves the key through the registered fn: the // record lands under exactly the payload's key. dispatcher.ingest(event).expect("ingests"); @@ -1203,6 +1355,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -1221,7 +1374,7 @@ mod tests { let key = Key::from("interactive:web"); let mut receiver = hub.subscribe(&key); dispatcher - .ingest(user_event(key.as_str(), "hello")) + .ingest(user_event(key.as_str(), "hello", &[])) .expect("ingests"); let driver = rhema_autonomous_agent::Driver::new(dispatcher.clone(), Arc::new(handler)) @@ -1233,7 +1386,8 @@ mod tests { assert_eq!( receiver.try_recv().unwrap(), TurnEvent::UserMessage { - text: "hello".into() + text: "hello".into(), + images: Vec::new() } ); assert_eq!( @@ -1256,6 +1410,142 @@ mod tests { ); } + /// A batch whose messages carry images: the bytes land in the + /// attachment store under image-. names, the transcript + /// carries the stored ids and media types in arrival order, and the + /// turn's input is the texts folded with the inline images behind + /// them — base64 verbatim, what the checkpoint will hold. + #[tokio::test] + async fn a_batched_dispatch_stores_images_and_the_turn_sees_them() { + use base64::Engine as _; + + let recorder = RecordingProvider::new(); + let store = Arc::new(ClawStore::in_memory().expect("store opens")); + let hub = Arc::new(EventHub::default()); + let workspace = tempfile::tempdir().expect("tempdir"); + let attachments = Arc::new( + crate::attachments::AttachmentStore::open(workspace.path().join("attachments")) + .expect("store opens"), + ); + let notifications = notifications(); + let extensions = test_extensions( + hub.clone(), + store, + notifications.clone(), + tasks(), + jobs(), + accounts(), + consents(), + bridge(), + attachments.clone(), + phone(), + identity(), + &workspace.path().join("interactive"), + ); + let handler = InteractiveHandler::new( + test_registry(recorder.clone()), + workspace.path().join("interactive"), + Arc::new(rhema_memory::VolatileStore::new()), + Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), + hub.clone(), + notifications.clone(), + attachments.clone(), + crate::config::DEFAULT_MAX_TURN_STEPS, + extensions, + true, + ); + let dispatcher = Arc::new(Dispatcher::new( + DispatcherConfig { + debounce_window_millis: 0, + ..DispatcherConfig::default() + }, + Arc::new(rhema_rsd::SystemClock), + Arc::new(VolatileStore::new()), + )); + register_user(&dispatcher); + let driver = rhema_autonomous_agent::Driver::new(dispatcher.clone(), Arc::new(handler)) + .with_holder("test"); + let key = Key::from("interactive:web"); + let png = base64::engine::general_purpose::STANDARD.encode(b"\x89PNG-fake"); + let jpeg = base64::engine::general_purpose::STANDARD.encode(b"\xff\xd8jpeg"); + + // Two messages with images and one without, one batch. + dispatcher + .ingest(user_event( + key.as_str(), + "look at this", + &[WireImage { + media_type: "image/png".to_owned(), + data: png.clone(), + }], + )) + .expect("ingests"); + dispatcher + .ingest(user_event( + key.as_str(), + "and at this", + &[WireImage { + media_type: "Image/JPEG".to_owned(), + data: jpeg.clone(), + }], + )) + .expect("ingests"); + dispatcher + .ingest(user_event(key.as_str(), "no picture here", &[])) + .expect("ingests"); + assert_eq!( + driver.run_once().await.expect("the run happens"), + vec![RsdOutcome::Handled] + ); + + // The transcript's question carries both stored images, in + // arrival order, media types canonical. + let (history, _) = hub.subscribe_with_history(&key); + let question = history + .iter() + .find_map(|event| match event { + TurnEvent::UserMessage { text, images } => Some((text.clone(), images.clone())), + _ => None, + }) + .expect("the transcript saw the question"); + assert_eq!(question.0, "look at this\n\nand at this\n\nno picture here"); + assert_eq!(question.1.len(), 2); + assert_eq!(question.1[0].media_type, "image/png"); + assert_eq!(question.1[1].media_type, "image/jpeg"); + + // Each id fetches the exact decoded bytes from the store, under + // the numbered image-. name. + let (stored, bytes) = attachments + .get(&question.1[0].id) + .expect("the store reads") + .expect("the image is stored"); + assert_eq!(stored.name, "image-1.png"); + // The name is what the store's media guess reads: an image, so + // a client renders it inline. + assert_eq!(crate::attachments::media_of(&stored.name), "image"); + assert_eq!(bytes, b"\x89PNG-fake"); + let (stored, bytes) = attachments + .get(&question.1[1].id) + .expect("the store reads") + .expect("the image is stored"); + assert_eq!(stored.name, "image-2.jpg", "jpeg stores under jpg"); + assert_eq!(bytes, b"\xff\xd8jpeg"); + + // The model saw the folded text, then the images inline, base64 + // verbatim. + assert_eq!( + recorder.last_user_text(), + "look at this\n\nand at this\n\nno picture here" + ); + assert_eq!( + recorder.last_user_images(), + vec![ + ("image/png".to_owned(), png), + ("image/jpeg".to_owned(), jpeg) + ] + ); + } + /// Memory rides the resolver chain across dispatches: the model saves /// in one session turn, and the record is in the durable store for /// every session after — searchable by the model, never injected. @@ -1299,6 +1589,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -1319,7 +1610,7 @@ mod tests { // Dispatch 1: the model saves through the chain's memory tools. dispatcher - .ingest(user_event(key.as_str(), "remember the channel")) + .ingest(user_event(key.as_str(), "remember the channel", &[])) .expect("ingests"); assert_eq!( driver.run_once().await.expect("first run"), @@ -1338,7 +1629,7 @@ mod tests { // Dispatch 2: the model searches the same chain and answers from // memory — a fresh session turn, the same durable store. dispatcher - .ingest(user_event(key.as_str(), "where do releases go?")) + .ingest(user_event(key.as_str(), "where do releases go?", &[])) .expect("ingests"); assert_eq!( driver.run_once().await.expect("second run"), @@ -1404,6 +1695,7 @@ mod tests { bootstrap.clone(), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -1425,7 +1717,7 @@ mod tests { // Dispatch 1: the model writes SOUL.md through the chain. A key // with a session continues, so both calls ride one key. dispatcher - .ingest(user_event(key.as_str(), "call me claw")) + .ingest(user_event(key.as_str(), "call me claw", &[])) .expect("ingests"); assert_eq!( driver.run_once().await.expect("first run"), @@ -1444,7 +1736,7 @@ mod tests { // Dispatch 2: USER.md, same chain — and the pending flag clears. dispatcher - .ingest(user_event(key.as_str(), "and I am Ana")) + .ingest(user_event(key.as_str(), "and I am Ana", &[])) .expect("ingests"); assert_eq!( driver.run_once().await.expect("second run"), @@ -1485,6 +1777,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -1501,7 +1794,7 @@ mod tests { register_user(&dispatcher); let key = Key::from("interactive:web"); dispatcher - .ingest(user_event(key.as_str(), "hello")) + .ingest(user_event(key.as_str(), "hello", &[])) .expect("ingests"); let driver = rhema_autonomous_agent::Driver::new(dispatcher, Arc::new(handler)); driver.run_once().await.expect("the run happens"); @@ -1567,6 +1860,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -1640,7 +1934,7 @@ mod tests { // An interactive reply is the chat surface's business: notified // nowhere. dispatcher - .ingest(user_event("interactive:web", "hello")) + .ingest(user_event("interactive:web", "hello", &[])) .expect("ingests"); assert_eq!( driver.run_once().await.expect("the run happens"), @@ -1691,6 +1985,7 @@ mod tests { Arc::new(Bootstrap::new(workspace.path().join("bootstrap")).expect("store opens")), hub.clone(), notifications.clone(), + attachments(), crate::config::DEFAULT_MAX_TURN_STEPS, extensions, true, @@ -1717,7 +2012,7 @@ mod tests { // The evening's work, in the chat session. dispatcher - .ingest(user_event(key.as_str(), "we shipped the claw tonight")) + .ingest(user_event(key.as_str(), "we shipped the claw tonight", &[])) .expect("ingests"); assert_eq!( driver.run_once().await.expect("first run"), diff --git a/crates/rhema-claw/src/openai.rs b/crates/rhema-claw/src/openai.rs index 8fc7cdb..7875f5a 100644 --- a/crates/rhema-claw/src/openai.rs +++ b/crates/rhema-claw/src/openai.rs @@ -191,9 +191,10 @@ pub async fn chat_completions( // Subscribe first, ingest second, drain third — the same contract // the messages route keeps. let mut receiver = state.hub().subscribe(&key); - if let Err(err) = state - .dispatcher() - .ingest(crate::interactive::user_event(key.as_str(), &text)) + if let Err(err) = + state + .dispatcher() + .ingest(crate::interactive::user_event(key.as_str(), &text, &[])) { return error_response(StatusCode::INTERNAL_SERVER_ERROR, &err.to_string()); } diff --git a/crates/rhema-claw/src/turn_store.rs b/crates/rhema-claw/src/turn_store.rs index df9d34b..c5cf14f 100644 --- a/crates/rhema-claw/src/turn_store.rs +++ b/crates/rhema-claw/src/turn_store.rs @@ -222,6 +222,7 @@ mod tests { &[ TurnEvent::UserMessage { text: "hello".into(), + images: Vec::new(), }, text_delta("hi"), ], @@ -240,7 +241,8 @@ mod tests { store.events("interactive:web").expect("reads"), vec![ TurnEvent::UserMessage { - text: "hello".into() + text: "hello".into(), + images: Vec::new() }, text_delta("hi"), TurnEvent::TurnDone { diff --git a/crates/rhema-claw/tests/http.rs b/crates/rhema-claw/tests/http.rs index ddf71b6..d3c3e1c 100644 --- a/crates/rhema-claw/tests/http.rs +++ b/crates/rhema-claw/tests/http.rs @@ -274,6 +274,163 @@ async fn chat_streams_turns_and_remembers_across_messages() { assert_eq!(refused.status(), 400, "only resident keys take messages"); } +/// A chat message with images: the route enforces the budget (media +/// type, non-empty body, decoded size, count), the transcript's +/// user_message card names the stored ids and media types, and each id +/// fetches the exact bytes back through the attachments route. A body +/// without `images` is the old shape — accepted unchanged. +#[tokio::test] +async fn chat_messages_carry_images_into_the_attachment_store() { + use base64::Engine as _; + + let provider = Arc::new(MockProvider::new([reply("nice pictures")])); + let dir = tempfile::tempdir().expect("tempdir"); + let daemon = Daemon::build( + &config(&dir, None), + mock_registry(provider), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + let messages = format!("{base}/v1/sessions/interactive:web/messages"); + + let events = client + .get(format!("{base}/v1/sessions/interactive:web/events")) + .send() + .await + .expect("the events request goes out"); + let mut stream = sse_body(events).await; + let mut seen = String::new(); + + // The old shape: no images field, accepted as ever. + let accepted = client + .post(&messages) + .json(&json!({"text": "no pictures"})) + .send() + .await + .expect("the message posts"); + assert_eq!(accepted.status(), 202); + read_until(&mut stream, &mut seen, r#""type":"turn_done""#).await; + assert!( + !seen.contains(r#""images":[{"#), + "a plain message carries no image cards: {seen}" + ); + + // The image shape: one png, posted with the text. + let raw: &[u8] = b"\x89PNG\r\n\x1a\n fake bytes"; + let png = base64::engine::general_purpose::STANDARD.encode(raw); + let accepted = client + .post(&messages) + .json(&json!({ + "text": "look at this", + "images": [{"media_type": "image/png", "data": png}], + })) + .send() + .await + .expect("the message posts"); + assert_eq!(accepted.status(), 202); + read_until(&mut stream, &mut seen, r#""images":[{"id""#).await; + read_until(&mut stream, &mut seen, r#""type":"turn_done""#).await; + let card = seen + .rsplit(r#""type":"user_message""#) + .next() + .expect("a card"); + let id = card + .split(r#""id":""#) + .nth(1) + .and_then(|rest| rest.split('"').next()) + .expect("the card names the stored id") + .to_owned(); + let media_type = card + .split(r#""media_type":""#) + .nth(1) + .and_then(|rest| rest.split('"').next()) + .expect("the card names the media type"); + assert_eq!(media_type, "image/png"); + assert!(id.starts_with("att-"), "{id}"); + + // The stored bytes are fetchable, and they are the original ones. + let fetched = client + .get(format!("{base}/v1/attachments/{id}")) + .send() + .await + .expect("the fetch goes out"); + assert_eq!(fetched.status(), 200); + assert_eq!( + fetched + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()), + Some("image/png"), + "the content type follows the stored name" + ); + assert_eq!(fetched.bytes().await.expect("the bytes arrive"), raw); + + // The budget: an unparseable media type, an empty body, an + // oversized body, and a ninth image are all refusals. + async fn refused(client: &reqwest::Client, messages: &str, body: Value) -> reqwest::StatusCode { + client + .post(messages) + .json(&body) + .send() + .await + .expect("the refusal posts") + .status() + } + let oversized = "A".repeat((rhema_claw::http::MAX_INBOUND_IMAGE_BYTES + 1) * 4 / 3 + 4); + let nine = |data: &str| { + json!({ + "text": "too many", + "images": vec![json!({"media_type": "image/png", "data": data}); 9], + }) + }; + assert_eq!( + refused( + &client, + &messages, + json!({ + "text": "look", + "images": [{"media_type": "image/svg+xml", "data": "cG5n"}], + }) + ) + .await, + 400, + "svg is not an image format the providers take" + ); + assert_eq!( + refused( + &client, + &messages, + json!({ + "text": "look", + "images": [{"media_type": "image/png", "data": " "}], + }) + ) + .await, + 400, + "an empty body carries no image" + ); + assert_eq!( + refused( + &client, + &messages, + json!({ + "text": "look", + "images": [{"media_type": "image/png", "data": oversized}], + }) + ) + .await, + 400, + "over the decoded budget" + ); + assert_eq!( + refused(&client, &messages, nine("cG5n")).await, + 400, + "eight images a message" + ); +} + /// A turn nobody watched is still replayable: a client arriving after /// the fact — a reloaded browser, a reopened phone — subscribes cold and /// gets the finished turns back before the live feed. The same holds