feat(claw): image attachments on chat messages

A chat message may carry inline images: the messages route checks the
budget (5 MiB decoded per image, eight a message) and the dispatch
copies the bytes into the attachment store, so the replay streams ids
and media types over SSE, never megabytes. The turn's input carries
the inline image blocks the providers take, base64 verbatim — the
session checkpoint holds what the model saw.

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"

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(),

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<WireImage>,
+}
+
+/// 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
+    if let Err(err) = state.dispatcher.ingest(crate::interactive::user_event(
-        .dispatcher
+        key.as_str(),
-        .ingest(crate::interactive::user_event(key.as_str(), &body.text))
+        &body.text,
-    {
+        &body.images,
+    )) {
         return (
             StatusCode::INTERNAL_SERVER_ERROR,
             Json(json!({"error": err.to_string()})),

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
+    /// The turn's input: the text (a batch folds into one) and the
-    /// answers, recorded by the handler when the dispatch starts. It
+    /// images the message batch carried, recorded by the handler when
-    /// exists for the replay — a chat surface shows its own messages
+    /// the dispatch starts. It exists for the replay — a chat surface
-    /// as it sends them, and skips the live echo.
+    /// 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<UserImage>,
     },
     /// 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(),
+            serde_json::to_string(&TurnEvent::UserMessage {
-            r#"{"type":"user_message","text":"hi"}"#
+                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);

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,
+    Agent, ChatParams, Content, Event as ModelEvent, EventSink, Image, ImageMediaType, Message,
-    Session, ToolContext, ToolResolver, run_turn,
+    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.
+/// never dedups a different message away. The payload carries the
-pub fn user_event(key: &str, text: &str) -> Event {
+/// 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<Vec<u8>> {
+    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<Bootstrap>,
+    /// 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<crate::attachments::AttachmentStore>,
     bootstrap_resolver: Arc<dyn rhema_core::ToolResolver>,
     hub: Arc<EventHub>,
     /// 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<BackendRegistry>,
         workspace: impl Into<PathBuf>,
@@ -205,6 +241,7 @@ impl InteractiveHandler {
         bootstrap: Arc<Bootstrap>,
         hub: Arc<EventHub>,
         notifications: Arc<NotificationStore>,
+        attachments: Arc<crate::attachments::AttachmentStore>,
         max_turn_steps: u32,
         extensions: Arc<crate::extension::Extensions>,
         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
+        // Texts fold in arrival order, and so do the images, across
-            .iter()
+        // the whole batch whichever events carried them.
-            .filter_map(|item| {
+        let mut texts = Vec::new();
-                item.event
+        let mut images = Vec::new();
-                    .payload
+        for item in &items {
-                    .get("text")
+            let payload = &item.event.payload;
-                    .or_else(|| item.event.payload.get("prompt"))
+            if let Some(text) = payload
-                    .and_then(Value::as_str)
+                .get("text")
-            })
+                .or_else(|| payload.get("prompt"))
-            .collect::<Vec<_>>()
+                .and_then(Value::as_str)
-            .join("\n\n");
+            {
+                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::<WireImage>(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 first = user_event("interactive:web", "hello", &[]);
-        let second = user_event("interactive:web", "again");
+        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-<n>.<ext> 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-<n>.<ext> 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"),

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
+    if let Err(err) =
-        .dispatcher()
+        state
-        .ingest(crate::interactive::user_event(key.as_str(), &text))
+            .dispatcher()
+            .ingest(crate::interactive::user_event(key.as_str(), &text, &[]))
     {
         return error_response(StatusCode::INTERNAL_SERVER_ERROR, &err.to_string());
     }

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 {

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