feat(runtime): preserve validated images through existing turns

Accept negotiated images on Runtime HTTP, relay Chat and daemon prompt paths
using the existing bounded decoder, Engine content blocks and turn store.
Reject malformed bytes, Auto and unknown/unsupported exact model routes before
provider dispatch; explicitly reject images on legacy remote Work.

Bind ordered images to replay identity, preserve image history through restore,
fork and retry, reject unsupported schema versions, and validate failed stream
and saved-session admission before durable empty-thread creation. Extend the
existing model catalog with exact named-route identity and scoped cursors.
Keep empty-image text hashes and trusted-local history bounds unchanged.

Validation: final fresh TUI/app-server/protocol gate 98 passed, 0 failed,
12308 skipped; production and test-target cargo check passed; cargo fmt check
and git diff --check passed. All providers mocked, HTTP loopback only.
The workspace npm test/check:web commands are unavailable in this Core package
(missing scripts); no Apps, hosted CI, live provider or deployment claim.

crates/app-server/src/lib.rs

@@ -1,3 +1,4 @@
+use codewhale_protocol::runtime::{MAX_RUNTIME_IMAGE_BODY_BYTES, RuntimeImageInput};
 use std::collections::{HashMap, VecDeque};
 use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
@@ -308,6 +309,8 @@ struct ThreadIdParams {
 struct ThreadMessageParams {
     thread_id: String,
     input: String,
+    #[serde(default)]
+    images: Vec<RuntimeImageInput>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -353,9 +356,19 @@ async fn shutdown_signal() {
 
 fn app_router(state: AppState, cors_origins: &[String]) -> Router {
     let protected_routes = Router::new()
-        .route("/thread", post(thread_handler))
+        .route(
+            "/thread",
+            post(thread_handler).layer(axum::extract::DefaultBodyLimit::max(
+                MAX_RUNTIME_IMAGE_BODY_BYTES,
+            )),
+        )
         .route("/app", post(app_handler))
-        .route("/prompt", post(prompt_handler))
+        .route(
+            "/prompt",
+            post(prompt_handler).layer(axum::extract::DefaultBodyLimit::max(
+                MAX_RUNTIME_IMAGE_BODY_BYTES,
+            )),
+        )
         .route("/tool", post(tool_handler))
         .route("/jobs", get(jobs_handler))
         .route("/mcp/startup", post(mcp_startup_handler))
@@ -521,6 +534,12 @@ enum ParsedStdioLine {
 }
 
 fn parse_stdio_line(line: &str) -> ParsedStdioLine {
+    if line.len() > MAX_RUNTIME_IMAGE_BODY_BYTES {
+        return ParsedStdioLine::Rejected(jsonrpc_error(
+            None,
+            JsonRpcError::invalid_params("request exceeds the 8 MiB transport limit"),
+        ));
+    }
     if line.trim().is_empty() {
         return ParsedStdioLine::Blank;
     }
@@ -648,8 +667,13 @@ async fn thread_handler(State(state): State<AppState>, Json(req): Json<ThreadReq
     // A message is a turn, and turns belong to the runtime — not to the
     // bookkeeping `Runtime` behind the other thread operations. This mirrors
     // the interception stdio `thread/message` has always done.
-    if let ThreadRequest::Message { thread_id, input } = req {
-        return match run_http_thread_message(&state, thread_id, input).await {
+    if let ThreadRequest::Message {
+        thread_id,
+        input,
+        images,
+    } = req
+    {
+        return match run_http_thread_message(&state, thread_id, input, images).await {
             Ok(res) => (StatusCode::OK, Json(res)).into_response(),
             Err(err) => http_error_from_jsonrpc(err).into_response(),
         };
@@ -1116,6 +1140,7 @@ struct BridgedTurn<'a> {
     /// Client-facing thread id; the bridge maps it to a runtime thread.
     thread_key: &'a str,
     input: &'a str,
+    images: &'a [RuntimeImageInput],
     /// Model for the runtime thread when this call is the one that creates
     /// it. An existing thread keeps the model it was created with.
     model_override: Option<String>,
@@ -1164,6 +1189,7 @@ async fn run_bridged_turn<W: AsyncWrite + Unpin>(
         .message_thread(
             &runtime_thread_id,
             turn.input,
+            turn.images,
             writer,
             registration,
             transcript,
@@ -1207,6 +1233,7 @@ async fn run_prompt_turn<W: AsyncWrite + Unpin>(
         BridgedTurn {
             thread_key: &thread_key,
             input: &req.prompt,
+            images: &req.images,
             model_override: req.model.clone(),
             // `thread/interrupt` addresses client-facing thread ids. A
             // one-shot prompt has none to hand back, and a caller-supplied
@@ -1256,6 +1283,7 @@ async fn run_http_thread_message(
     state: &AppState,
     thread_id: String,
     input: String,
+    images: Vec<RuntimeImageInput>,
 ) -> std::result::Result<ThreadResponse, JsonRpcError> {
     let mut transcript = TurnTranscript::default();
     let mut sink = tokio::io::sink();
@@ -1265,6 +1293,7 @@ async fn run_http_thread_message(
         BridgedTurn {
             thread_key: &thread_id,
             input: &input,
+            images: &images,
             model_override: None,
             interruptible: false,
             ephemeral: false,
@@ -1302,6 +1331,7 @@ async fn handle_stdio_thread_message<W: AsyncWrite + Unpin>(
         BridgedTurn {
             thread_key: &parsed.thread_id,
             input: &parsed.input,
+            images: &parsed.images,
             model_override: None,
             interruptible: true,
             ephemeral: false,
@@ -1557,17 +1587,39 @@ impl RuntimeBridge {
         &mut self,
         thread_id: &str,
         input: &str,
+        images: &[RuntimeImageInput],
         writer: &mut W,
         registration: Option<(TurnRegistry, String)>,
         mut transcript: Option<&mut TurnTranscript>,
     ) -> Result<Value> {
+        let mut request = json!({ "prompt": input });
+        if !images.is_empty() {
+            let info = self
+                .request_json(
+                    self.authed(
+                        self.client
+                            .get(format!("{}/v1/runtime/info", self.base_url)),
+                    ),
+                )
+                .await?;
+            if info
+                .pointer("/capabilities/turn_image_inputs")
+                .and_then(Value::as_bool)
+                != Some(true)
+            {
+                bail!(
+                    "Runtime image input is unavailable; update the Runtime before sending attachments"
+                );
+            }
+            request["images"] = json!(images);
+        }
         let turn = self
             .request_json(
                 self.authed(
                     self.client
                         .post(format!("{}/v1/threads/{thread_id}/turns", self.base_url)),
                 )
-                .json(&json!({ "prompt": input })),
+                .json(&request),
             )
             .await?;
         let turn_id = turn
@@ -1943,6 +1995,7 @@ async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
                 result: json!({
                     "transport": transport.label(),
                     "families": ["thread/*", "app/*", "prompt/*"],
+                    "turn_image_inputs": true,
                     "methods": methods,
                 }),
                 should_exit: false,
@@ -1950,6 +2003,7 @@ async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
         }
         "thread/capabilities" => StdioDispatchResult {
             result: json!({
+                "turn_image_inputs": true,
                 "methods": [
                     "thread/request",
                     "thread/create",
@@ -1972,11 +2026,20 @@ async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
         },
         "thread/request" => {
             let request: ThreadRequest = parse_params(params)?;
-            if let ThreadRequest::Message { thread_id, input } = request {
+            if let ThreadRequest::Message {
+                thread_id,
+                input,
+                images,
+            } = request
+            {
                 let response = handle_stdio_thread_message(
                     state,
                     writer,
-                    ThreadMessageParams { thread_id, input },
+                    ThreadMessageParams {
+                        thread_id,
+                        input,
+                        images,
+                    },
                 )
                 .await?;
                 return Ok(StdioDispatchResult {
@@ -3461,7 +3524,7 @@ mod tests {
         let (mut reader, mut writer) = tokio::io::duplex(4096);
 
         let result = bridge
-            .message_thread("thr_test", "hello", &mut writer, None, None)
+            .message_thread("thr_test", "hello", &[], &mut writer, None, None)
             .await
             .expect("message_thread should succeed");
         drop(writer);
@@ -3753,9 +3816,10 @@ mod tests {
         let (base_url, prompts, server) = spawn_stub_runtime().await;
         seed_bridge_at(&state, base_url).await;
 
-        let response = run_http_thread_message(&state, "thr_http".to_string(), "go".to_string())
-            .await
-            .expect("http thread message");
+        let response =
+            run_http_thread_message(&state, "thr_http".to_string(), "go".to_string(), Vec::new())
+                .await
+                .expect("http thread message");
 
         assert_eq!(response.status, "completed");
         assert_eq!(response.thread_id, "thr_http");
@@ -3778,9 +3842,10 @@ mod tests {
         let (state, _tmp) = capability_test_state();
         seed_bridge_at(&state, "http://127.0.0.1:9".to_string()).await;
 
-        let err = run_http_thread_message(&state, "thr_http".to_string(), "go".to_string())
-            .await
-            .expect_err("no runtime means no turn");
+        let err =
+            run_http_thread_message(&state, "thr_http".to_string(), "go".to_string(), Vec::new())
+                .await
+                .expect_err("no runtime means no turn");
         assert_eq!(err.code, RUNTIME_UNAVAILABLE_CODE);
     }
 
@@ -4161,4 +4226,76 @@ mod tests {
         assert!(DEFAULT_CORS_ORIGINS.contains(&"http://localhost:5173"));
         assert!(DEFAULT_CORS_ORIGINS.contains(&"tauri://localhost"));
     }
+    #[tokio::test]
+    async fn runtime_image_daemon_bridge_checks_transport_and_forwards_exact_wire() {
+        async fn capture(
+            State(seen): State<Arc<Mutex<Vec<Value>>>>,
+            Json(body): Json<Value>,
+        ) -> (StatusCode, Json<Value>) {
+            seen.lock().await.push(body);
+            (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error":"fixture stops before an Engine"})),
+            )
+        }
+        for supported in [false, true] {
+            let seen = Arc::new(Mutex::new(Vec::new()));
+            let app = Router::new()
+                .route(
+                    "/v1/runtime/info",
+                    get(move || async move {
+                        Json(json!({"capabilities":{"turn_image_inputs":supported}}))
+                    }),
+                )
+                .route("/v1/threads/{id}/turns", post(capture))
+                .with_state(seen.clone());
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let addr = listener.local_addr().unwrap();
+            let server = tokio::spawn(async move {
+                axum::serve(listener, app).await.unwrap();
+            });
+            let mut bridge = RuntimeBridge::from_base_url_for_test(format!("http://{addr}"));
+            let images = vec![RuntimeImageInput {
+                mime: "image/png".into(),
+                data_base64: "fixture-bytes-validated-by-Core".into(),
+            }];
+            let mut writer = tokio::io::sink();
+            assert!(
+                bridge
+                    .message_thread("thr_fixture", "look", &images, &mut writer, None, None)
+                    .await
+                    .is_err()
+            );
+            let requests = seen.lock().await;
+            assert_eq!(requests.len(), usize::from(supported));
+            if supported {
+                assert_eq!(requests[0], json!({"prompt":"look","images":images}));
+            }
+            server.abort();
+        }
+    }
+
+    #[test]
+    fn runtime_image_daemon_all_input_families_preserve_images() {
+        let image = json!({"mime":"image/png","dataBase64":"AQ=="});
+        let thread: ThreadMessageParams = serde_json::from_value(
+            json!({"thread_id":"thr_fixture","input":"look","images":[image.clone()]}),
+        )
+        .unwrap();
+        let prompt: PromptRequest =
+            serde_json::from_value(json!({"prompt":"look","images":[image.clone()]})).unwrap();
+        let generic: ThreadRequest = serde_json::from_value(
+            json!({"kind":"message","thread_id":"thr_fixture","input":"look","images":[image]}),
+        )
+        .unwrap();
+        assert_eq!(thread.images, prompt.images);
+        let ThreadRequest::Message { images, .. } = generic else {
+            panic!("message");
+        };
+        assert_eq!(thread.images, images);
+        assert!(matches!(
+            parse_stdio_line(&" ".repeat(MAX_RUNTIME_IMAGE_BODY_BYTES + 1)),
+            ParsedStdioLine::Rejected(_)
+        ));
+    }
 }

crates/core/src/lib.rs

@@ -3269,6 +3269,7 @@ mod tests {
 
         let err = runtime
             .handle_thread(ThreadRequest::Message {
+                images: Vec::new(),
                 thread_id: thread_id.clone(),
                 input: "run this".to_string(),
             })

crates/protocol/src/lib.rs

@@ -359,6 +359,8 @@ pub enum ThreadRequest {
     Message {
         thread_id: String,
         input: String,
+        #[serde(default, skip_serializing_if = "Vec::is_empty")]
+        images: Vec<runtime::RuntimeImageInput>,
     },
 }
 
@@ -462,6 +464,8 @@ pub struct PromptRequest {
     pub thread_id: Option<String>,
     /// The prompt text.
     pub prompt: String,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub images: Vec<runtime::RuntimeImageInput>,
     /// Model override, or the default if omitted.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub model: Option<String>,

crates/protocol/src/op.rs

@@ -116,6 +116,8 @@ pub enum Op {
     /// `Op::SendMessage` payloads.
     SendMessage {
         content: String,
+        #[serde(default, skip_serializing_if = "Vec::is_empty")]
+        images: Vec<crate::runtime::RuntimeImageInput>,
         /// Effective mode for this turn (`"plan" | "agent" | "operate"` etc).
         #[serde(default = "default_mode")]
         mode: String,
@@ -500,6 +502,7 @@ pub fn headless_send_message_op(thread_id: ThreadId, content: impl Into<String>)
         session_id: SessionId::new(),
         op: Op::SendMessage {
             content: content.into(),
+            images: Vec::new(),
             mode: default_mode(),
             model: None,
             model_provider: None,

crates/protocol/src/runtime/mod.rs

@@ -7,6 +7,20 @@ use serde_json::Value;
 pub const RUNTIME_EVENT_ENVELOPE_SCHEMA_VERSION: u32 = 1;
 pub const RUNTIME_API_VERSION: &str = "1.0";
 
+/// Maximum JSON input (including base64 expansion) for an image turn.
+pub const MAX_RUNTIME_IMAGE_BODY_BYTES: usize = 8 * 1024 * 1024;
+pub const MAX_RUNTIME_IMAGES: usize = 10;
+pub const MAX_RUNTIME_IMAGE_BYTES: usize = 4 * 1024 * 1024;
+pub const MAX_RUNTIME_IMAGE_TOTAL_BYTES: usize = 5 * 1024 * 1024;
+
+/// Inline bytes only: neither host paths nor remote URLs confer attachment authority.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct RuntimeImageInput {
+    pub mime: String,
+    pub data_base64: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RuntimeEventEnvelope {
     #[serde(default = "default_runtime_event_envelope_schema_version")]
@@ -47,6 +61,9 @@ pub struct RuntimeCapabilities {
     /// `operation_key` and returns the original turn for exact retries.
     #[serde(default)]
     pub turn_operation_idempotency: bool,
+    /// Bounded inline image inputs, persisted and replayed with their turn.
+    #[serde(default)]
+    pub turn_image_inputs: bool,
     pub turn_steer: bool,
     pub turn_interrupt: bool,
     pub event_replay: bool,
@@ -377,6 +394,7 @@ mod tests {
             threads: true,
             turns: true,
             turn_operation_idempotency: true,
+            turn_image_inputs: true,
             turn_steer: true,
             turn_interrupt: true,
             event_replay: true,
@@ -400,6 +418,14 @@ mod tests {
         assert_eq!(obj.get("threads").unwrap(), &json!(true));
         assert_eq!(obj.get("account_session").unwrap(), &json!(true));
         assert_eq!(obj.get("turn_operation_idempotency").unwrap(), &json!(true));
+        assert_eq!(obj.get("turn_image_inputs").unwrap(), &json!(true));
+        let mut legacy = value.clone();
+        legacy.as_object_mut().unwrap().remove("turn_image_inputs");
+        assert!(
+            !serde_json::from_value::<RuntimeCapabilities>(legacy)
+                .unwrap()
+                .turn_image_inputs
+        );
         assert_eq!(obj.get("external_tools").unwrap(), &json!(false));
         assert!(obj.contains_key("worker_runtime"));
         assert_eq!(obj.get("fleet_run_create").unwrap(), &json!(true));

crates/tui/src/core/engine.rs

@@ -128,13 +128,13 @@ const MCP_REGISTRY_FIRST_INSTRUCTION_SOURCE: &str = "runtime:mcp-registry-first"
 const MCP_REGISTRY_FIRST_INSTRUCTION: &str = "## MCP Registry-first policy\n\nFor any task centered on a specialized capability, including media or document conversion, data transformation, browser automation, database or service access, or a developer utility, you must call `registry_sync` with a `query` describing that capability before `exec_shell`, `fetch_url`, code execution, local programs, custom code, or a manual implementation. It scores the local Registry snapshot host-side and returns at most eight matches; the full catalog never enters the conversation. Treat a returned server as a match when it plausibly covers the core capability; wording need not be exact. If any plausible match exists, you must call `start_registry_mcp_server` with its exact name and inspect its tools before considering a local alternative. If nothing matches, refine the query once; a still-empty refined result means every Registry entry is clearly irrelevant. An installed or familiar shell command is not a reason to skip Registry discovery. Use local tools directly only for ordinary repo-native work and simple file operations, or after the matching server fails to start.";
 const ISOLATED_CHAT_ENGINE_PROMPT: &str = "You are Codewhale Chat. Answer the user's request directly and conversationally. This isolated chat-only session has no local workspace, project, memory, skill, account, credential, path, runtime context, or tools.";
 
-fn sanitize_isolated_chat_attachments(mut text: String) -> String {
+pub(crate) fn sanitize_isolated_chat_attachments(mut text: String) -> String {
     let references = crate::tui::file_mention::media_attachment_references(&text);
     for reference in references.into_iter().rev() {
         let replacement = if text[reference.start_byte..reference.end_byte].ends_with('\n') {
-            "[Attachment omitted: Runtime Chat is text-only.]\n"
+            "[Attachment omitted: Runtime Chat cannot read local file references.]\n"
         } else {
-            "[Attachment omitted: Runtime Chat is text-only.]"
+            "[Attachment omitted: Runtime Chat cannot read local file references.]"
         };
         text.replace_range(reference.start_byte..reference.end_byte, replacement);
     }
@@ -2564,6 +2564,7 @@ impl Engine {
                 self.config.hook_executor.clone(),
                 self.config.verbosity.clone(),
                 UserInputProvenance::Runtime,
+                Vec::new(),
             )
             .await;
     }
@@ -2606,6 +2607,7 @@ impl Engine {
                 EngineRunInput::Operation(op) => match *op {
                     Op::SendMessage {
                         content,
+                        images,
                         mode,
                         route,
                         compaction,
@@ -2658,6 +2660,7 @@ impl Engine {
                             hook_executor,
                             verbosity,
                             provenance,
+                            images,
                         )
                         .await;
                     }
@@ -2763,6 +2766,7 @@ impl Engine {
                                 self.config.hook_executor.clone(),
                                 self.config.verbosity.clone(),
                                 UserInputProvenance::Runtime,
+                                Vec::new(),
                             )
                             .await;
                     }
@@ -3311,6 +3315,7 @@ impl Engine {
                             self.config.hook_executor.clone(),
                             self.config.verbosity.clone(),
                             UserInputProvenance::ExternalUser,
+                            Vec::new(),
                         )
                         .await;
                     }
@@ -3705,8 +3710,8 @@ impl Engine {
     /// Whether the model can *see* the result is decided per request, not
     /// here — see `image_attach::strip_images_when_unsupported`.
     fn user_content_blocks(&self, text: String) -> Vec<ContentBlock> {
-        // Managed Chat currently accepts text only. Treat attachment-marker
-        // syntax as an omitted attachment so an account prompt can never make
+        // Managed Chat accepts validated inline bytes, never host paths. Treat
+        // attachment-marker syntax as an omitted attachment so an account prompt can never make
         // this host read a local path or echo that host path to a provider.
         if self.api_config.runtime_chat_isolated {
             return vec![ContentBlock::Text {
@@ -3960,6 +3965,7 @@ impl Engine {
                 self.config.hook_executor.clone(),
                 self.config.verbosity.clone(),
                 UserInputProvenance::SubAgentHandoff,
+                Vec::new(),
             )
             .await;
         if !outcome.started() {
@@ -4741,7 +4747,29 @@ impl Engine {
         hook_executor: Option<std::sync::Arc<crate::hooks::HookExecutor>>,
         verbosity: Option<String>,
         provenance: UserInputProvenance,
+        images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
     ) -> SendMessageOutcome {
+        // All surfaces reuse the same bounded validator. Runtime already checks
+        // before admission; this also protects direct in-process operations.
+        let images = match crate::image_attach::prepare_stored_images(&images) {
+            Ok(images) => images,
+            Err(error) => {
+                let message = error.to_string();
+                let _ = self
+                    .tx_event
+                    .send(Event::error(ErrorEnvelope::new(
+                        crate::error_taxonomy::ErrorCategory::InvalidInput,
+                        crate::error_taxonomy::ErrorSeverity::Error,
+                        true,
+                        "image_input_invalid",
+                        message.clone(),
+                    )))
+                    .await;
+                return SendMessageOutcome::NotStarted {
+                    error: Some(message),
+                };
+            }
+        };
         let turn_control = self.begin_turn_control();
         let mut goal_objective = goal_objective;
         let mut goal_token_budget = goal_token_budget;
@@ -5220,7 +5248,7 @@ impl Engine {
         // Add the user message through the same explicit snapshot constructor
         // preview uses. Route limits and mode in resource metadata therefore
         // belong to this turn even when the previous route was different.
-        let user_msg = self.user_text_message_from_snapshot(
+        let mut user_msg = self.user_text_message_from_snapshot(
             content,
             &model,
             auto_model,
@@ -5235,6 +5263,12 @@ impl Engine {
                 policy_narrowing: self.last_policy_narrowing.as_ref(),
             },
         );
+        let image_index = if self.api_config.runtime_chat_isolated {
+            user_msg.content.len()
+        } else {
+            user_msg.content.len().saturating_sub(1)
+        };
+        user_msg.content.splice(image_index..image_index, images);
         self.session.add_message(user_msg);
 
         self.emit_session_updated().await;

crates/tui/src/core/engine/preview/tests.rs

@@ -979,6 +979,7 @@ async fn assert_preview_matches_first_wire_body(
             None,
             verbosity,
             UserInputProvenance::ExternalUser,
+            Vec::new(),
         )
         .await;
 
@@ -1944,6 +1945,7 @@ async fn provider_reported_usage_is_unavailable_until_a_response_reports_it() {
             None,
             None,
             UserInputProvenance::ExternalUser,
+            Vec::new(),
         )
         .await;
 

crates/tui/src/core/engine/tests.rs

@@ -869,6 +869,7 @@ async fn exact_turn_snapshot_restores_custom_endpoint_and_turn_receipt_after_bui
     handle
         .send(Op::SendMessage {
             content: "verify exact route".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: Box::new(
                 resolve_runtime_route(&config, ApiProvider::Custom, Some("local-model"))
@@ -1243,6 +1244,7 @@ async fn goal_continuation_preserves_goal_and_resolves_updated_authoritative_rou
     handle
         .send(Op::SendMessage {
             content: "first turn".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, "local-model"),
             compaction: Box::new(CompactionConfig::default()),
@@ -1523,6 +1525,7 @@ async fn saturated_mailbox_does_not_deadlock_goal_continuation_self_dispatch() {
     handle
         .send(Op::SendMessage {
             content: "start the saturated goal turn".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, "local-model"),
             compaction: Box::new(CompactionConfig::default()),
@@ -1652,6 +1655,7 @@ async fn queued_ordinary_turn_does_not_multiply_engine_goal_continuations() {
     let run_task = tokio::spawn(engine.run());
     let send_message = |content: &str| Op::SendMessage {
         content: content.to_string(),
+        images: Vec::new(),
         mode: AppMode::Agent,
         route: resolved_route_for_test(&config, "local-model"),
         compaction: Box::new(CompactionConfig::default()),
@@ -2850,6 +2854,7 @@ async fn cross_turn_token_budget_exhaustion_does_not_pause_goal() {
     handle
         .send(Op::SendMessage {
             content: "start budgeted goal".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -3308,6 +3313,7 @@ async fn explicit_natural_goal_activates_before_provider_request() {
     handle
         .send(Op::SendMessage {
             content: "hello - take over and make it your /goal to solve navier stokes".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, "local-model"),
             compaction: Box::new(CompactionConfig::default()),
@@ -3419,6 +3425,7 @@ async fn operate_goal_probe(mode: AppMode, prompt: &str) -> (Option<String>, boo
     handle
         .send(Op::SendMessage {
             content: prompt.to_string(),
+            images: Vec::new(),
             mode,
             route: resolved_route_for_test(&config, "local-model"),
             compaction: Box::new(CompactionConfig::default()),
@@ -3550,6 +3557,7 @@ async fn operate_contract_is_appended_once_and_an_existing_goal_is_never_replace
 
     let send = |content: &str, goal_objective: Option<String>, goal_status| Op::SendMessage {
         content: content.to_string(),
+        images: Vec::new(),
         mode: AppMode::Operate,
         route: resolved_route_for_test(&config, "local-model"),
         compaction: Box::new(CompactionConfig::default()),
@@ -4236,6 +4244,7 @@ async fn host_managed_engine_does_not_self_dispatch_goal_continuation() {
     handle
         .send(Op::SendMessage {
             content: "one host-owned turn".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, "local-model"),
             compaction: Box::new(CompactionConfig::default()),
@@ -4361,6 +4370,7 @@ async fn host_managed_engine_defers_idle_subagent_completion_to_explicit_turn()
     handle
         .send(Op::SendMessage {
             content: "claim the next turn".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, "local-model"),
             compaction: Box::new(CompactionConfig::default()),
@@ -6013,6 +6023,7 @@ fn active_goal_message_op(
 ) -> Op {
     Op::SendMessage {
         content: content.to_string(),
+        images: Vec::new(),
         mode: AppMode::Agent,
         route: resolved_route_for_test(config, "local-model"),
         compaction: Box::new(CompactionConfig::default()),
@@ -6050,6 +6061,7 @@ fn system_prompt_text(prompt: SystemPrompt) -> String {
 fn external_user_message_op(content: &str, mode: AppMode, config: &Config) -> Op {
     Op::SendMessage {
         content: content.to_string(),
+        images: Vec::new(),
         mode,
         route: resolved_route_for_test(config, crate::config::DEFAULT_TEXT_MODEL),
         compaction: Box::new(CompactionConfig::default()),
@@ -6076,6 +6088,7 @@ fn external_user_message_op(content: &str, mode: AppMode, config: &Config) -> Op
 fn auto_review_message_op(content: &str, config: &Config) -> Op {
     Op::SendMessage {
         content: content.to_string(),
+        images: Vec::new(),
         mode: AppMode::Agent,
         route: resolved_route_for_test(config, crate::config::DEFAULT_TEXT_MODEL),
         compaction: Box::new(CompactionConfig::default()),
@@ -12180,6 +12193,7 @@ async fn operate_model_shell_uses_normal_approval_and_workspace_sandbox() {
     handle
         .send(Op::SendMessage {
             content: "write the requested local fixture".to_string(),
+            images: Vec::new(),
             mode: AppMode::Operate,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -12335,6 +12349,7 @@ async fn full_access_subagent_handoff_keeps_model_shell_free_of_approval_prompts
     handle
         .send(Op::SendMessage {
             content: "continue from the completed child".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -12472,6 +12487,7 @@ async fn assert_full_access_model_tool_batch_is_blocked(
     handle
         .send(Op::SendMessage {
             content: "exercise the Full Access execution boundary".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -12678,6 +12694,7 @@ async fn assert_full_access_model_tool_batch_runs(
     handle
         .send(Op::SendMessage {
             content: "exercise the Full Access auto-approval boundary".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -12955,6 +12972,7 @@ async fn auto_review_auto_resolves_hallucinated_question_without_prompting() {
     handle
         .send(Op::SendMessage {
             content: "continue autonomously".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -13142,6 +13160,7 @@ async fn full_access_permission_allow_cannot_bypass_background_catastrophic_floo
     handle
         .send(Op::SendMessage {
             content: "please run a background shell".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -13283,6 +13302,7 @@ async fn yolo_mode_does_not_prompt_for_background_shell() {
     handle
         .send(Op::SendMessage {
             content: "please run a background shell".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -13420,6 +13440,7 @@ async fn yolo_mode_executes_publish_like_shell_without_prompt() {
     handle
         .send(Op::SendMessage {
             content: "please publish this crate".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -13561,6 +13582,7 @@ async fn yolo_mode_does_not_prompt_for_mcp_action() {
     handle
         .send(Op::SendMessage {
             content: "please open the PR".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&api_config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -19941,6 +19963,7 @@ async fn run_headless_turn_with_flaky_network(
     handle
         .send(Op::SendMessage {
             content: "solve the task".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -20067,6 +20090,7 @@ async fn terminal_output_limit_followed_by_stream_error_is_charged_and_not_retri
     handle
         .send(Op::SendMessage {
             content: "solve the task".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -20172,6 +20196,7 @@ async fn midstream_error_frame_stops_the_stream_and_drops_trailing_deltas() {
     handle
         .send(Op::SendMessage {
             content: "solve the task".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -20420,6 +20445,7 @@ async fn run_interactive_turn_with_flaky_network(
     handle
         .send(Op::SendMessage {
             content: "solve the task".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -20649,6 +20675,7 @@ async fn interactive_thinking_only_drop_preserves_nothing_and_never_claims_it_di
     handle
         .send(Op::SendMessage {
             content: "solve the task".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),
@@ -20897,6 +20924,7 @@ async fn run_reasoning_only_turn_with_reprompts(
     handle
         .send(Op::SendMessage {
             content: "solve the task".to_string(),
+            images: Vec::new(),
             mode: AppMode::Agent,
             route: resolved_route_for_test(&config, crate::config::DEFAULT_TEXT_MODEL),
             compaction: Box::new(CompactionConfig::default()),

crates/tui/src/core/ops.rs

@@ -109,6 +109,8 @@ pub enum Op {
     /// Send a message to the AI
     SendMessage {
         content: String,
+        /// Inline image bytes validated by Runtime admission; no file references.
+        images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
         mode: AppMode,
         /// Exact, structurally resolved route authority for this turn. The
         /// engine activates its client before mutating turn state; injected

crates/tui/src/core/protocol_parity.rs

@@ -931,6 +931,7 @@ pub fn op_to_protocol(op: &Op) -> wire_op::Op {
     match op {
         Op::SendMessage {
             content,
+            images,
             mode,
             route,
             compaction,
@@ -954,6 +955,7 @@ pub fn op_to_protocol(op: &Op) -> wire_op::Op {
             provenance,
         } => wire_op::Op::SendMessage {
             content: content.clone(),
+            images: images.clone(),
             mode: app_mode_str(*mode).to_string(),
             model: Some(route.model.clone()),
             model_provider: Some(route.identity.key.clone()),

crates/tui/src/exec_agent.rs

@@ -429,6 +429,7 @@ pub(crate) async fn run_exec_agent(
     engine_handle
         .send(Op::SendMessage {
             content: prompt.to_string(),
+            images: Vec::new(),
             mode,
             route: Box::new(validated_route.into_resolved()),
             compaction: Box::new(compaction.clone()),

crates/tui/src/image_attach.rs

@@ -42,6 +42,12 @@
 //! PNG/JPEG/GIF/WebP, and so, therefore, do we. Refusing a BMP here with a
 //! readable message beats letting one through to a provider-side 400.
 
+use anyhow::{Result, bail};
+use codewhale_protocol::runtime::{
+    MAX_RUNTIME_IMAGE_BYTES, MAX_RUNTIME_IMAGE_TOTAL_BYTES, MAX_RUNTIME_IMAGES, RuntimeImageInput,
+};
+use image::{DynamicImage, ImageReader, Limits};
+use std::io::Cursor;
 use std::path::Path;
 
 use base64::{Engine as _, engine::general_purpose::STANDARD};
@@ -57,6 +63,147 @@ use crate::models::{ContentBlock, ImageUrlContent};
 /// accepted it" verdict portable across routes.
 pub const MAX_IMAGE_BYTES: usize = 5 * 1024 * 1024;
 
+/// Maximum width or height admitted for an input image (8192 px).
+pub const MAX_IMAGE_DIMENSION: u32 = 8192;
+
+/// Maximum total pixels admitted before decoding is aborted (~33.5 megapixels).
+pub const MAX_IMAGE_PIXELS: u64 = 33_554_432;
+
+/// Memory allocation limit for image decoding (64 MiB).
+pub const MAX_DECODE_ALLOC_BYTES: u64 = 64 * 1024 * 1024;
+
+pub(crate) fn decode_and_guard_image(bytes: &[u8]) -> Result<(DynamicImage, u32, u32)> {
+    let limits = || {
+        let mut limits = Limits::default();
+        limits.max_alloc = Some(MAX_DECODE_ALLOC_BYTES);
+        limits.max_image_width = Some(MAX_IMAGE_DIMENSION);
+        limits.max_image_height = Some(MAX_IMAGE_DIMENSION);
+        limits
+    };
+    let mut reader = ImageReader::new(Cursor::new(bytes)).with_guessed_format()?;
+    reader.limits(limits());
+    let (width, height) = reader
+        .into_dimensions()
+        .map_err(|_| anyhow::anyhow!("invalid image header or decompression bomb guard"))?;
+    if u64::from(width) * u64::from(height) > MAX_IMAGE_PIXELS
+        || width > MAX_IMAGE_DIMENSION
+        || height > MAX_IMAGE_DIMENSION
+    {
+        bail!("image dimensions exceed the decompression bomb guard; downscale or crop first");
+    }
+    let mut reader = ImageReader::new(Cursor::new(bytes)).with_guessed_format()?;
+    reader.limits(limits());
+    let decoded = reader
+        .decode()
+        .map_err(|_| anyhow::anyhow!("invalid image content or decode allocation limit"))?;
+    Ok((decoded, width, height))
+}
+
+/// Validate untrusted inline input before route selection or durable admission.
+/// Return the existing provider-neutral history representation; no file is opened.
+pub(crate) fn prepare_runtime_images(images: &[RuntimeImageInput]) -> Result<Vec<ContentBlock>> {
+    if images.len() > MAX_RUNTIME_IMAGES {
+        bail!("images exceed the {MAX_RUNTIME_IMAGES} attachment limit");
+    }
+    prepare_images_with_limit(
+        images,
+        MAX_RUNTIME_IMAGE_BYTES,
+        Some(MAX_RUNTIME_IMAGE_TOTAL_BYTES),
+    )
+}
+
+/// Internal Engine/history input retains the established local 5 MiB ceiling.
+/// Network callers must first pass `prepare_runtime_images` (4 MiB per image,
+/// 10 images and 5 MiB total). Local history never had those aggregate/count
+/// limits; impose only its existing per-image bound and bounded full decode.
+pub(crate) fn prepare_stored_images(images: &[RuntimeImageInput]) -> Result<Vec<ContentBlock>> {
+    prepare_images_with_limit(images, MAX_IMAGE_BYTES, None)
+}
+
+fn prepare_images_with_limit(
+    images: &[RuntimeImageInput],
+    per_image_limit: usize,
+    total_limit: Option<usize>,
+) -> Result<Vec<ContentBlock>> {
+    let mut total = 0usize;
+    images
+        .iter()
+        .enumerate()
+        .map(|(index, image)| {
+            if image.data_base64.len() > per_image_limit.div_ceil(3) * 4 {
+                bail!(
+                    "image {} exceeds the {} MiB limit",
+                    index + 1,
+                    per_image_limit / (1024 * 1024)
+                );
+            }
+            let bytes = STANDARD
+                .decode(&image.data_base64)
+                .map_err(|_| anyhow::anyhow!("image {} has invalid base64", index + 1))?;
+            if bytes.len() > per_image_limit {
+                bail!(
+                    "image {} exceeds the {} MiB limit",
+                    index + 1,
+                    per_image_limit / (1024 * 1024)
+                );
+            }
+            total = total.saturating_add(bytes.len());
+            if total_limit.is_some_and(|limit| total > limit) {
+                bail!("images exceed the 5 MiB total limit");
+            }
+            let attached = encode_image_bytes(&bytes, &format!("image {}", index + 1))?;
+            if image.mime != attached.media_type {
+                bail!("image {} MIME does not match its content", index + 1);
+            }
+            decode_and_guard_image(&bytes)?;
+            // Standard padded base64 is the one replay representation.
+            if STANDARD.encode(&bytes) != image.data_base64 {
+                bail!("image {} base64 is not canonical", index + 1);
+            }
+            Ok(attached.content_block())
+        })
+        .collect()
+}
+
+/// Reuse durable canonical bytes for retry, never reread a path or URL.
+pub(crate) fn runtime_images_from_blocks(
+    blocks: &[ContentBlock],
+) -> Result<Vec<RuntimeImageInput>> {
+    let mut images = Vec::new();
+    for block in blocks {
+        if let ContentBlock::ImageUrl { image_url } = block {
+            if image_url.url.len() > MAX_IMAGE_BYTES.div_ceil(3) * 4 + 32 {
+                bail!("stored image exceeds the attachment limit");
+            }
+            let (mime, data) = parse_data_url(&image_url.url)
+                .ok_or_else(|| anyhow::anyhow!("stored image requires canonical inline content"))?;
+            images.push(RuntimeImageInput {
+                mime: mime.to_string(),
+                data_base64: data.to_string(),
+            });
+        }
+    }
+    prepare_stored_images(&images)?;
+    Ok(images)
+}
+
+/// Validate new image-bearing durable records without rewriting their block order.
+/// Legacy schema 2 history continues to use its original interpretation.
+pub(crate) fn validate_stored_image_content(blocks: &[ContentBlock]) -> Result<()> {
+    if blocks.iter().any(|block| {
+        !matches!(
+            block,
+            ContentBlock::Text { .. } | ContentBlock::ImageUrl { .. }
+        )
+    }) {
+        bail!("invalid persisted user image content kind");
+    }
+    if runtime_images_from_blocks(blocks)?.is_empty() {
+        bail!("persisted image input must contain an image");
+    }
+    Ok(())
+}
+
 /// Why a file could not be attached as an image.
 ///
 /// Every variant renders to a sentence naming the file and the reason. These

crates/tui/src/image_attach/tests.rs

@@ -457,3 +457,95 @@ fn attach_from_path_reports_an_unreadable_file() {
         "got {error:?}"
     );
 }
+
+pub(crate) fn runtime_image_fixture(color: u8) -> codewhale_protocol::runtime::RuntimeImageInput {
+    let image = image::DynamicImage::ImageRgba8(image::RgbaImage::from_pixel(
+        2,
+        2,
+        image::Rgba([color, 31, 99, 255]),
+    ));
+    let mut bytes = std::io::Cursor::new(Vec::new());
+    image.write_to(&mut bytes, image::ImageFormat::Png).unwrap();
+    codewhale_protocol::runtime::RuntimeImageInput {
+        mime: "image/png".into(),
+        data_base64: STANDARD.encode(bytes.into_inner()),
+    }
+}
+
+#[test]
+fn runtime_image_validation_preserves_exact_bytes_and_rejects_corruption() {
+    let input = runtime_image_fixture(7);
+    let blocks = prepare_runtime_images(std::slice::from_ref(&input)).unwrap();
+    assert_eq!(
+        runtime_images_from_blocks(&blocks).unwrap(),
+        [input.clone()]
+    );
+    for bad in [
+        codewhale_protocol::runtime::RuntimeImageInput {
+            mime: "image/jpeg".into(),
+            ..input.clone()
+        },
+        codewhale_protocol::runtime::RuntimeImageInput {
+            data_base64: "not base64".into(),
+            ..input.clone()
+        },
+        codewhale_protocol::runtime::RuntimeImageInput {
+            data_base64: String::new(),
+            ..input.clone()
+        },
+        // Existing signature sniffing alone accepted this truncated PNG.
+        codewhale_protocol::runtime::RuntimeImageInput {
+            data_base64: STANDARD.encode(b"\x89PNG\r\n\x1a\n"),
+            ..input.clone()
+        },
+    ] {
+        assert!(prepare_runtime_images(&[bad]).is_err());
+    }
+}
+
+#[test]
+fn runtime_image_validation_bounds_count_encoded_size_and_decode_dimensions() {
+    let input = runtime_image_fixture(7);
+    assert!(prepare_runtime_images(&vec![input.clone(); 11]).is_err());
+    assert!(
+        prepare_runtime_images(&[codewhale_protocol::runtime::RuntimeImageInput {
+            data_base64: "A".repeat(MAX_IMAGE_BYTES.div_ceil(3) * 4 + 1),
+            ..input
+        }])
+        .is_err()
+    );
+    let wide = image::DynamicImage::ImageRgba8(image::RgbaImage::new(MAX_IMAGE_DIMENSION + 1, 1));
+    let mut bytes = std::io::Cursor::new(Vec::new());
+    wide.write_to(&mut bytes, image::ImageFormat::Png).unwrap();
+    let input = codewhale_protocol::runtime::RuntimeImageInput {
+        mime: "image/png".into(),
+        data_base64: STANDARD.encode(bytes.into_inner()),
+    };
+    assert!(prepare_runtime_images(&[input]).is_err());
+}
+
+pub(crate) fn runtime_image_fixture_bytes(
+    size: usize,
+) -> codewhale_protocol::runtime::RuntimeImageInput {
+    let mut image = runtime_image_fixture(42);
+    let mut bytes = STANDARD.decode(&image.data_base64).unwrap();
+    bytes.resize(size, 0);
+    image.data_base64 = STANDARD.encode(bytes);
+    image
+}
+
+#[test]
+fn runtime_image_network_four_mib_and_historical_five_mib_bounds_are_distinct() {
+    let at_network_limit = runtime_image_fixture_bytes(MAX_RUNTIME_IMAGE_BYTES);
+    assert!(prepare_runtime_images(&[at_network_limit]).is_ok());
+    let historical = runtime_image_fixture_bytes(MAX_RUNTIME_IMAGE_BYTES + 1);
+    assert!(prepare_runtime_images(std::slice::from_ref(&historical)).is_err());
+    let stored = prepare_stored_images(std::slice::from_ref(&historical)).unwrap();
+    assert_eq!(runtime_images_from_blocks(&stored).unwrap(), [historical]);
+    assert!(prepare_stored_images(&[runtime_image_fixture_bytes(MAX_IMAGE_BYTES)]).is_ok());
+    assert!(prepare_stored_images(&[runtime_image_fixture_bytes(MAX_IMAGE_BYTES + 1)]).is_err());
+    let three_mib = runtime_image_fixture_bytes(3 * 1024 * 1024);
+    assert!(prepare_runtime_images(&[three_mib.clone(), three_mib.clone()]).is_err());
+    assert!(prepare_stored_images(&[three_mib.clone(), three_mib]).is_ok());
+    assert!(prepare_stored_images(&vec![runtime_image_fixture(1); 11]).is_ok());
+}

crates/tui/src/remote_control.rs

@@ -4404,6 +4404,9 @@ fn parse_remote_command(value: &Value, expected_run_id: &str) -> Result<RemoteCo
     }
     match value.get("type").and_then(Value::as_str) {
         Some("prompt.request") => {
+            if value.get("images").is_some() && value.get("runtimeBindingId").is_none() {
+                return Err("Image input is unavailable for legacy remote Work; use native Runtime or Runtime Chat.".to_string());
+            }
             let exact_legacy_prompt = value.as_object().is_some_and(|record| {
                 record.len() == 4
                     && ["type", "runId", "turnId", "prompt"]
@@ -4585,7 +4588,12 @@ async fn runner_request(
             None => format!("The remote-control server rejected a request ({status})."),
         });
     }
-    read_bounded_json(response).await
+    let limit = if segments.last() == Some(&"commands") {
+        codewhale_protocol::runtime::MAX_RUNTIME_IMAGE_BODY_BYTES
+    } else {
+        MAX_RESPONSE_BYTES
+    };
+    read_bounded_json_with_limit(response, limit).await
 }
 
 async fn public_request(
@@ -4656,18 +4664,29 @@ fn sanitized_rejection_excerpt(body: &[u8]) -> Option<String> {
 }
 
 async fn read_bounded_json(response: reqwest::Response) -> Result<Value, String> {
+    read_bounded_json_with_limit(response, MAX_RESPONSE_BYTES).await
+}
+
+async fn read_bounded_json_with_limit(
+    mut response: reqwest::Response,
+    limit: usize,
+) -> Result<Value, String> {
     if response
         .content_length()
-        .is_some_and(|length| length > MAX_RESPONSE_BYTES as u64)
+        .is_some_and(|length| length > limit as u64)
     {
         return Err("Codewhale returned an oversized remote-control response.".to_string());
     }
-    let bytes = response
-        .bytes()
+    let mut bytes = Vec::new();
+    while let Some(chunk) = response
+        .chunk()
         .await
-        .map_err(|_| "Codewhale returned an unreadable response.".to_string())?;
-    if bytes.len() > MAX_RESPONSE_BYTES {
-        return Err("Codewhale returned an oversized remote-control response.".to_string());
+        .map_err(|_| "Codewhale returned an unreadable response.".to_string())?
+    {
+        if bytes.len().saturating_add(chunk.len()) > limit {
+            return Err("Codewhale returned an oversized remote-control response.".to_string());
+        }
+        bytes.extend_from_slice(&chunk);
     }
     serde_json::from_slice(&bytes)
         .map_err(|_| "Codewhale returned an invalid remote-control response.".to_string())
@@ -8629,4 +8648,59 @@ mod tests {
         };
         assert_eq!(envelopes[0]["payload"]["delta"], "again");
     }
+    #[test]
+    fn runtime_image_legacy_work_never_downgrades_to_text() {
+        let image = crate::image_attach::tests::runtime_image_fixture(1);
+        let command = json!({"type":"prompt.request","runId":"run_fixture","turnId":"turn_fixture","prompt":"look","images":[image]});
+        assert!(
+            parse_remote_command(&command, "run_fixture")
+                .unwrap_err()
+                .contains("legacy remote Work")
+        );
+        let mut text = command;
+        text.as_object_mut().unwrap().remove("images");
+        assert!(matches!(
+            parse_remote_command(&text, "run_fixture").unwrap(),
+            RemoteCommand::Prompt { .. }
+        ));
+    }
+
+    #[tokio::test]
+    async fn runtime_image_command_response_has_bounded_larger_budget() {
+        use axum::{Router, routing::get};
+        let payload = serde_json::to_string(
+            &json!({"commands":[],"fixture": "x".repeat(MAX_RESPONSE_BYTES + 1)}),
+        )
+        .unwrap();
+        let app = Router::new().route(
+            "/",
+            get(move || {
+                let payload = payload.clone();
+                async move { payload }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        let client = crate::tls::reqwest_client();
+        let url = format!("http://{addr}/");
+        assert!(
+            read_bounded_json(client.get(&url).send().await.unwrap())
+                .await
+                .is_err()
+        );
+        let parsed = read_bounded_json_with_limit(
+            client.get(&url).send().await.unwrap(),
+            codewhale_protocol::runtime::MAX_RUNTIME_IMAGE_BODY_BYTES,
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            parsed["fixture"].as_str().unwrap().len(),
+            MAX_RESPONSE_BYTES + 1
+        );
+        server.abort();
+    }
 }

crates/tui/src/runtime_api.rs

@@ -10,7 +10,7 @@ use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow, bail};
 use async_stream::stream;
-use axum::extract::{ConnectInfo, Path, Query, Request, State};
+use axum::extract::{ConnectInfo, DefaultBodyLimit, Path, Query, Request, State};
 use axum::http::header;
 use axum::http::{HeaderName, HeaderValue, Method, StatusCode};
 use axum::middleware;
@@ -264,6 +264,8 @@ impl Default for RuntimeApiOptions {
 #[derive(Debug, Deserialize)]
 struct StreamTurnRequest {
     prompt: String,
+    #[serde(default)]
+    images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
     model: Option<String>,
     mode: Option<String>,
     permission_posture: Option<String>,
@@ -539,6 +541,7 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
         threads: true,
         turns: true,
         turn_operation_idempotency: true,
+        turn_image_inputs: true,
         turn_steer: true,
         turn_interrupt: true,
         event_replay: true,
@@ -1115,7 +1118,12 @@ pub fn build_router(state: RuntimeApiState) -> Router {
             "/v1/fleet/workers/{worker_id}/restart",
             post(restart_fleet_worker),
         )
-        .route("/v1/stream", post(stream_turn))
+        .route(
+            "/v1/stream",
+            post(stream_turn).layer(DefaultBodyLimit::max(
+                codewhale_protocol::runtime::MAX_RUNTIME_IMAGE_BODY_BYTES,
+            )),
+        )
         .route("/v1/threads", get(list_threads).post(create_thread))
         .route("/v1/threads/summary", get(list_threads_summary))
         .route("/v1/threads/{id}", get(get_thread).patch(update_thread))
@@ -1124,7 +1132,12 @@ pub fn build_router(state: RuntimeApiState) -> Router {
         .route("/v1/threads/{id}/undo", post(undo_thread_turn))
         .route("/v1/threads/{id}/patch-undo", post(patch_undo_thread_turn))
         .route("/v1/threads/{id}/retry", post(retry_thread_turn))
-        .route("/v1/threads/{id}/turns", post(start_thread_turn))
+        .route(
+            "/v1/threads/{id}/turns",
+            post(start_thread_turn).layer(DefaultBodyLimit::max(
+                codewhale_protocol::runtime::MAX_RUNTIME_IMAGE_BODY_BYTES,
+            )),
+        )
         .route(
             "/v1/threads/{id}/turns/{turn_id}/steer",
             post(steer_thread_turn),
@@ -4420,6 +4433,8 @@ struct UndoTurnResponse {
     /// The original user message text from the first dropped turn,
     /// so the GUI can pre-populate the input box.
     original_user_text: Option<String>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    original_user_images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
 }
 
 async fn undo_thread_turn(
@@ -4428,7 +4443,7 @@ async fn undo_thread_turn(
     Json(req): Json<UndoTurnRequest>,
 ) -> Result<(StatusCode, Json<UndoTurnResponse>), ApiError> {
     let depth = req.depth.unwrap_or(0);
-    let (forked_thread, original_user_text) = state
+    let (forked_thread, original_user_text, original_user_images) = state
         .runtime_threads
         .fork_at_user_message(&id, depth)
         .await
@@ -4438,6 +4453,7 @@ async fn undo_thread_turn(
         Json(UndoTurnResponse {
             thread: forked_thread,
             original_user_text,
+            original_user_images,
         }),
     ))
 }
@@ -4462,6 +4478,8 @@ struct PatchUndoResponse {
     thread: ThreadRecord,
     /// The original user text from the removed turn (for re-editing).
     original_user_text: Option<String>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    original_user_images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
 }
 
 async fn patch_undo_thread_turn(
@@ -4480,7 +4498,7 @@ async fn patch_undo_thread_turn(
     let patch_result = patch_undo_workspace_files(&thread.workspace, thread.session_id.as_deref());
 
     // Step 2: Remove the last conversation turn (undo_conversation).
-    let (forked_thread, original_user_text) = state
+    let (forked_thread, original_user_text, original_user_images) = state
         .runtime_threads
         .fork_at_user_message(&id, depth)
         .await
@@ -4492,6 +4510,7 @@ async fn patch_undo_thread_turn(
             patch_result,
             thread: forked_thread,
             original_user_text,
+            original_user_images,
         }),
     ))
 }
@@ -4611,7 +4630,7 @@ async fn retry_thread_turn(
     Json(req): Json<RetryTurnRequest>,
 ) -> Result<(StatusCode, Json<RetryTurnResponse>), ApiError> {
     let depth = req.depth.unwrap_or(0);
-    let (forked_thread, original_user_text) = state
+    let (forked_thread, original_user_text, original_user_images) = state
         .runtime_threads
         .fork_at_user_message(&id, depth)
         .await
@@ -4626,10 +4645,11 @@ async fn retry_thread_turn(
 
     let turn = state
         .runtime_threads
-        .start_turn(
+        .start_turn_from_stored_images(
             &forked_thread.id,
             StartTurnRequest {
                 prompt: retry_prompt,
+                images: original_user_images,
                 operation_key: None,
                 input_summary: None,
                 model: None,
@@ -5249,6 +5269,8 @@ async fn stream_turn(
         return Err(ApiError::bad_request("prompt is required"));
     }
 
+    crate::image_attach::prepare_runtime_images(&req.images).map_err(map_thread_err)?;
+
     let model = runtime_request_model(&state.config.read(), req.model.as_deref())?;
     let workspace = req
         .workspace
@@ -5292,12 +5314,13 @@ async fn stream_turn(
             .map_err(|_| ApiError::internal("Compatibility stream test hook dropped resume"))?;
     }
 
-    let turn = state
+    let turn_result = state
         .runtime_threads
         .start_turn(
             &thread.id,
             StartTurnRequest {
                 prompt,
+                images: req.images,
                 input_summary: None,
                 model: Some(model.clone()),
                 mode: Some(mode.clone()),
@@ -5308,8 +5331,20 @@ async fn stream_turn(
                 ..Default::default()
             },
         )
-        .await
-        .map_err(|e| ApiError::internal(format!("Failed to start stream turn: {e}")))?;
+        .await;
+    let turn = match turn_result {
+        Ok(turn) => turn,
+        Err(error) => {
+            // This helper refuses loaded threads and any thread owning a turn.
+            // A failed/uncertain handoff must remain recoverable; only an empty,
+            // never-loaded admission can be discarded.
+            if let Err(cleanup_error) = state.runtime_threads.discard_empty_thread(&thread.id).await
+            {
+                tracing::warn!(thread_id = %thread.id, %cleanup_error, "Retained stream thread after failed admission");
+            }
+            return Err(map_thread_err(error));
+        }
+    };
 
     // Subscribe before reading the durable replay. Events produced while the
     // replay is loaded then exist in at least one source, and the sequence
@@ -6045,6 +6080,8 @@ struct ProviderModelEntry {
 #[derive(Debug, Clone, Serialize)]
 struct ProviderModelsResponse {
     provider: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    model_provider_id: Option<String>,
     models: Vec<ProviderModelEntry>,
     total: usize,
     #[serde(rename = "nextCursor", skip_serializing_if = "Option::is_none")]
@@ -6064,6 +6101,8 @@ struct ProviderModelsCursor {
     provider: String,
     filter: String,
     catalog_fingerprint: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    route_fingerprint: Option<String>,
     offset: usize,
 }
 
@@ -6119,6 +6158,7 @@ fn paginate_provider_models(
     provider: &str,
     mut models: Vec<ProviderModelEntry>,
     params: &ListProviderModelsParams,
+    route_fingerprint: Option<String>,
 ) -> Result<ProviderModelsResponse, ApiError> {
     let filter = normalized_provider_model_filter(params.filter.as_deref())?;
     let limit = params.limit.unwrap_or(DEFAULT_PROVIDER_MODELS_PAGE_SIZE);
@@ -6156,9 +6196,12 @@ fn paginate_provider_models(
         .collect::<String>();
     let start = if let Some(encoded) = params.cursor.as_deref() {
         let cursor = decode_provider_models_cursor(encoded)?;
-        if cursor.provider != provider || cursor.filter != filter {
+        if cursor.provider != provider
+            || cursor.filter != filter
+            || cursor.route_fingerprint != route_fingerprint
+        {
             return Err(ApiError::bad_request(
-                "Provider model cursor does not match this provider and filter",
+                "Provider model cursor does not match this provider, configured route, and filter",
             ));
         }
         if cursor.catalog_fingerprint != catalog_fingerprint {
@@ -6182,6 +6225,7 @@ fn paginate_provider_models(
             provider: provider.to_string(),
             filter,
             catalog_fingerprint,
+            route_fingerprint,
             offset: end,
         })?)
     } else {
@@ -6190,6 +6234,7 @@ fn paginate_provider_models(
 
     Ok(ProviderModelsResponse {
         provider: provider.to_string(),
+        model_provider_id: params.model_provider_id.clone(),
         models: page,
         total,
         next_cursor,
@@ -6401,6 +6446,7 @@ pub(crate) fn runtime_chat_relay_catalog(
                 "relay_chat_v1": true,
                 "isolated_chat_threads": true,
                 "turn_operation_idempotency": true,
+                "turn_image_inputs": true,
                 "tool_execution": false,
                 "stable_event_ids": true,
             },
@@ -6412,10 +6458,7 @@ pub(crate) fn runtime_chat_relay_catalog(
             "defaultModel": default_model,
             "credentialState": credential_state,
             "models": models.into_iter().map(|model| json!({
-                // Runtime Chat commands currently carry text only. Provider
-                // vision support is not relay capability truth until bounded
-                // image bytes are part of this protocol.
-                "imageInput": "unsupported",
+                "imageInput": provider_model_image_input_for_api(config, provider, &model),
                 "id": model,
             })).collect::<Vec<_>>(),
         }],
@@ -6457,8 +6500,11 @@ async fn list_providers(
     Ok(Json(ProvidersResponse { current, providers }))
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Default, Deserialize)]
 struct ListProviderModelsParams {
+    /// Exact configured provider identity; omission retains the legacy projection.
+    #[serde(default)]
+    model_provider_id: Option<String>,
     /// Optional case-insensitive substring filter applied before pagination.
     #[serde(default)]
     filter: Option<String>,
@@ -6475,8 +6521,7 @@ async fn list_provider_models(
     Path(id): Path<String>,
     Query(params): Query<ListProviderModelsParams>,
 ) -> Result<Json<ProviderModelsResponse>, ApiError> {
-    let config = state.config.read().clone();
-    let active_provider = config.api_provider();
+    let mut config = state.config.read().clone();
     let api_provider = ApiProvider::parse(&id)
         .ok_or_else(|| ApiError::bad_request(format!("Unknown provider id '{id}'")))?;
     // Reject requests for the legacy deepseek-cn alias that has no
@@ -6486,14 +6531,46 @@ async fn list_provider_models(
             "provider 'deepseek-cn' is a legacy alias; use 'deepseek' instead",
         ));
     }
-    let models = provider_models_for_api(&config, active_provider, api_provider)
+    let route_fingerprint = if let Some(exact_id) = params.model_provider_id.as_deref() {
+        if exact_id.is_empty()
+            || exact_id != exact_id.trim()
+            || exact_id.chars().any(char::is_control)
+        {
+            return Err(ApiError::bad_request(
+                "model_provider_id must be an exact configured identity",
+            ));
+        }
+        let identity = config
+            .resolve_persisted_provider_identity(Some(api_provider.as_str()), Some(exact_id))
+            .map_err(ApiError::bad_request)?;
+        if identity.provider != api_provider || identity.persisted_id() != Some(exact_id) {
+            return Err(ApiError::bad_request(
+                "model_provider_id does not match this provider route",
+            ));
+        }
+        config.scope_to_provider_identity(&identity);
+        // Do not expose the endpoint in an opaque cursor. Its hash binds even
+        // identical catalogs under distinct named routes or a changed base URL.
+        let route = serde_json::to_vec(&(
+            api_provider.as_str(),
+            exact_id,
+            config.base_url_for_route_identity(api_provider, &identity.key),
+        ))
+        .map_err(|error| {
+            ApiError::internal(format!("Could not fingerprint provider route: {error}"))
+        })?;
+        Some(crate::hashing::sha256_hex(route))
+    } else {
+        None
+    };
+    let models = provider_models_for_api(&config, config.api_provider(), api_provider)
         .into_iter()
         .map(|id| ProviderModelEntry {
             image_input: provider_model_image_input_for_api(&config, api_provider, &id),
             id,
         })
         .collect();
-    paginate_provider_models(api_provider.as_str(), models, &params).map(Json)
+    paginate_provider_models(api_provider.as_str(), models, &params, route_fingerprint).map(Json)
 }
 
 /// Request body for `POST /v1/providers/{id}/switch`.

crates/tui/src/runtime_api/sessions.rs

@@ -297,6 +297,19 @@ pub(super) async fn resume_session_thread(
         .load_session(&id)
         .map_err(|e| map_session_err(&id, e, "read"))?;
 
+    // Validate imported image bytes before allocating a Runtime thread. This
+    // retains local history's existing bounds; invalid content cannot leave an
+    // empty session, and no path or remote image reference is dereferenced.
+    for message in session
+        .messages
+        .iter()
+        .filter(|message| message.role == Role::User)
+    {
+        crate::image_attach::runtime_images_from_blocks(&message.content).map_err(|error| {
+            ApiError::bad_request(format!("Cannot restore session image: {error}"))
+        })?;
+    }
+
     let model = req.model.unwrap_or_else(|| session.metadata.model.clone());
     let mode = req.mode.unwrap_or_else(|| {
         session

crates/tui/src/runtime_api/tests.rs

@@ -7868,7 +7868,9 @@ fn provider_model_catalog_paginates_all_six_hundred_rows_without_truncation() {
                 filter: None,
                 cursor,
                 limit: Some(MAX_PROVIDER_MODELS_PAGE_SIZE),
+                ..Default::default()
             },
+            None,
         )
         .expect("page should be valid");
         page_count += 1;
@@ -7913,7 +7915,9 @@ fn provider_model_catalog_applies_filter_before_cursor_and_rejects_cross_scope_r
             filter: Some(" ALPHA ".to_string()),
             cursor: None,
             limit: Some(1),
+            ..Default::default()
         },
+        None,
     )
     .expect("filtered first page should be valid");
     assert_eq!(first.total, 2);
@@ -7927,7 +7931,9 @@ fn provider_model_catalog_applies_filter_before_cursor_and_rejects_cross_scope_r
             filter: Some("alpha".to_string()),
             cursor: Some(cursor.clone()),
             limit: Some(1),
+            ..Default::default()
         },
+        None,
     )
     .expect("matching filter should continue");
     assert_eq!(second.models[0].id, "alpha-two");
@@ -7940,7 +7946,9 @@ fn provider_model_catalog_applies_filter_before_cursor_and_rejects_cross_scope_r
             filter: Some("alpha".to_string()),
             cursor: Some(cursor),
             limit: Some(1),
+            ..Default::default()
         },
+        None,
     );
     assert!(replay.is_err(), "a cursor cannot cross provider ownership");
 }
@@ -7961,7 +7969,9 @@ fn provider_model_cursor_rejects_catalog_change_between_pages() {
             filter: None,
             cursor: None,
             limit: Some(1),
+            ..Default::default()
         },
+        None,
     )
     .expect("first page");
     let cursor = first.next_cursor.expect("second page remains");
@@ -7978,7 +7988,9 @@ fn provider_model_cursor_rejects_catalog_change_between_pages() {
                 filter: None,
                 cursor: Some(cursor.clone()),
                 limit: Some(1),
-            }
+                ..Default::default()
+            },
+            None,
         )
         .is_err(),
         "an insertion before the cursor must force a restart, not disappear"
@@ -7993,7 +8005,9 @@ fn provider_model_cursor_rejects_catalog_change_between_pages() {
                 filter: None,
                 cursor: Some(cursor),
                 limit: Some(1),
-            }
+                ..Default::default()
+            },
+            None,
         )
         .is_err(),
         "capability changes also invalidate a catalog snapshot"
@@ -8017,7 +8031,9 @@ fn provider_model_cursor_round_trips_multibyte_filter_at_the_allowed_limit() {
             filter: Some(filter.clone()),
             cursor: None,
             limit: Some(1),
+            ..Default::default()
         },
+        None,
     )
     .expect("first page");
     let second = paginate_provider_models(
@@ -8027,7 +8043,9 @@ fn provider_model_cursor_round_trips_multibyte_filter_at_the_allowed_limit() {
             filter: Some(filter),
             cursor: first.next_cursor,
             limit: Some(1),
+            ..Default::default()
         },
+        None,
     )
     .expect("all emitted cursors must be accepted, including multibyte filters");
     assert_eq!(second.models.len(), 1);
@@ -12221,3 +12239,405 @@ async fn api_config_reports_local_default_availability_without_blocking_config_r
     crate::provider_lake::clear_live_snapshot();
     Ok(())
 }
+
+#[tokio::test]
+async fn runtime_image_http_rejects_before_dispatch_and_accepts_large_canonical_bytes() -> Result<()>
+{
+    use crate::image_attach::tests::runtime_image_fixture;
+    use base64::Engine as _;
+    let _env = crate::test_support::lock_test_env();
+    let dir = tempfile::tempdir()?;
+    let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
+    let mut config = Config {
+        provider: Some("deepseek".into()),
+        default_text_model: Some("deepseek-v4-flash-vision-exp".into()),
+        api_key: Some("synthetic-image-key".into()),
+        ..Default::default()
+    };
+    config.set_provider_model_override(
+        ApiProvider::Deepseek,
+        Some("deepseek-v4-flash-vision-exp".into()),
+    );
+    let (addr, manager, server) = spawn_test_server_with_root_token_mobile_workspace_and_overrides(
+        dir.path().to_path_buf(),
+        dir.path().join("sessions"),
+        None,
+        false,
+        dir.path().join("workspace"),
+        TestServerOverrides {
+            config: Some(config),
+            ..Default::default()
+        },
+    )
+    .await?
+    .context("loopback listener required for image contract proof")?;
+    let thread = manager.create_thread(Default::default()).await?;
+    let mut harness = crate::core::engine::mock_engine_handle();
+    manager
+        .install_test_engine(&thread.id, harness.handle.clone())
+        .await?;
+    let client = crate::tls::reqwest_client();
+    let url = format!("http://{addr}/v1/threads/{}/turns", thread.id);
+    let good = runtime_image_fixture(7);
+    for body in [
+        json!({"prompt":"look", "images":[{"mime":"image/png", "dataBase64":"garbage"}]}),
+        json!({"prompt":"", "images":[good.clone()]}),
+        json!({"prompt":"look", "model":"auto", "images":[good.clone()]}),
+        json!({"prompt":"look", "model":"deepseek-v4-flash", "images":[good.clone()]}),
+        json!({"prompt":"look", "images":[{"mime":"image/png", "dataBase64":good.data_base64, "path":"/private/host-only"}]}),
+    ] {
+        let response = client.post(&url).json(&body).send().await?;
+        assert!(response.status().is_client_error());
+        assert!(
+            harness.rx_op.try_recv().is_err(),
+            "rejection must not dispatch any Engine op"
+        );
+        assert!(
+            manager
+                .get_thread_detail(&thread.id)
+                .await?
+                .turns
+                .is_empty()
+        );
+    }
+    // An actual incompressible PNG crosses the former 2 MiB HTTP ceiling.
+    let mut random = 0x4a11_3317_u32;
+    let bytes: Vec<u8> = (0..768 * 768 * 4)
+        .map(|_| {
+            random ^= random << 13;
+            random ^= random >> 17;
+            random ^= random << 5;
+            random as u8
+        })
+        .collect();
+    let image =
+        image::DynamicImage::ImageRgba8(image::RgbaImage::from_raw(768, 768, bytes).unwrap());
+    let mut encoded = std::io::Cursor::new(Vec::new());
+    image.write_to(&mut encoded, image::ImageFormat::Png)?;
+    let large = codewhale_protocol::runtime::RuntimeImageInput {
+        mime: "image/png".into(),
+        data_base64: base64::engine::general_purpose::STANDARD.encode(encoded.into_inner()),
+    };
+    let body = json!({"prompt":"compare screenshot", "operation_key":"http-image-once", "images":[large.clone(),good.clone()]});
+    assert!(serde_json::to_vec(&body)?.len() > 2 * 1024 * 1024);
+    let response = client.post(&url).json(&body).send().await?;
+    assert_eq!(response.status(), StatusCode::CREATED);
+    let accepted: Value = response.json().await?;
+    let Op::SendMessage { images, .. } =
+        harness.rx_op.recv().await.context("accepted Engine op")?
+    else {
+        bail!("expected SendMessage");
+    };
+    assert_eq!(images, [large, good]);
+    let replay: Value = client
+        .post(&url)
+        .json(&body)
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(replay["turn"]["id"], accepted["turn"]["id"]);
+    assert!(harness.rx_op.try_recv().is_err());
+    harness
+        .tx_event
+        .send(EngineEvent::TurnStarted {
+            turn_id: "image-http-fixture".into(),
+            created_at: Utc::now(),
+            route: None,
+        })
+        .await?;
+    harness
+        .tx_event
+        .send(EngineEvent::TurnComplete {
+            usage: Default::default(),
+            parent_route_usage: Default::default(),
+            routed_usage_dropped_records: 0,
+            status: crate::core::events::TurnOutcomeStatus::Completed,
+            error: None,
+            tool_catalog: None,
+            base_url: None,
+        })
+        .await?;
+    server.abort();
+    Ok(())
+}
+
+#[tokio::test]
+async fn runtime_image_stream_rejection_does_not_leave_empty_threads() -> Result<()> {
+    use crate::image_attach::tests::{runtime_image_fixture, runtime_image_fixture_bytes};
+    let _env = crate::test_support::lock_test_env();
+    let dir = tempfile::tempdir()?;
+    let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
+    let mut config = Config {
+        provider: Some("deepseek".into()),
+        default_text_model: Some("deepseek-v4-flash-vision-exp".into()),
+        api_key: Some("synthetic-image-key".into()),
+        ..Default::default()
+    };
+    config.set_provider_model_override(
+        ApiProvider::Deepseek,
+        Some("deepseek-v4-flash-vision-exp".into()),
+    );
+    let (addr, manager, server) = spawn_test_server_with_root_token_mobile_workspace_and_overrides(
+        dir.path().to_path_buf(),
+        dir.path().join("sessions"),
+        None,
+        false,
+        dir.path().join("workspace"),
+        TestServerOverrides {
+            config: Some(config),
+            ..Default::default()
+        },
+    )
+    .await?
+    .context("loopback listener required")?;
+    let client = crate::tls::reqwest_client();
+    let good = runtime_image_fixture(3);
+    for body in [
+        json!({"prompt":"look", "images":[{"mime":"image/png","dataBase64":"garbage"}]}),
+        json!({"prompt":"", "images":[good.clone()]}),
+        json!({"prompt":"look", "model":"auto", "images":[good.clone()]}),
+        json!({"prompt":"look", "model":"deepseek-v4-flash", "images":[good]}),
+        json!({"prompt":"look", "images":[runtime_image_fixture_bytes(4 * 1024 * 1024 + 1)]}),
+    ] {
+        let response = client
+            .post(format!("http://{addr}/v1/stream"))
+            .json(&body)
+            .send()
+            .await?;
+        assert!(response.status().is_client_error(), "{}", response.status());
+        assert!(
+            manager
+                .list_threads(
+                    crate::runtime_threads::ThreadListFilter::IncludeArchived,
+                    None
+                )
+                .await?
+                .is_empty(),
+            "failed admission left an empty thread"
+        );
+    }
+    // Cleanup refuses a loaded or accepted session instead of erasing uncertain work.
+    let retained = manager.create_thread(Default::default()).await?;
+    let harness = crate::core::engine::mock_engine_handle();
+    manager
+        .install_test_engine(&retained.id, harness.handle.clone())
+        .await?;
+    assert!(manager.discard_empty_thread(&retained.id).await.is_err());
+    assert!(manager.get_thread(&retained.id).await.is_ok());
+    server.abort();
+    Ok(())
+}
+
+#[tokio::test]
+async fn runtime_image_named_catalog_resolves_and_echoes_exact_configured_identity() -> Result<()> {
+    let _env = crate::test_support::lock_test_env();
+    let dir = tempfile::tempdir()?;
+    let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
+    let route = crate::config::ProviderConfig {
+        kind: Some("openai-compatible".into()),
+        base_url: Some("http://127.0.0.1:9/v1".into()),
+        model: Some("same-model".into()),
+        ..Default::default()
+    };
+    let config = Config {
+        provider: Some("vision-personal".into()),
+        providers: Some(crate::config::ProvidersConfig {
+            custom: std::collections::HashMap::from([
+                ("vision-personal".into(), route.clone()),
+                ("vision-work".into(), route),
+            ]),
+            ..Default::default()
+        }),
+        ..Default::default()
+    };
+    let (addr, _, server) = spawn_test_server_with_root_token_mobile_workspace_and_overrides(
+        dir.path().to_path_buf(),
+        dir.path().join("sessions"),
+        None,
+        false,
+        dir.path().join("workspace"),
+        TestServerOverrides {
+            config: Some(config),
+            ..Default::default()
+        },
+    )
+    .await?
+    .context("loopback listener required")?;
+    let client = crate::tls::reqwest_client();
+    for identity in ["vision-personal", "vision-work"] {
+        let response: Value = client
+            .get({
+                let mut url =
+                    reqwest::Url::parse(&format!("http://{addr}/v1/providers/custom/models"))?;
+                url.query_pairs_mut()
+                    .append_pair("model_provider_id", identity);
+                url
+            })
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        assert_eq!(response["model_provider_id"], identity);
+        assert_eq!(response["provider"], "custom");
+        assert_eq!(response["models"][0]["id"], "same-model");
+        assert_eq!(response["models"][0]["image_input"], "unknown");
+    }
+    for (kind, identity) in [
+        ("custom", "missing"),
+        ("deepseek", "vision-work"),
+        ("custom", " vision-work"),
+    ] {
+        assert_eq!(
+            client
+                .get({
+                    let mut url =
+                        reqwest::Url::parse(&format!("http://{addr}/v1/providers/{kind}/models"))?;
+                    url.query_pairs_mut()
+                        .append_pair("model_provider_id", identity);
+                    url
+                })
+                .send()
+                .await?
+                .status(),
+            StatusCode::BAD_REQUEST
+        );
+    }
+    let legacy: Value = client
+        .get(format!("http://{addr}/v1/providers/custom/models"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert!(legacy.get("model_provider_id").is_none());
+    server.abort();
+    Ok(())
+}
+
+#[test]
+fn runtime_image_named_catalog_cursor_binds_identity_endpoint_and_catalog() -> Result<()> {
+    let models = ["one", "two"]
+        .into_iter()
+        .map(|id| ProviderModelEntry {
+            id: id.into(),
+            image_input: codewhale_config::route::CapabilityState::Supported,
+        })
+        .collect::<Vec<_>>();
+    let fingerprint = |id: &str, base: &str| {
+        crate::hashing::sha256_hex(serde_json::to_vec(&("custom", id, base)).unwrap())
+    };
+    let params = ListProviderModelsParams {
+        model_provider_id: Some("vision-work".into()),
+        limit: Some(1),
+        ..Default::default()
+    };
+    let route = fingerprint("vision-work", "http://127.0.0.1:9/v1");
+    let first = paginate_provider_models("custom", models.clone(), &params, Some(route.clone()))
+        .map_err(|e| anyhow::anyhow!(e.message))?;
+    let mut next = params;
+    next.cursor = first.next_cursor;
+    assert!(paginate_provider_models("custom", models.clone(), &next, Some(route)).is_ok());
+    for changed in [
+        Some(fingerprint("vision-personal", "http://127.0.0.1:9/v1")),
+        Some(fingerprint("vision-work", "http://127.0.0.1:10/v1")),
+        None,
+    ] {
+        assert!(paginate_provider_models("custom", models.clone(), &next, changed).is_err());
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn runtime_image_saved_session_import_validates_before_creating_a_thread() -> Result<()> {
+    use crate::models::{ContentBlock, ImageUrlContent};
+    let _env = crate::test_support::lock_test_env();
+    let dir = tempfile::tempdir()?;
+    let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
+    let sessions_dir = dir.path().join("sessions");
+    fs::create_dir_all(&sessions_dir)?;
+    let (addr, manager, server) =
+        spawn_test_server_with_root(dir.path().to_path_buf(), sessions_dir.clone())
+            .await?
+            .context("loopback listener required")?;
+    let client = crate::tls::reqwest_client();
+    let expected =
+        vec![crate::image_attach::tests::runtime_image_fixture_bytes(3 * 1024 * 1024); 2];
+    let historical = crate::image_attach::prepare_stored_images(&expected)?;
+    let cases = [
+        vec![ContentBlock::ImageUrl {
+            image_url: ImageUrlContent {
+                url: "https://host.invalid/never-fetch".into(),
+            },
+        }],
+        vec![ContentBlock::ImageUrl {
+            image_url: ImageUrlContent {
+                url: "data:image/png;base64,iVBORw0KGgo=".into(),
+            },
+        }],
+        historical,
+    ];
+    for (index, mut content) in cases.into_iter().enumerate() {
+        content.insert(
+            0,
+            ContentBlock::Text {
+                text: "saved image prompt".into(),
+                cache_control: None,
+            },
+        );
+        let id = format!("sess_image_import_{index}");
+        let session = json!({
+            "schema_version": 1,
+            "metadata": { "id": id, "title": "Image import fixture", "created_at": "2026-09-08T00:00:00Z", "updated_at": "2026-09-08T00:00:00Z", "message_count": 1, "total_tokens": 0, "model": "deepseek-v4-pro", "model_provider": "deepseek", "workspace": dir.path(), "mode": "agent" },
+            "messages": [{ "role": "user", "content": content }], "system_prompt": null,
+        });
+        fs::write(
+            sessions_dir.join(format!("{id}.json")),
+            serde_json::to_vec(&session)?,
+        )?;
+        let response = client
+            .post(format!("http://{addr}/v1/sessions/{id}/resume-thread"))
+            .json(&json!({}))
+            .send()
+            .await?;
+        if index < 2 {
+            assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+            assert!(
+                manager
+                    .list_threads(
+                        crate::runtime_threads::ThreadListFilter::IncludeArchived,
+                        None
+                    )
+                    .await?
+                    .is_empty()
+            );
+        } else {
+            assert_eq!(response.status(), StatusCode::CREATED);
+            let response: Value = response.json().await?;
+            let detail = manager
+                .get_thread_detail(
+                    response["thread_id"]
+                        .as_str()
+                        .context("imported thread id")?,
+                )
+                .await?;
+            let item = detail
+                .items
+                .iter()
+                .find(|item| item.kind == crate::runtime_threads::TurnItemKind::UserMessage)
+                .context("imported user item")?;
+            let blocks: Vec<ContentBlock> = serde_json::from_value(
+                item.metadata.as_ref().context("image metadata")?["runtime_image_content"].clone(),
+            )?;
+            assert_eq!(
+                crate::image_attach::runtime_images_from_blocks(&blocks)?,
+                expected
+            );
+            assert_eq!(detail.thread.schema_version, 3);
+        }
+    }
+    server.abort();
+    Ok(())
+}

crates/tui/src/runtime_chat_relay.rs

@@ -79,6 +79,8 @@ pub(crate) struct RuntimeChatPrompt {
     pub runtime_binding_id: String,
     pub runtime_thread_id: String,
     pub prompt: String,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub system_prompt: Option<String>,
     pub model: String,
@@ -759,6 +761,7 @@ impl RuntimeChatRelayHost {
                 &binding.native_thread_id,
                 StartTurnRequest {
                     prompt: command.prompt.clone(),
+                    images: command.images.clone(),
                     operation_key: Some(command.operation_key.clone()),
                     input_summary: None,
                     model: Some(command.model.clone()),
@@ -989,6 +992,9 @@ impl RuntimeChatRelayHost {
                 .is_some_and(|models| {
                     models.iter().any(|model| {
                         model.get("id").and_then(Value::as_str) == Some(command.model.as_str())
+                            && (command.images.is_empty()
+                                || model.get("imageInput").and_then(Value::as_str)
+                                    == Some("supported"))
                     })
                 });
         if !route_matches {
@@ -1170,6 +1176,14 @@ impl RuntimeChatRelayHost {
 
 impl RuntimeChatPrompt {
     pub(crate) fn validate_shape(&self) -> Result<(), String> {
+        crate::image_attach::prepare_runtime_images(&self.images)
+            .map_err(|error| error.to_string())?;
+        if !self.images.is_empty() && self.model.trim().eq_ignore_ascii_case("auto") {
+            return Err(
+                "Image inputs require an exact named model; Auto is unavailable for images."
+                    .to_string(),
+            );
+        }
         if self.command_type != "prompt.request" {
             return Err("Codewhale sent an unsupported Runtime Chat command.".to_string());
         }
@@ -1857,6 +1871,7 @@ mod tests {
     #[test]
     fn chat_command_shape_requires_empty_tools_and_exact_chat_modes() {
         let mut prompt = RuntimeChatPrompt {
+            images: Vec::new(),
             command_type: "prompt.request".to_string(),
             run_id: "run_fixture".to_string(),
             turn_id: format!("local_turn_{}", "b".repeat(24)),
@@ -1905,6 +1920,7 @@ mod tests {
             .unwrap();
         host.authorize_run("run_fixture").unwrap();
         let prompt = RuntimeChatPrompt {
+            images: Vec::new(),
             command_type: "prompt.request".to_string(),
             run_id: "run_fixture".to_string(),
             turn_id: format!("local_turn_{}", "e".repeat(24)),
@@ -1970,6 +1986,7 @@ mod tests {
             .unwrap_or_else(|| provider.as_str())
             .to_string();
         let prompt = RuntimeChatPrompt {
+            images: Vec::new(),
             command_type: "prompt.request".to_string(),
             run_id: "run_fixture".to_string(),
             turn_id: format!("local_turn_{}", "8".repeat(24)),
@@ -2035,7 +2052,11 @@ mod tests {
         assert_eq!(catalog["providers"].as_array().unwrap().len(), 1);
         assert_eq!(
             catalog["providers"][0]["models"][0]["imageInput"],
-            "unsupported"
+            "unknown"
+        );
+        assert_eq!(
+            catalog["runtime"]["capabilities"]["turn_image_inputs"],
+            true
         );
         let serialized = catalog.to_string();
         assert!(!serialized.contains("must-not-cross"));
@@ -2482,4 +2503,31 @@ mod tests {
         assert!(!reopened.has_unsettled_authorized_turns());
         reopened.authorize_run("run_other").unwrap();
     }
+    #[test]
+    fn runtime_image_relay_hash_preserves_text_and_binds_order() {
+        let legacy = json!({"type":"prompt.request","runId":"run_fixture","turnId":format!("local_turn_{}", "b".repeat(24)),"operationKey":"operation-1","runtimeBindingId":"binding_fixture","runtimeThreadId":format!("local_thread_{}", "a".repeat(24)),"prompt":"look","model":"deepseek-v4-flash-vision-exp","modelProvider":"deepseek","modelProviderId":"deepseek","allowedTools":[],"mode":"chat","requestedMode":"chat","workspace":{"id":"workspace_fixture","targetRef":"target_fixture"}});
+        let mut command: RuntimeChatPrompt = serde_json::from_value(legacy.clone()).unwrap();
+        command.validate_shape().unwrap();
+        let expected = hex_digest(Sha256::digest(
+            serde_json::to_vec(&canonical_json_value(&legacy)).unwrap(),
+        ));
+        assert_eq!(
+            runtime_chat_request_fingerprint(&command).unwrap(),
+            expected
+        );
+        let mut with_empty = legacy;
+        with_empty["images"] = json!([]);
+        let empty: RuntimeChatPrompt = serde_json::from_value(with_empty).unwrap();
+        assert_eq!(runtime_chat_request_fingerprint(&empty).unwrap(), expected);
+        command.images = vec![
+            crate::image_attach::tests::runtime_image_fixture(1),
+            crate::image_attach::tests::runtime_image_fixture(2),
+        ];
+        command.validate_shape().unwrap();
+        let first = runtime_chat_request_fingerprint(&command).unwrap();
+        command.images.reverse();
+        assert_ne!(runtime_chat_request_fingerprint(&command).unwrap(), first);
+        command.model = "auto".into();
+        assert!(command.validate_shape().is_err());
+    }
 }

crates/tui/src/runtime_threads.rs

@@ -490,7 +490,11 @@ fn sort_turn_items_by_start(items: &mut [TurnItemRecord]) {
 /// Bumped to 2 for v0.6.6 after live engine semantics changed. The persisted
 /// thread/turn/item records did not change shape, but a v1 reader on a v2
 /// session should still fail closed rather than silently mis-replay.
+// Text-only writes retain v2. Image-bearing records require v3 so an older
+// binary refuses recovery instead of silently dropping accepted attachments.
 const CURRENT_RUNTIME_SCHEMA_VERSION: u32 = 2;
+const IMAGE_RUNTIME_SCHEMA_VERSION: u32 = 3;
+const MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION: u32 = IMAGE_RUNTIME_SCHEMA_VERSION;
 
 fn is_zero_u64(value: &u64) -> bool {
     *value == 0
@@ -1299,6 +1303,44 @@ pub struct TurnItemRecord {
     pub ended_at: Option<DateTime<Utc>>,
 }
 
+impl TurnItemRecord {
+    fn set_image_content(&mut self, content: Vec<ContentBlock>) {
+        self.schema_version = IMAGE_RUNTIME_SCHEMA_VERSION;
+        let metadata = self.metadata.get_or_insert_with(|| json!({}));
+        metadata["runtime_image_content"] = json!(content);
+    }
+
+    fn user_content(&self) -> Result<Vec<ContentBlock>> {
+        if self.schema_version == IMAGE_RUNTIME_SCHEMA_VERSION
+            && self.kind == TurnItemKind::UserMessage
+        {
+            let content = self
+                .metadata
+                .as_ref()
+                .and_then(|meta| meta.get("runtime_image_content"))
+                .context("persisted image input is missing its content")?;
+            let blocks: Vec<ContentBlock> =
+                serde_json::from_value(content.clone()).context("invalid persisted image input")?;
+            crate::image_attach::validate_stored_image_content(&blocks)?;
+            if self.detail.as_deref().is_some_and(|text| !text.trim().is_empty())
+                && !blocks.iter().any(|block| matches!(block, ContentBlock::Text { text, .. } if !text.trim().is_empty()))
+            {
+                bail!("persisted image input is missing its prompt");
+            }
+            return Ok(blocks);
+        }
+        let text = self.detail.as_ref().unwrap_or(&self.summary);
+        Ok(if text.trim().is_empty() {
+            Vec::new()
+        } else {
+            vec![ContentBlock::Text {
+                text: text.clone(),
+                cache_control: None,
+            }]
+        })
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RuntimeEventRecord {
     #[serde(default = "default_runtime_schema_version")]
@@ -2118,11 +2160,11 @@ impl RuntimeThreadStore {
                 &path,
             )
         })?;
-        if record.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
+        if record.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
             bail!(
                 "Thread schema v{} is newer than supported v{}",
                 record.schema_version,
-                CURRENT_RUNTIME_SCHEMA_VERSION
+                MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION
             );
         }
         Ok(record)
@@ -2146,11 +2188,11 @@ impl RuntimeThreadStore {
                 &path,
             )
         })?;
-        if record.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
+        if record.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
             bail!(
                 "Turn schema v{} is newer than supported v{}",
                 record.schema_version,
-                CURRENT_RUNTIME_SCHEMA_VERSION
+                MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION
             );
         }
         Ok(record)
@@ -2174,13 +2216,18 @@ impl RuntimeThreadStore {
                 &path,
             )
         })?;
-        if record.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
+        if record.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
             bail!(
                 "Item schema v{} is newer than supported v{}",
                 record.schema_version,
-                CURRENT_RUNTIME_SCHEMA_VERSION
+                MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION
             );
         }
+        if record.schema_version == IMAGE_RUNTIME_SCHEMA_VERSION
+            && record.kind == TurnItemKind::UserMessage
+        {
+            record.user_content()?;
+        }
         Ok(record)
     }
 
@@ -2199,11 +2246,11 @@ impl RuntimeThreadStore {
                 .with_context(|| format!("Failed to read {}", path.display()))?;
             let thread: ThreadRecord = serde_json::from_str(&raw)
                 .with_context(|| format!("Failed to parse {}", path.display()))?;
-            if thread.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
+            if thread.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
                 bail!(
                     "Thread schema v{} is newer than supported v{}",
                     thread.schema_version,
-                    CURRENT_RUNTIME_SCHEMA_VERSION
+                    MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION
                 );
             }
             out.push(thread);
@@ -2240,11 +2287,11 @@ impl RuntimeThreadStore {
                 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
             let turn: TurnRecord = serde_json::from_str(&raw)
                 .with_context(|| format!("Failed to parse {}", path.display()))?;
-            if turn.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
+            if turn.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
                 bail!(
                     "Turn schema v{} is newer than supported v{}",
                     turn.schema_version,
-                    CURRENT_RUNTIME_SCHEMA_VERSION
+                    MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION
                 );
             }
             out.push(turn);
@@ -2288,13 +2335,18 @@ impl RuntimeThreadStore {
                     &path,
                 )
             })?;
-            if item.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
+            if item.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
                 bail!(
                     "Item schema v{} is newer than supported v{}",
                     item.schema_version,
-                    CURRENT_RUNTIME_SCHEMA_VERSION
+                    MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION
                 );
             }
+            if item.schema_version == IMAGE_RUNTIME_SCHEMA_VERSION
+                && item.kind == TurnItemKind::UserMessage
+            {
+                item.user_content()?;
+            }
             if item.turn_id == turn_id {
                 out.push(item);
             }
@@ -2349,13 +2401,18 @@ impl RuntimeThreadStore {
                     &path,
                 )
             })?;
-            if item.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
+            if item.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
                 bail!(
                     "Item schema v{} is newer than supported v{}",
                     item.schema_version,
-                    CURRENT_RUNTIME_SCHEMA_VERSION
+                    MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION
                 );
             }
+            if item.schema_version == IMAGE_RUNTIME_SCHEMA_VERSION
+                && item.kind == TurnItemKind::UserMessage
+            {
+                item.user_content()?;
+            }
             if wanted.contains(item.turn_id.as_str()) {
                 out.entry(item.turn_id.clone()).or_default().push(item);
             }
@@ -2901,6 +2958,8 @@ pub struct UpdateThreadRequest {
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct StartTurnRequest {
     pub prompt: String,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
     /// Optional caller-supplied idempotency key, scoped to this Runtime store
     /// and thread. The raw key is validated but never persisted.
     #[serde(default, alias = "operationKey")]
@@ -3018,8 +3077,9 @@ fn runtime_turn_request_fingerprint(
     trust_mode: bool,
     dynamic_tools: &[DynamicToolSpec],
     environment_id: Option<&str>,
+    images: &[codewhale_protocol::runtime::RuntimeImageInput],
 ) -> Result<String> {
-    let payload = json!({
+    let mut payload = json!({
         "version": 1,
         "thread_id": thread.id,
         "provider": thread.model_provider,
@@ -3039,6 +3099,10 @@ fn runtime_turn_request_fingerprint(
         "workspace": thread.workspace,
         "system_prompt": thread.system_prompt,
     });
+    // Preserve the exact historical text-only fingerprint payload.
+    if !images.is_empty() {
+        payload["images"] = json!(images);
+    }
     Ok(crate::hashing::sha256_hex(crate::client::canonical_json(
         &payload,
     )))
@@ -4051,6 +4115,7 @@ enum SeedItem {
 /// A turn being assembled from session messages.
 struct TurnSeed {
     user_text: String,
+    image_content: Vec<ContentBlock>,
     items: Vec<SeedItem>,
 }
 
@@ -5218,6 +5283,7 @@ impl RuntimeThreadManager {
     ) -> Result<TurnRecord> {
         let req = StartTurnRequest {
             prompt,
+            images: Vec::new(),
             operation_key: None,
             input_summary: Some(if continuation_index == 0 {
                 "goal kickoff".to_string()
@@ -5240,6 +5306,7 @@ impl RuntimeThreadManager {
             req,
             RuntimeTurnInputSource::GoalContinuation { continuation_index },
             None,
+            false,
         )
         .await
     }
@@ -5674,6 +5741,7 @@ impl RuntimeThreadManager {
                 thread_id,
                 StartTurnRequest {
                     prompt,
+                    images: Vec::new(),
                     operation_key: None,
                     input_summary: Some(input_summary),
                     model: None,
@@ -5692,6 +5760,7 @@ impl RuntimeThreadManager {
                     persisted_summary: claimed.summary.clone(),
                 },
                 None,
+                false,
             )
             .await;
 
@@ -7051,7 +7120,11 @@ impl RuntimeThreadManager {
         &self,
         id: &str,
         depth_from_tail: usize,
-    ) -> Result<(ThreadRecord, Option<String>)> {
+    ) -> Result<(
+        ThreadRecord,
+        Option<String>,
+        Vec<codewhale_protocol::runtime::RuntimeImageInput>,
+    )> {
         let source = self.get_thread(id).await?;
         let source_turns = self.store.list_turns_for_thread(&source.id)?;
 
@@ -7087,6 +7160,15 @@ impl RuntimeThreadManager {
             .iter()
             .find(|item| item.kind == TurnItemKind::UserMessage)
             .and_then(|item| item.detail.clone());
+        let original_images = target_items
+            .iter()
+            .find(|item| item.kind == TurnItemKind::UserMessage)
+            .map(|item| {
+                item.user_content()
+                    .and_then(|content| crate::image_attach::runtime_images_from_blocks(&content))
+            })
+            .transpose()?
+            .unwrap_or_default();
 
         // Copy turns strictly before `target_turn_idx` into a new thread.
         // Mirrors `fork_thread` but stops at the cutoff instead of copying
@@ -7166,7 +7248,7 @@ impl RuntimeThreadManager {
             }),
         )
         .await?;
-        Ok((forked, original_user_text))
+        Ok((forked, original_user_text, original_images))
     }
 
     /// Persist cloned records before publishing their thread. Until the final
@@ -7264,6 +7346,24 @@ impl RuntimeThreadManager {
                 "user" => {
                     let mut user_text = String::new();
                     let mut tool_results = Vec::new();
+                    let image_content: Vec<_> = if msg
+                        .content
+                        .iter()
+                        .any(|block| matches!(block, ContentBlock::ImageUrl { .. }))
+                    {
+                        msg.content
+                            .iter()
+                            .filter(|block| {
+                                matches!(
+                                    block,
+                                    ContentBlock::Text { .. } | ContentBlock::ImageUrl { .. }
+                                )
+                            })
+                            .cloned()
+                            .collect()
+                    } else {
+                        Vec::new()
+                    };
 
                     for block in &msg.content {
                         match block {
@@ -7292,7 +7392,7 @@ impl RuntimeThreadManager {
                         }
                     }
 
-                    if !user_text.is_empty() {
+                    if !user_text.is_empty() || !image_content.is_empty() {
                         // A real user prompt begins a new turn. Tool results
                         // without text belong to the preceding assistant turn.
                         if let Some(t) = current_turn.take() {
@@ -7300,11 +7400,13 @@ impl RuntimeThreadManager {
                         }
                         current_turn = Some(TurnSeed {
                             user_text,
+                            image_content,
                             items: tool_results,
                         });
                     } else if !tool_results.is_empty() {
                         let turn = current_turn.get_or_insert_with(|| TurnSeed {
                             user_text: String::new(),
+                            image_content: Vec::new(),
                             items: Vec::new(),
                         });
                         turn.items.extend(tool_results);
@@ -7314,6 +7416,7 @@ impl RuntimeThreadManager {
                         }
                         current_turn = Some(TurnSeed {
                             user_text: String::new(),
+                            image_content: Vec::new(),
                             items: Vec::new(),
                         });
                     }
@@ -7323,6 +7426,7 @@ impl RuntimeThreadManager {
                     // an assistant message), create a placeholder turn.
                     let turn = current_turn.get_or_insert_with(|| TurnSeed {
                         user_text: String::new(),
+                        image_content: Vec::new(),
                         items: Vec::new(),
                     });
                     for block in &msg.content {
@@ -7367,6 +7471,14 @@ impl RuntimeThreadManager {
             turns.push(t);
         }
 
+        // Validate the entire import before the first durable write. Saved local
+        // images keep their 5 MiB ceiling; no path or remote URL is dereferenced.
+        for turn_seed in &turns {
+            if !turn_seed.image_content.is_empty() {
+                crate::image_attach::validate_stored_image_content(&turn_seed.image_content)?;
+            }
+        }
+
         for turn_seed in turns {
             let turn_at = next_seed_stamp();
             let turn_id = format!("turn_{}", &Uuid::new_v4().to_string()[..8]);
@@ -7375,10 +7487,10 @@ impl RuntimeThreadManager {
             let mut item_ids = Vec::new();
 
             // Save user message item.
-            if !turn_seed.user_text.is_empty() {
+            if !turn_seed.user_text.is_empty() || !turn_seed.image_content.is_empty() {
                 let item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
                 let item_at = next_seed_stamp();
-                self.store.save_item(&TurnItemRecord {
+                let mut item = TurnItemRecord {
                     schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
                     id: item_id.clone(),
                     turn_id: turn_id.clone(),
@@ -7390,7 +7502,12 @@ impl RuntimeThreadManager {
                     artifact_refs: Vec::new(),
                     started_at: Some(item_at),
                     ended_at: Some(item_at),
-                })?;
+                };
+                if !turn_seed.image_content.is_empty() {
+                    item.set_image_content(turn_seed.image_content.clone());
+                    thread.schema_version = IMAGE_RUNTIME_SCHEMA_VERSION;
+                }
+                self.store.save_item(&item)?;
                 item_ids.push(item_id);
             }
 
@@ -7519,7 +7636,11 @@ impl RuntimeThreadManager {
             // Only create a turn if there's content.
             if !item_ids.is_empty() {
                 self.store.save_turn(&TurnRecord {
-                    schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
+                    schema_version: if turn_seed.image_content.is_empty() {
+                        CURRENT_RUNTIME_SCHEMA_VERSION
+                    } else {
+                        IMAGE_RUNTIME_SCHEMA_VERSION
+                    },
                     id: turn_id.clone(),
                     thread_id: thread_id.to_string(),
                     status: RuntimeTurnStatus::Completed,
@@ -8124,6 +8245,25 @@ impl RuntimeThreadManager {
             req,
             RuntimeTurnInputSource::ExternalUser,
             reserved_turn_id,
+            false,
+        )
+        .await
+    }
+
+    /// Retry only bytes recovered from validated durable history. This internal
+    /// entry point changes the historical byte ceiling, never model or policy
+    /// admission, and cannot be selected by a network request field.
+    pub(crate) async fn start_turn_from_stored_images(
+        &self,
+        thread_id: &str,
+        req: StartTurnRequest,
+    ) -> Result<TurnRecord> {
+        self.start_turn_with_source(
+            thread_id,
+            req,
+            RuntimeTurnInputSource::ExternalUser,
+            None,
+            true,
         )
         .await
     }
@@ -8134,6 +8274,7 @@ impl RuntimeThreadManager {
         req: StartTurnRequest,
         input_source: RuntimeTurnInputSource,
         reserved_turn_id: Option<&str>,
+        stored_image_bytes: bool,
     ) -> Result<TurnRecord> {
         // Heap-allocate the turn-start state machine. Its future holds two full
         // Config clones plus ThreadRecord/EngineHandle/TurnRecord/TurnItemRecord
@@ -8150,6 +8291,11 @@ impl RuntimeThreadManager {
         // engine handoff, so a completed reload is a hard boundary: no later
         // dispatch can carry its predecessor's URL, key, model, or policy.
         let _config_admission = self.config_admission.read().await;
+        let image_blocks = if stored_image_bytes {
+            crate::image_attach::prepare_stored_images(&req.images)?
+        } else {
+            crate::image_attach::prepare_runtime_images(&req.images)?
+        };
         let prompt = req.prompt.trim().to_string();
         if prompt.is_empty() {
             bail!("prompt is required");
@@ -8186,6 +8332,12 @@ impl RuntimeThreadManager {
         let mode = policy.mode;
         let requested_model = req.model.as_deref().unwrap_or(&thread.model).to_string();
         let auto_model = requested_model.trim().eq_ignore_ascii_case("auto");
+        if !image_blocks.is_empty() && (requested_model.is_empty() || requested_model.trim() != requested_model) {
+            bail!("image inputs require an exact nonempty named model");
+        }
+        if !image_blocks.is_empty() && auto_model {
+            bail!("image inputs require an exact named model with supported image input; Auto is unavailable for images");
+        }
         let cfg_snapshot = self.config.read().clone();
         let configured_reasoning_preference = cfg_snapshot
             .reasoning_effort()
@@ -8216,6 +8368,7 @@ impl RuntimeThreadManager {
                 trust_mode,
                 &req.dynamic_tools,
                 req.environment_id.as_deref(),
+                &req.images,
             )?;
             self.prepare_runtime_turn_operation(
                 thread_id,
@@ -8231,6 +8384,13 @@ impl RuntimeThreadManager {
         {
             return Ok(original_turn);
         }
+        if !image_blocks.is_empty() {
+            let identity = self.provider_identity_for_thread(&cfg_snapshot, &thread)?;
+            let route = resolve_runtime_thread_route_for_identity(&cfg_snapshot, &identity, Some(&requested_model))?;
+            if route.candidate.capabilities().image_input != codewhale_config::route::CapabilityState::Supported {
+                bail!("image inputs require a model with explicitly supported image input");
+            }
+        }
         let engine = self.ensure_engine_loaded(&thread).await?;
 
         let client_preflight_required = {
@@ -8425,7 +8585,7 @@ impl RuntimeThreadManager {
         // adding TurnComplete at settlement would count the same gap twice.
 
         let user_item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
-        let user_item = TurnItemRecord {
+        let mut user_item = TurnItemRecord {
             schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
             id: user_item_id.clone(),
             turn_id: turn_id.clone(),
@@ -8438,6 +8598,12 @@ impl RuntimeThreadManager {
             started_at: Some(now),
             ended_at: Some(now),
         };
+        if !image_blocks.is_empty() {
+            let mut content = vec![ContentBlock::Text { text: if cfg_snapshot.runtime_chat_isolated { crate::core::engine::sanitize_isolated_chat_attachments(prompt.clone()) } else { prompt.clone() }, cache_control: None }];
+            content.extend(image_blocks.iter().cloned());
+            user_item.set_image_content(content);
+            turn.schema_version = IMAGE_RUNTIME_SCHEMA_VERSION;
+        }
         turn.item_ids.push(user_item_id.clone());
 
         // Every turn carries the persisted goal alongside its message. The
@@ -8464,6 +8630,7 @@ impl RuntimeThreadManager {
 
         let op = Op::SendMessage {
             content: prompt,
+            images: req.images,
             mode,
             route: Box::new(route),
             compaction: Box::new(compaction),
@@ -8558,6 +8725,7 @@ impl RuntimeThreadManager {
                 }
                 self.store.save_item(&user_item)?;
                 self.store.save_turn(&turn)?;
+                current_thread.schema_version = current_thread.schema_version.max(turn.schema_version);
                 current_thread.latest_turn_id = Some(turn_id.clone());
                 current_thread.updated_at = now;
                 self.store.save_thread(&current_thread)
@@ -9479,13 +9647,7 @@ impl RuntimeThreadManager {
                 match item.kind {
                     TurnItemKind::UserMessage => {
                         flush_assistant(&mut assistant_blocks, &mut messages);
-                        let text = item.detail.unwrap_or(item.summary);
-                        if !text.trim().is_empty() {
-                            user_blocks.push(ContentBlock::Text {
-                                text,
-                                cache_control: None,
-                            });
-                        }
+                        user_blocks.extend(item.user_content()?);
                     }
                     TurnItemKind::AgentMessage => {
                         flush_user(&mut user_blocks, &mut messages);

crates/tui/src/runtime_threads/tests.rs

@@ -201,9 +201,9 @@ mod recovery {
         let thread = manager.get_thread(&thread.id).await?;
         let fork = manager.fork_thread(&thread.id).await?;
         assert_eq!(manager.restore_thread_messages(&fork)?, messages);
-        let (backtrack, _) = manager.fork_at_user_message(&thread.id, 0).await?;
+        let (backtrack, _, _) = manager.fork_at_user_message(&thread.id, 0).await?;
         assert_eq!(manager.restore_thread_messages(&backtrack)?, messages[..2]);
-        let (empty, _) = manager.fork_at_user_message(&thread.id, 1).await?;
+        let (empty, _, _) = manager.fork_at_user_message(&thread.id, 1).await?;
         assert!(manager.restore_thread_messages(&empty)?.is_empty());
         assert_eq!(manager.restore_thread_messages(&thread)?, messages);
         let mut legacy = thread.clone();
@@ -1540,6 +1540,7 @@ fn runtime_turn_operation_keys_are_bounded_scoped_and_fingerprint_execution_poli
         false,
         &[],
         None,
+        &[],
     )?;
 
     let mut different_provider = thread.clone();
@@ -1558,6 +1559,7 @@ fn runtime_turn_operation_keys_are_bounded_scoped_and_fingerprint_execution_poli
             false,
             &[],
             None,
+            &[],
         )?,
         runtime_turn_request_fingerprint(
             &thread,
@@ -1571,6 +1573,7 @@ fn runtime_turn_operation_keys_are_bounded_scoped_and_fingerprint_execution_poli
             false,
             &[],
             None,
+            &[],
         )?,
         runtime_turn_request_fingerprint(
             &thread,
@@ -1584,6 +1587,7 @@ fn runtime_turn_operation_keys_are_bounded_scoped_and_fingerprint_execution_poli
             false,
             &[],
             None,
+            &[],
         )?,
         runtime_turn_request_fingerprint(
             &thread,
@@ -1597,6 +1601,7 @@ fn runtime_turn_operation_keys_are_bounded_scoped_and_fingerprint_execution_poli
             false,
             &[],
             None,
+            &[],
         )?,
         runtime_turn_request_fingerprint(
             &thread,
@@ -1610,6 +1615,7 @@ fn runtime_turn_operation_keys_are_bounded_scoped_and_fingerprint_execution_poli
             false,
             &[],
             None,
+            &[],
         )?,
         runtime_turn_request_fingerprint(
             &thread,
@@ -1623,6 +1629,7 @@ fn runtime_turn_operation_keys_are_bounded_scoped_and_fingerprint_execution_poli
             true,
             &[],
             None,
+            &[],
         )?,
     ];
     assert!(
@@ -1924,6 +1931,7 @@ async fn operation_key_replays_torn_response_survives_restart_and_rejects_mismat
     let operation_key = "cwc-op-sk-fixture-must-not-persist".to_string();
     let request = StartTurnRequest {
         prompt: "idempotent turn payload".to_string(),
+        images: Vec::new(),
         operation_key: Some(operation_key.clone()),
         reasoning_effort: Some("high".to_string()),
         allowed_tools: Some(Vec::new()),
@@ -2121,6 +2129,7 @@ async fn restart_removes_torn_operation_item_before_reserved_turn_retry() -> Res
             &thread.id,
             StartTurnRequest {
                 prompt: "retry after the pre-submit crash".to_string(),
+                images: Vec::new(),
                 operation_key: Some(operation_key.to_string()),
                 ..StartTurnRequest::default()
             },
@@ -4558,7 +4567,7 @@ fn store_load_scenario() {
 
         // Construct a thread record persisted with a future schema version.
         let mut thread = sample_thread("thr_future");
-        thread.schema_version = CURRENT_RUNTIME_SCHEMA_VERSION + 1;
+        thread.schema_version = MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION + 1;
 
         // Bypass save_thread (which would respect our local schema_version)
         // by writing the JSON directly so we can simulate a future writer.
@@ -4582,7 +4591,7 @@ fn store_load_scenario() {
         let store = RuntimeThreadStore::open(dir.clone()).expect("open store");
 
         let mut turn = sample_turn("thr_t", "trn_future", RuntimeTurnStatus::InProgress);
-        turn.schema_version = CURRENT_RUNTIME_SCHEMA_VERSION + 1;
+        turn.schema_version = MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION + 1;
 
         let path = store.turns_dir.join(format!("{}.json", turn.id));
         std::fs::create_dir_all(path.parent().unwrap()).expect("mkdirs");
@@ -4605,7 +4614,7 @@ fn store_load_scenario() {
         let store = RuntimeThreadStore::open(dir.clone()).expect("open store");
 
         let mut item = sample_item("trn_t", "itm_future", TurnItemLifecycleStatus::InProgress);
-        item.schema_version = CURRENT_RUNTIME_SCHEMA_VERSION + 1;
+        item.schema_version = MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION + 1;
 
         let path = store.items_dir.join(format!("{}.json", item.id));
         std::fs::create_dir_all(path.parent().unwrap()).expect("mkdirs");
@@ -13260,7 +13269,7 @@ async fn fork_at_user_message_drops_tail_and_returns_user_text() -> Result<()> {
         .await?;
     seed_turns_with_user_messages(&manager, &thread.id, &["first", "second", "third"])?;
 
-    let (forked, original_text) = manager.fork_at_user_message(&thread.id, 0).await?;
+    let (forked, original_text, _) = manager.fork_at_user_message(&thread.id, 0).await?;
     assert_eq!(original_text.as_deref(), Some("third"));
     assert_ne!(forked.id, thread.id);
 
@@ -13297,7 +13306,7 @@ async fn fork_at_user_message_depth_one_drops_two_turns() -> Result<()> {
         .await?;
     seed_turns_with_user_messages(&manager, &thread.id, &["a", "b", "c", "d"])?;
 
-    let (forked, original_text) = manager.fork_at_user_message(&thread.id, 1).await?;
+    let (forked, original_text, _) = manager.fork_at_user_message(&thread.id, 1).await?;
     assert_eq!(original_text.as_deref(), Some("c"));
     let forked_turns = manager.store.list_turns_for_thread(&forked.id)?;
     let summaries: Vec<&str> = forked_turns
@@ -14237,3 +14246,520 @@ async fn report_store_failure_names_the_item_file_without_terminalizing() -> Res
     let _ = std::fs::remove_dir_all(dir);
     Ok(())
 }
+
+mod runtime_image_inputs {
+    use super::*;
+    use crate::core::engine::Engine;
+    use crate::image_attach::tests::runtime_image_fixture;
+    use crate::llm_client::mock::{MockLlmClient, canned};
+
+    fn config() -> Config {
+        let mut config = Config {
+            provider: Some("deepseek".into()),
+            default_text_model: Some("deepseek-v4-flash-vision-exp".into()),
+            api_key: Some("synthetic-image-fixture-key".into()),
+            runtime_chat_isolated: true,
+            ..Config::default()
+        };
+        config.set_provider_model_override(
+            ApiProvider::Deepseek,
+            Some("deepseek-v4-flash-vision-exp".into()),
+        );
+        config.set_feature("mcp", false).unwrap();
+        config.set_feature("subagents", false).unwrap();
+        config
+    }
+
+    #[tokio::test]
+    async fn runtime_image_admission_rejects_auto_unknown_and_malformed_without_engine()
+    -> Result<()> {
+        let _env = crate::test_support::lock_test_env();
+        let dir = tempfile::tempdir()?;
+        let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
+        let manager = RuntimeThreadManager::open(
+            config(),
+            dir.path().to_path_buf(),
+            test_manager_config(dir.path().join("runtime")),
+        )?;
+        let thread = manager
+            .create_thread(CreateThreadRequest::default())
+            .await?;
+        let good = runtime_image_fixture(17);
+        for (prompt, model, image) in [
+            ("look", "auto", good.clone()),
+            ("look", "deepseek-v4-flash", good.clone()),
+            ("look", "unknown-image-fixture", good.clone()),
+            ("", "deepseek-v4-flash-vision-exp", good.clone()),
+            (
+                "look",
+                "deepseek-v4-flash-vision-exp",
+                codewhale_protocol::runtime::RuntimeImageInput {
+                    mime: "image/jpeg".into(),
+                    ..good
+                },
+            ),
+        ] {
+            assert!(
+                manager
+                    .start_turn(
+                        &thread.id,
+                        StartTurnRequest {
+                            prompt: prompt.into(),
+                            model: Some(model.into()),
+                            images: vec![image],
+                            operation_key: Some("image-rejected".into()),
+                            ..Default::default()
+                        }
+                    )
+                    .await
+                    .is_err()
+            );
+            assert!(
+                manager.active.lock().await.engines.is_empty(),
+                "invalid image must fail before Engine load/classifier"
+            );
+            assert!(manager.store.list_turns_for_thread(&thread.id)?.is_empty());
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn runtime_image_engine_replay_restart_fork_and_retry_keep_exact_order() -> Result<()> {
+        let _env = crate::test_support::lock_test_env();
+        let dir = tempfile::tempdir()?;
+        let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
+        let config = config();
+        let manager_config = test_manager_config(dir.path().join("runtime"));
+        let manager = RuntimeThreadManager::open(
+            config.clone(),
+            dir.path().to_path_buf(),
+            manager_config.clone(),
+        )?;
+        let thread = manager
+            .create_thread(CreateThreadRequest::default())
+            .await?;
+        let images = vec![runtime_image_fixture(17), runtime_image_fixture(33)];
+        let mock = Arc::new(MockLlmClient::new(vec![canned::simple_text_turn(
+            "fixture image response",
+        )]));
+        let (engine, handle) = Engine::new_with_model_client(
+            EngineConfig {
+                workspace: dir.path().to_path_buf(),
+                model: thread.model.clone(),
+                subagents_enabled: false,
+                snapshots_enabled: false,
+                memory_enabled: false,
+                terminal_chrome_enabled: false,
+                runtime_services: crate::tools::spec::RuntimeToolServices {
+                    active_thread_id: Some(thread.id.clone()),
+                    ..Default::default()
+                },
+                ..EngineConfig::default()
+            },
+            &config,
+            mock.clone(),
+        );
+        manager
+            .install_test_engine(&thread.id, handle.clone())
+            .await?;
+        let run = tokio::spawn(engine.run());
+        let request = StartTurnRequest {
+            prompt: "compare the two images\n[Attached image: /private/host-only.png]".into(),
+            images: images.clone(),
+            operation_key: Some("image-exact-replay".into()),
+            ..Default::default()
+        };
+        let turn = manager.start_turn(&thread.id, request.clone()).await?;
+        assert_eq!(
+            wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(10))
+                .await?
+                .status,
+            RuntimeTurnStatus::Completed
+        );
+        assert_eq!(mock.call_count(), 1);
+        let provider_request = mock.last_request().context("mock request")?;
+        let sent: Vec<_> = provider_request
+            .messages
+            .iter()
+            .flat_map(|m| m.content.clone())
+            .collect();
+        assert_eq!(
+            crate::image_attach::runtime_images_from_blocks(&sent)?,
+            images
+        );
+        assert!(!serde_json::to_string(&provider_request.messages)?.contains("host-only.png"));
+        assert_eq!(
+            manager.start_turn(&thread.id, request.clone()).await?.id,
+            turn.id
+        );
+        let mut changed = request.clone();
+        changed.images.reverse();
+        assert!(manager.start_turn(&thread.id, changed).await.is_err());
+        assert_eq!(mock.call_count(), 1);
+        let item = manager
+            .store
+            .list_items_for_turn(&turn.id)?
+            .into_iter()
+            .find(|i| i.kind == TurnItemKind::UserMessage)
+            .unwrap();
+        assert_eq!(item.schema_version, IMAGE_RUNTIME_SCHEMA_VERSION);
+        assert!(
+            item.schema_version > 2,
+            "pre-image readers reject this schema"
+        );
+        assert_eq!(
+            manager.get_thread(&thread.id).await?.schema_version,
+            IMAGE_RUNTIME_SCHEMA_VERSION
+        );
+        handle.send(Op::Shutdown).await?;
+        tokio::time::timeout(Duration::from_secs(10), run).await??;
+        drop(handle);
+        drop(manager);
+        let reopened =
+            RuntimeThreadManager::open(config.clone(), dir.path().to_path_buf(), manager_config)?;
+        assert_eq!(reopened.start_turn(&thread.id, request).await?.id, turn.id);
+        assert!(
+            reopened.active.lock().await.engines.is_empty(),
+            "exact replay needs no provider or Engine"
+        );
+        let stored_thread = reopened.get_thread(&thread.id).await?;
+        let messages = reopened.restore_thread_messages(&stored_thread)?;
+        assert_eq!(
+            crate::image_attach::runtime_images_from_blocks(
+                &messages
+                    .into_iter()
+                    .flat_map(|m| m.content)
+                    .collect::<Vec<_>>()
+            )?,
+            images
+        );
+        let fork = reopened.fork_thread(&thread.id).await?;
+        let messages = reopened.restore_thread_messages(&fork)?;
+        assert_eq!(
+            crate::image_attach::runtime_images_from_blocks(
+                &messages
+                    .into_iter()
+                    .flat_map(|m| m.content)
+                    .collect::<Vec<_>>()
+            )?,
+            images
+        );
+        let (_, original_text, retry_images) = reopened.fork_at_user_message(&thread.id, 0).await?;
+        assert!(original_text.unwrap().starts_with("compare the two images"));
+        assert_eq!(retry_images, images);
+        let followup_mock = Arc::new(MockLlmClient::new(vec![canned::simple_text_turn(
+            "remembered screenshots",
+        )]));
+        let (engine, handle) = Engine::new_with_model_client(
+            EngineConfig {
+                workspace: dir.path().to_path_buf(),
+                model: thread.model.clone(),
+                subagents_enabled: false,
+                snapshots_enabled: false,
+                memory_enabled: false,
+                terminal_chrome_enabled: false,
+                runtime_services: crate::tools::spec::RuntimeToolServices {
+                    active_thread_id: Some(thread.id.clone()),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            &config,
+            followup_mock.clone(),
+        );
+        reopened
+            .install_test_engine(&thread.id, handle.clone())
+            .await?;
+        let run = tokio::spawn(engine.run());
+        handle
+            .send(Op::SyncSession {
+                session_id: stored_thread.session_id.clone(),
+                messages: reopened.restore_thread_messages(&stored_thread)?,
+                system_prompt: None,
+                system_prompt_override: false,
+                model: stored_thread.model.clone(),
+                workspace: stored_thread.workspace.clone(),
+                mode: crate::tui::app::AppMode::Agent,
+            })
+            .await?;
+        let followup = reopened
+            .start_turn(
+                &thread.id,
+                StartTurnRequest {
+                    prompt: "What differed between those screenshots?".into(),
+                    ..Default::default()
+                },
+            )
+            .await?;
+        assert_eq!(
+            wait_for_terminal_turn(&reopened, &followup.id, Duration::from_secs(10))
+                .await?
+                .status,
+            RuntimeTurnStatus::Completed
+        );
+        let outbound = followup_mock
+            .last_request()
+            .context("restored follow-up request")?;
+        assert_eq!(
+            crate::image_attach::runtime_images_from_blocks(
+                &outbound
+                    .messages
+                    .iter()
+                    .flat_map(|m| m.content.clone())
+                    .collect::<Vec<_>>()
+            )?,
+            images
+        );
+        assert!(!serde_json::to_string(&outbound.messages)?.contains("host-only.png"));
+        handle.send(Op::Shutdown).await?;
+        tokio::time::timeout(Duration::from_secs(10), run).await??;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn runtime_image_seed_roundtrips_user_block_order_and_future_schema_refuses() -> Result<()>
+    {
+        let dir = tempfile::tempdir()?;
+        let manager = test_manager(dir.path().join("runtime"))?;
+        let thread = manager
+            .create_thread(CreateThreadRequest::default())
+            .await?;
+        let mut content = crate::image_attach::prepare_runtime_images(&[
+            runtime_image_fixture(8),
+            runtime_image_fixture(9),
+        ])?;
+        content.insert(
+            1,
+            ContentBlock::Text {
+                text: "between images".into(),
+                cache_control: None,
+            },
+        );
+        let original = vec![Message {
+            role: Role::User,
+            content,
+        }];
+        manager
+            .seed_thread_from_messages(&thread.id, &original)
+            .await?;
+        let updated = manager.get_thread(&thread.id).await?;
+        assert_eq!(manager.restore_thread_messages(&updated)?, original);
+        let turn = manager.store.list_turns_for_thread(&thread.id)?.remove(0);
+        assert_eq!(turn.schema_version, IMAGE_RUNTIME_SCHEMA_VERSION);
+        let mut item = manager.store.list_items_for_turn(&turn.id)?.remove(0);
+        item.schema_version = MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION + 1;
+        manager.store.save_item(&item)?;
+        assert!(
+            manager
+                .store
+                .load_item(&item.id)
+                .unwrap_err()
+                .to_string()
+                .contains("newer than supported")
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn runtime_image_durable_corruption_and_bad_import_fail_closed() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let manager = test_manager(dir.path().join("runtime"))?;
+        let thread = manager.create_thread(Default::default()).await?;
+        let mut valid = crate::image_attach::prepare_runtime_images(&[runtime_image_fixture(4)])?;
+        valid.insert(
+            0,
+            ContentBlock::Text {
+                text: "keep my prompt".into(),
+                cache_control: None,
+            },
+        );
+        manager
+            .seed_thread_from_messages(
+                &thread.id,
+                &[Message {
+                    role: Role::User,
+                    content: valid.clone(),
+                }],
+            )
+            .await?;
+        let turn = manager.store.list_turns_for_thread(&thread.id)?.remove(0);
+        let original = manager.store.list_items_for_turn(&turn.id)?.remove(0);
+        let mut bad_url = valid.clone();
+        if let ContentBlock::ImageUrl { image_url } = &mut bad_url[1] {
+            image_url.url = "https://host.invalid/private-image".into();
+        }
+        let mut truncated = valid.clone();
+        if let ContentBlock::ImageUrl { image_url } = &mut truncated[1] {
+            image_url.url = "data:image/png;base64,iVBORw0KGgo=".into();
+        }
+        for content in [
+            vec![],
+            valid[..1].to_vec(),
+            valid[1..].to_vec(),
+            bad_url,
+            truncated,
+        ] {
+            let mut damaged = original.clone();
+            damaged.set_image_content(content);
+            manager.store.save_item(&damaged)?;
+            assert!(manager.store.load_item(&damaged.id).is_err());
+            assert!(manager.store.list_items_for_turn(&turn.id).is_err());
+            assert!(
+                manager
+                    .restore_thread_messages(&manager.get_thread(&thread.id).await?)
+                    .is_err()
+            );
+        }
+        // Legacy records never give this additive metadata field new authority.
+        let mut legacy = original.clone();
+        legacy.schema_version = 2;
+        legacy.metadata = Some(json!({"runtime_image_content": []}));
+        manager.store.save_item(&legacy)?;
+        assert_eq!(
+            manager.store.load_item(&legacy.id)?.user_content()?,
+            valid[..1]
+        );
+        let empty = manager.create_thread(Default::default()).await?;
+        let mut invalid = valid.clone();
+        if let ContentBlock::ImageUrl { image_url } = &mut invalid[1] {
+            image_url.url = "file:///private/never-read".into();
+        }
+        assert!(
+            manager
+                .seed_thread_from_messages(
+                    &empty.id,
+                    &[
+                        Message {
+                            role: Role::User,
+                            content: valid
+                        },
+                        Message {
+                            role: Role::User,
+                            content: invalid
+                        },
+                    ]
+                )
+                .await
+                .is_err()
+        );
+        assert!(manager.store.list_turns_for_thread(&empty.id)?.is_empty());
+        assert!(
+            manager
+                .get_thread(&empty.id)
+                .await?
+                .latest_turn_id
+                .is_none()
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn runtime_image_stored_retry_keeps_five_mib_bytes_without_relaxing_route_policy()
+    -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let config = config();
+        let manager = RuntimeThreadManager::open(
+            config,
+            dir.path().to_path_buf(),
+            test_manager_config(dir.path().join("runtime")),
+        )?;
+        let cases = [
+            vec![crate::image_attach::tests::runtime_image_fixture_bytes(
+                4 * 1024 * 1024 + 1,
+            )],
+            vec![crate::image_attach::tests::runtime_image_fixture_bytes(3 * 1024 * 1024); 2],
+            vec![runtime_image_fixture(5); 11],
+        ];
+        for expected in cases {
+            let thread = manager.create_thread(Default::default()).await?;
+            let mut content = crate::image_attach::prepare_stored_images(&expected)?;
+            content.insert(
+                0,
+                ContentBlock::Text {
+                    text: "stored screenshot".into(),
+                    cache_control: None,
+                },
+            );
+            manager
+                .seed_thread_from_messages(
+                    &thread.id,
+                    &[Message {
+                        role: Role::User,
+                        content,
+                    }],
+                )
+                .await?;
+            let (fork, prompt, images) = manager.fork_at_user_message(&thread.id, 0).await?;
+            assert_eq!(images, expected);
+            let request = StartTurnRequest {
+                prompt: prompt.unwrap(),
+                images,
+                ..Default::default()
+            };
+            assert!(manager.start_turn(&fork.id, request.clone()).await.is_err());
+            let mut unsupported = request.clone();
+            unsupported.model = Some("deepseek-v4-flash".into());
+            assert!(
+                manager
+                    .start_turn_from_stored_images(&fork.id, unsupported)
+                    .await
+                    .is_err()
+            );
+            assert!(!manager.active.lock().await.engines.contains_key(&fork.id));
+            let mut harness = crate::core::engine::mock_engine_handle();
+            manager
+                .install_test_engine(&fork.id, harness.handle.clone())
+                .await?;
+            let turn = manager
+                .start_turn_from_stored_images(&fork.id, request.clone())
+                .await?;
+            let Some(Op::SendMessage { images, .. }) = harness.rx_op.recv().await else {
+                bail!("expected stored image retry");
+            };
+            assert_eq!(images, request.images);
+            assert_eq!(turn.schema_version, IMAGE_RUNTIME_SCHEMA_VERSION);
+        }
+        Ok(())
+    }
+}
+
+#[test]
+fn runtime_image_native_fingerprint_keeps_legacy_text_and_binds_bytes() -> Result<()> {
+    let thread = sample_thread("thr_image_legacy");
+    let policy = RuntimePolicyProjection::from_persisted("agent", Some("ask"), false);
+    let fingerprint = |images: &[codewhale_protocol::runtime::RuntimeImageInput]| {
+        runtime_turn_request_fingerprint(
+            &thread,
+            "hello",
+            None,
+            "fixture-model",
+            None,
+            None,
+            policy,
+            false,
+            false,
+            &[],
+            None,
+            images,
+        )
+    };
+    // SHA-256 of the canonical committed version-1 text request fields.
+    assert_eq!(
+        fingerprint(&[])?,
+        "4dd89138dfd3bfdcffbfa4fed71664e98f23e40a8b46ee6a4829871d94103d67"
+    );
+    let one = crate::image_attach::tests::runtime_image_fixture(1);
+    let two = crate::image_attach::tests::runtime_image_fixture(2);
+    assert_ne!(
+        fingerprint(std::slice::from_ref(&one))?,
+        fingerprint(std::slice::from_ref(&two))?
+    );
+    assert_ne!(
+        fingerprint(&[one.clone(), two.clone()])?,
+        fingerprint(&[two, one])?
+    );
+    let absent: StartTurnRequest = serde_json::from_value(json!({"prompt":"hello"}))?;
+    let empty: StartTurnRequest = serde_json::from_value(json!({"prompt":"hello","images":[]}))?;
+    assert_eq!(serde_json::to_value(absent)?, serde_json::to_value(empty)?);
+    Ok(())
+}

crates/tui/src/tools/read_media.rs

@@ -15,6 +15,9 @@
 //! ladder result fits the budget the tool fails closed — nothing is sent —
 //! with the exact conversion command to retry with.
 
+#[cfg(test)]
+use image::ImageReader;
+#[cfg(test)]
 use std::io::Cursor;
 
 use async_trait::async_trait;
@@ -23,7 +26,7 @@ use codewhale_config::route::CapabilityState;
 use image::codecs::jpeg::JpegEncoder;
 use image::codecs::png::{CompressionType, FilterType as PngFilter, PngEncoder};
 use image::imageops::FilterType;
-use image::{DynamicImage, ExtendedColorType, GenericImageView, ImageEncoder, ImageReader, Limits};
+use image::{DynamicImage, ExtendedColorType, GenericImageView, ImageEncoder};
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 
@@ -35,14 +38,7 @@ use super::spec::{
 /// Maximum source image size before decoding (20 MiB).
 pub const MAX_SOURCE_IMAGE_BYTES: usize = 20 * 1024 * 1024;
 
-/// Maximum width or height admitted for an input image (8192 px).
-pub const MAX_IMAGE_DIMENSION: u32 = 8192;
-
-/// Maximum total pixels admitted before decoding is aborted (~33.5 megapixels).
-pub const MAX_IMAGE_PIXELS: u64 = 33_554_432;
-
-/// Memory allocation limit for image decoding (64 MiB).
-pub const MAX_DECODE_ALLOC_BYTES: u64 = 64 * 1024 * 1024;
+use crate::image_attach::decode_and_guard_image;
 
 /// Maximum inline image payload admitted on the wire (5 MiB).
 pub const MAX_WIRE_IMAGE_BYTES: usize = crate::image_attach::MAX_IMAGE_BYTES;
@@ -643,7 +639,8 @@ fn process_media_file(
     };
 
     // 6. Bounded decoding with decompression-bomb guards
-    let (processed_image, orig_width, orig_height) = decode_and_guard_image(&raw_bytes)?;
+    let (processed_image, orig_width, orig_height) = decode_and_guard_image(&raw_bytes)
+        .map_err(|error| ToolError::execution_failed(format!("read_media: {error}")))?;
 
     // 7. Apply crop if requested
     let (cropped_image, crop_applied) = if let Some(crop) = crop_region {
@@ -731,55 +728,6 @@ fn process_media_file(
     })
 }
 
-fn decode_and_guard_image(bytes: &[u8]) -> Result<(DynamicImage, u32, u32), ToolError> {
-    let mut reader = ImageReader::new(Cursor::new(bytes))
-        .with_guessed_format()
-        .map_err(|e| {
-            ToolError::execution_failed(format!("read_media: failed to detect format: {e}"))
-        })?;
-
-    let mut limits = Limits::default();
-    limits.max_alloc = Some(MAX_DECODE_ALLOC_BYTES);
-    limits.max_image_width = Some(MAX_IMAGE_DIMENSION);
-    limits.max_image_height = Some(MAX_IMAGE_DIMENSION);
-    reader.limits(limits);
-
-    // Read header/dimensions first
-    let (width, height) = reader.into_dimensions().map_err(|e| {
-        ToolError::execution_failed(format!(
-            "read_media: decompression bomb guard or invalid header detected: {e}"
-        ))
-    })?;
-
-    let total_pixels = (width as u64) * (height as u64);
-    if total_pixels > MAX_IMAGE_PIXELS
-        || width > MAX_IMAGE_DIMENSION
-        || height > MAX_IMAGE_DIMENSION
-    {
-        return Err(ToolError::execution_failed(format!(
-            "read_media: decompression bomb guard triggered: image dimensions ({width}x{height}, {total_pixels} pixels) exceed safe limits (max {MAX_IMAGE_DIMENSION}x{MAX_IMAGE_DIMENSION} / {MAX_IMAGE_PIXELS} pixels). Please downscale or crop the image first."
-        )));
-    }
-
-    // Decode full dynamic image with limits
-    let mut decode_reader = ImageReader::new(Cursor::new(bytes))
-        .with_guessed_format()
-        .map_err(|e| {
-            ToolError::execution_failed(format!("read_media: failed to read image: {e}"))
-        })?;
-    let mut decode_limits = Limits::default();
-    decode_limits.max_alloc = Some(MAX_DECODE_ALLOC_BYTES);
-    decode_limits.max_image_width = Some(MAX_IMAGE_DIMENSION);
-    decode_limits.max_image_height = Some(MAX_IMAGE_DIMENSION);
-    decode_reader.limits(decode_limits);
-
-    let dynamic_img = decode_reader.decode().map_err(|e| {
-        ToolError::execution_failed(format!("read_media: failed to decode image: {e}"))
-    })?;
-
-    Ok((dynamic_img, width, height))
-}
-
 /// How the encoding ladder should treat an image's content.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum EncodePolicy {

crates/tui/src/tui/ui/dispatch.rs

@@ -788,6 +788,7 @@ pub(crate) async fn spawned_dispatch_inner(
     if let Err(err) = engine_handle
         .send(Op::SendMessage {
             content: prepare.content.clone(),
+            images: Vec::new(),
             mode: prepare.mode,
             route: Box::new(turn_route),
             compaction: Box::new(turn_compaction.clone()),

docs/RUNTIME_API.md

@@ -692,6 +692,43 @@ resolved, and `auto` remains a per-prompt reasoning decision even when the
 thread uses a fixed model. The request still enters the existing
 `Op::SendMessage` path and the single `Engine::run_turn` loop.
 
+Image input uses the same turn path: `"images": [{"mime": "image/png",
+"dataBase64": "..."}]`. Clients must first observe
+`capabilities.turn_image_inputs: true` in `/v1/runtime/info` (or the isolated
+Runtime Chat relay catalog). Older HTTP runtimes ignore unknown fields, so a
+successful text response is not evidence that an attachment was accepted.
+The field is omitted when empty. It is also accepted by app-server
+`thread/message`, `thread/request` messages, and prompt requests; that bridge
+checks the underlying Runtime capability before forwarding image bytes.
+Legacy remote Work commands do not support images and explicitly refuse them.
+
+New inline images require a named model whose exact resolved route reports
+`image_input: "supported"`; Auto and unknown/unsupported image routes are
+refused before classifier or provider dispatch. This does not change the
+existing trusted-local attachment behavior for routes with unknown capability.
+A nonempty prompt is required. Inputs are limited to 10 images, 4 MiB decoded
+bytes per image, 5 MiB total, and an 8 MiB JSON body. PNG, JPEG, GIF and WebP
+must have matching MIME, canonical padded base64 and valid bounded image
+content: at most 8192 pixels per dimension, 33,554,432 pixels total and 64 MiB
+decoder allocation. The Runtime does not fetch paths or URLs from this field.
+Malformed images refuse the whole turn; callers can retain the draft for
+correction. Relay command polling uses an 8 MiB response budget; the sender
+must paginate by serialized bytes without advancing past unserved commands.
+
+Accepted image bytes and order are retained in the existing turn records and
+reconstructed after restart, import and fork. Retry retains those images even
+when its optional `prompt` changes the text; undo responses include
+`original_user_images` when present. Image-bearing records require schema v3,
+which older readers refuse. Text-only records and operation fingerprints retain
+their prior representation. Validated stored local images retain the existing
+5 MiB per-image ceiling and prior aggregate/count semantics on import/retry;
+this internal storage authority does not
+relax exact model or permission checks. Image bytes, MIME and order participate in request
+identity, so changing an image under the same operation key conflicts.
+Compaction can summarize older context; retaining the original attachment does
+not promise that every later model request includes it. Image pixels are not
+subject to text-secret redaction.
+
 `operation_key` is an optional idempotency key for clients that may lose an
 HTTP response after the Runtime accepted a turn. It is scoped to the current
 Runtime store and thread, may contain at most 128 UTF-8 bytes, and may not be
@@ -964,6 +1001,14 @@ non-empty list does not prove that the route can currently serve a request.
 }
 ```
 
+For an exact configured route, supply `?model_provider_id=vision-work` and
+require the response to echo that same `model_provider_id`. The Runtime resolves
+that identity under the requested provider kind before reading model support.
+Unknown or mismatched identities return `400`. Named pagination cursors bind the
+configuration identity, endpoint and catalog snapshot; changing any of those
+requires restarting pagination. Omitting the query preserves the legacy catalog
+projection and omits the identity echo.
+
 The catalog for one provider. Returns `400` for an unknown id, and for the
 legacy `deepseek-cn` alias, which has no provider metadata — use `deepseek`.
 An empty `models` array means the Runtime has no discoverable or configured