diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 4f13167..3173fe9 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -480,11 +480,15 @@ File: [`src/claude.rs`](src/claude.rs) | New chat | `claude -p --session-id ` | | Resumed chat | `claude -p --resume ` | | Instructions | `--append-system-prompt ` | +| Images | base64 content blocks over `--input-format stream-json` stdin | | Unattended job | `--permission-mode bypassPermissions` | | Evaluator | safe mode, no tools, no MCP, no Chrome, no session persistence | Push chooses Claude's initial session ID and marks it started before execution -so a partial create failure is not retried as another create. +so a partial create failure is not retried as another create. Text-only turns +keep the single-result JSON command contract. Image turns use one streaming JSON +user message and parse the final result event without putting image data in +process arguments. ### Codex diff --git a/Cargo.lock b/Cargo.lock index 9022494..ed3545b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -800,6 +800,7 @@ name = "push" version = "0.10.0" dependencies = [ "anyhow", + "base64", "chrono", "chrono-tz", "futures-util", diff --git a/Cargo.toml b/Cargo.toml index 8f1d1ea..ed5850f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,6 +22,7 @@ toml = "1" toml_edit = "0.23" uuid = { version = "1", features = ["v4"] } anyhow = "1" +base64 = "0.22" humantime = "2" sha2 = "0.10" chrono-tz = "0.10" diff --git a/docs/telegram.md b/docs/telegram.md index e968322..391f906 100644 --- a/docs/telegram.md +++ b/docs/telegram.md @@ -52,10 +52,10 @@ Telegram-only preflight does not open `chat.db` and does not require macOS or ## Image Messages -With Codex or Pi selected for the conversation, send the bot a Telegram photo -or an image document. Push accepts JPEG, PNG, and WebP files. A caption becomes -the message text; an image without a caption is still accepted. Telegram media -groups are processed as separate messages. +Send the bot a Telegram photo or an image document with Claude Code, Codex, or +Pi selected for the conversation. Push accepts JPEG, PNG, and WebP files. A +caption becomes the message text; an image without a caption is still accepted. +Telegram media groups are processed as separate messages. Push downloads an image only after the private-chat and sender allowlist checks pass. The declared and downloaded size must fit within the 6 MiB per-message @@ -67,13 +67,12 @@ Validated bytes are written to an owner-only temporary directory under `$PUSH_HOME/cache`, attached to the agent turn, and removed after success, failure, timeout, interruption, or session recovery. Push stores the caption or an `[Image attachment]` placeholder in conversation history, not the image -bytes. The image is sent through the configured Codex or Pi model provider, so -review that provider's image and data controls before using this feature. Pi -receives images using its native `@image-path` arguments while message text -continues through standard input. +bytes. The image is sent through the configured model provider, so review that +provider's image and data controls before using this feature. Claude Code +receives base64 content blocks through streaming JSON input. Pi receives native +`@image-path` arguments while message text continues through standard input. -Image input for Claude Code, and sending generated images back through Telegram, -are not supported yet. +Sending generated images back through Telegram is not supported yet. ## Voice Messages diff --git a/src/agent.rs b/src/agent.rs index 8c30609..529fff5 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -124,17 +124,6 @@ impl Runner { matches!(self, Runner::Claude(_)) } - pub fn supports_images(&self) -> bool { - if matches!(self, Runner::Codex(_) | Runner::Pi(_)) { - return true; - } - #[cfg(test)] - if let Self::Fake(runner) = self { - return matches!(runner.backend, AgentBackend::Codex | AgentBackend::Pi); - } - false - } - pub async fn run(&self, req: Request<'_>, timeout: Duration) -> Result { match self { Runner::Claude(r) => r.run(req, timeout).await, diff --git a/src/claude.rs b/src/claude.rs index 5d9ae28..b738ad7 100644 --- a/src/claude.rs +++ b/src/claude.rs @@ -1,11 +1,17 @@ //! Runs the Claude Code CLI headlessly for a single message. +use std::io; +use std::process::Stdio; use std::time::Duration; +use base64::Engine as _; use serde::Deserialize; +use serde_json::{json, Value}; +use tokio::io::AsyncWriteExt; use tokio::process::Command; use crate::agent::{final_reply, Request, RunError, RunOutput}; +use crate::image::{media_type, MAX_IMAGE_BYTES, MAX_IMAGE_COUNT}; use crate::util::non_empty_session_id; /// Runner invokes the `claude` binary in print mode. @@ -61,9 +67,35 @@ impl Runner { mode: RunMode, ) -> Result { let is_resume = !req.is_new; + let stream_input = if req.images.is_empty() { + None + } else { + Some(image_stream_input(&req)?) + }; + let uses_stream_input = stream_input.is_some(); let attempt = crate::agent::output_with_retry(|| { - let mut cmd = self.command(&req, mode); - async move { cmd.output().await } + let mut cmd = self.command(&req, mode, uses_stream_input); + let stream_input = stream_input.clone(); + async move { + let Some(input) = stream_input else { + return cmd.output().await; + }; + let mut child = cmd.spawn()?; + let mut stdin = child.stdin.take().ok_or_else(|| { + io::Error::new(io::ErrorKind::BrokenPipe, "claude stdin unavailable") + })?; + let write_input = async move { + let result = stdin.write_all(&input).await; + drop(stdin); + result + }; + let (write_result, output) = tokio::join!(write_input, child.wait_with_output()); + let output = output?; + if output.status.success() { + write_result?; + } + Ok(output) + } }); let out = match tokio::time::timeout(timeout, attempt).await { Err(_) => return Err(RunError::Timeout), @@ -71,15 +103,28 @@ impl Runner { Ok(Ok(o)) => o, }; - self.parse_output(out, is_resume) + if uses_stream_input { + self.parse_stream_output(out, is_resume) + } else { + self.parse_output(out, is_resume) + } } - fn command(&self, req: &Request<'_>, mode: RunMode) -> Command { + fn command(&self, req: &Request<'_>, mode: RunMode, uses_stream_input: bool) -> Command { let mut cmd = Command::new(&self.bin); - cmd.arg("-p") - .arg(req.prompt) - .arg("--output-format") - .arg("json"); + cmd.arg("-p"); + if uses_stream_input { + cmd.arg("--input-format") + .arg("stream-json") + .arg("--output-format") + .arg("stream-json") + .arg("--verbose"); + cmd.stdin(Stdio::piped()); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + } else { + cmd.arg(req.prompt).arg("--output-format").arg("json"); + } if mode == RunMode::Evaluator { cmd.arg("--safe-mode") .arg("--tools") @@ -105,6 +150,54 @@ impl Runner { cmd } + fn parse_stream_output( + &self, + out: std::process::Output, + is_resume: bool, + ) -> Result { + let result = stream_result(&out.stdout); + let diagnostic = result + .as_ref() + .ok() + .map(|result| result.result.as_str()) + .filter(|message| !message.trim().is_empty()) + .map(str::to_string) + .or_else(|| { + let stderr = String::from_utf8_lossy(&out.stderr).trim().to_string(); + (!stderr.is_empty()).then_some(stderr) + }); + if is_resume && diagnostic.as_deref().is_some_and(missing_resume_error) { + return Err(RunError::SessionMissing( + "Claude could not find the saved session; Push will rebuild it from conversation history" + .to_string(), + )); + } + let result = match result { + Ok(result) => result, + Err(()) if out.status.success() => { + return Err(RunError::Failed( + "Claude returned malformed streaming JSON output".to_string(), + )); + } + Err(()) => { + return Err(RunError::Failed( + "Claude image request failed; check Claude Code provider and authentication settings" + .to_string(), + )); + } + }; + if result.is_error || !out.status.success() { + return Err(RunError::Failed( + "Claude image request failed; check Claude Code provider and authentication settings" + .to_string(), + )); + } + Ok(RunOutput { + reply: final_reply("claude", &result.result)?, + session_id: non_empty_session_id(&result.session_id).map(str::to_string), + }) + } + fn parse_output( &self, out: std::process::Output, @@ -148,6 +241,99 @@ impl Runner { } } +fn image_stream_input(req: &Request<'_>) -> Result, RunError> { + if req.images.len() > MAX_IMAGE_COUNT { + return Err(RunError::Failed(format!( + "Claude image request has more than {MAX_IMAGE_COUNT} attachments" + ))); + } + + let mut total = 0usize; + let mut images = Vec::with_capacity(req.images.len()); + for path in req.images { + let declared = std::fs::metadata(path) + .map_err(|error| { + RunError::Failed(format!( + "read Claude image attachment {}: {error}", + path.display() + )) + })? + .len(); + let declared = usize::try_from(declared).map_err(|_| { + RunError::Failed("Claude image request exceeds the 6 MiB limit".to_string()) + })?; + total = total.checked_add(declared).ok_or_else(|| { + RunError::Failed("Claude image request exceeds the 6 MiB limit".to_string()) + })?; + if total > MAX_IMAGE_BYTES { + return Err(RunError::Failed( + "Claude image request exceeds the 6 MiB limit".to_string(), + )); + } + + let bytes = std::fs::read(path).map_err(|error| { + RunError::Failed(format!( + "read Claude image attachment {}: {error}", + path.display() + )) + })?; + let media_type = media_type(&bytes).map_err(|_| { + RunError::Failed( + "Claude image request contains an unsupported image attachment".to_string(), + ) + })?; + images.push((media_type, bytes)); + } + + let actual_total = images + .iter() + .try_fold(0usize, |total, (_, bytes)| total.checked_add(bytes.len())); + if !matches!(actual_total, Some(total) if total <= MAX_IMAGE_BYTES) { + return Err(RunError::Failed( + "Claude image request exceeds the 6 MiB limit".to_string(), + )); + } + + let prompt = if req.prompt.trim().is_empty() { + "[Image attachment]" + } else { + req.prompt + }; + let mut content = vec![json!({"type": "text", "text": prompt})]; + content.extend(images.into_iter().map(|(media_type, bytes)| { + json!({ + "type": "image", + "source": { + "type": "base64", + "media_type": media_type, + "data": base64::engine::general_purpose::STANDARD.encode(bytes), + } + }) + })); + let mut input = serde_json::to_vec(&json!({ + "type": "user", + "message": { + "role": "user", + "content": content, + } + })) + .map_err(|_| RunError::Failed("build Claude streaming JSON input".to_string()))?; + input.push(b'\n'); + Ok(input) +} + +fn stream_result(stdout: &[u8]) -> Result { + let stdout = std::str::from_utf8(stdout).map_err(|_| ())?; + let mut result = None; + for line in stdout.lines().filter(|line| !line.trim().is_empty()) { + let value: Value = serde_json::from_str(line).map_err(|_| ())?; + if value.get("type").and_then(Value::as_str) == Some("result") { + result = Some(serde_json::from_value(value).map_err(|_| ())?); + } + } + result.ok_or(()) +} + fn missing_resume_error(message: &str) -> bool { message .to_ascii_lowercase() @@ -332,6 +518,265 @@ mod tests { let args = read_args(&args_path); assert!(!args.contains(&"--permission-mode".to_string())); + assert_arg_pair(&args, "-p", "hello"); + assert_arg_pair(&args, "--output-format", "json"); + assert!(!args.contains(&"--input-format".to_string())); + assert!(!args.contains(&"--verbose".to_string())); + } + + #[tokio::test] + async fn sends_ordered_images_with_text_on_new_and_resumed_sessions() { + for is_new in [true, false] { + let args_path = temp_path("claude-image-args"); + let input_path = temp_path("claude-image-input"); + let work_dir = temp_dir("claude-image-work"); + let jpeg = temp_path("claude-image.jpg"); + let png = temp_path("claude-image.png"); + let webp = temp_path("claude-image.webp"); + let jpeg_bytes = b"\xff\xd8\xffjpeg-body"; + let png_bytes = b"\x89PNG\r\n\x1a\npng-body"; + let webp_bytes = b"RIFF\x04\x00\x00\x00WEBPwebp-body"; + std::fs::write(&jpeg, jpeg_bytes).unwrap(); + std::fs::write(&png, png_bytes).unwrap(); + std::fs::write(&webp, webp_bytes).unwrap(); + let script = format!( + "#!/bin/sh\nprintf '%s\\n' \"$@\" > {}\ncat > {}\nprintf '%s\\n' '{{\"type\":\"system\",\"subtype\":\"init\"}}'\nprintf '%s\\n' '{{\"type\":\"result\",\"subtype\":\"success\",\"is_error\":false,\"result\":\"saw images\",\"session_id\":\"returned-session\"}}'\n", + sh_arg(&args_path), + sh_arg(&input_path), + ); + let cli = FakeCli::new("claude", &script); + let runner = Runner { bin: cli.bin() }; + let prompt = "compare these images"; + + let output = runner + .run( + Request { + session_id: if is_new { + "new-session" + } else { + "existing-session" + }, + is_new, + work_dir: work_dir.to_str().unwrap(), + instructions: "assistant identity", + prompt, + images: &[jpeg.clone(), png.clone(), webp.clone()], + }, + Duration::from_secs(5), + ) + .await + .unwrap(); + + assert_eq!(output.reply, "saw images"); + assert_eq!(output.session_id.as_deref(), Some("returned-session")); + let args = read_args(&args_path); + assert!(args.contains(&"-p".to_string())); + assert_arg_pair(&args, "--input-format", "stream-json"); + assert_arg_pair(&args, "--output-format", "stream-json"); + assert!(args.contains(&"--verbose".to_string())); + assert!(!args.contains(&prompt.to_string())); + let raw_args = std::fs::read_to_string(&args_path).unwrap(); + for bytes in [ + jpeg_bytes.as_slice(), + png_bytes.as_slice(), + webp_bytes.as_slice(), + ] { + assert!( + !raw_args.contains(&base64::engine::general_purpose::STANDARD.encode(bytes)) + ); + } + if is_new { + assert_arg_pair(&args, "--session-id", "new-session"); + assert!(!args.contains(&"--resume".to_string())); + } else { + assert_arg_pair(&args, "--resume", "existing-session"); + assert!(!args.contains(&"--session-id".to_string())); + } + + let input: Value = + serde_json::from_slice(&std::fs::read(&input_path).unwrap()).unwrap(); + assert_eq!(input["type"], "user"); + assert_eq!(input["message"]["role"], "user"); + let content = input["message"]["content"].as_array().unwrap(); + assert_eq!(content.len(), 4); + assert_eq!(content[0], json!({"type": "text", "text": prompt})); + for (part, media_type, bytes) in [ + (&content[1], "image/jpeg", jpeg_bytes.as_slice()), + (&content[2], "image/png", png_bytes.as_slice()), + (&content[3], "image/webp", webp_bytes.as_slice()), + ] { + assert_eq!(part["type"], "image"); + assert_eq!(part["source"]["type"], "base64"); + assert_eq!(part["source"]["media_type"], media_type); + assert_eq!( + part["source"]["data"], + base64::engine::general_purpose::STANDARD.encode(bytes) + ); + } + } + } + + #[test] + fn image_only_stream_input_uses_a_text_placeholder() { + let work_dir = temp_dir("claude-image-only-work"); + let image = temp_path("claude-image-only.png"); + std::fs::write(&image, b"\x89PNG\r\n\x1a\nbody").unwrap(); + let input = image_stream_input(&Request { + prompt: " \t ", + images: std::slice::from_ref(&image), + ..request(work_dir.to_str().unwrap()) + }) + .unwrap(); + let input: Value = serde_json::from_slice(&input).unwrap(); + let content = input["message"]["content"].as_array().unwrap(); + + assert_eq!(content[0]["type"], "text"); + assert_eq!(content[0]["text"], "[Image attachment]"); + assert_eq!(content[1]["type"], "image"); + } + + #[tokio::test] + async fn rejects_images_above_the_raw_limit_before_spawning_claude() { + let work_dir = temp_dir("claude-large-image-work"); + let image = temp_path("claude-large-image.jpg"); + let marker = temp_path("claude-large-image-spawned"); + let mut bytes = vec![0; MAX_IMAGE_BYTES + 1]; + bytes[..3].copy_from_slice(b"\xff\xd8\xff"); + std::fs::write(&image, bytes).unwrap(); + let cli = FakeCli::new("claude", &format!("#!/bin/sh\ntouch {}\n", sh_arg(&marker))); + let runner = Runner { bin: cli.bin() }; + + let error = runner + .run( + Request { + images: std::slice::from_ref(&image), + ..request(work_dir.to_str().unwrap()) + }, + Duration::from_secs(5), + ) + .await + .unwrap_err(); + + assert_failed(error, "Claude image request exceeds the 6 MiB limit"); + assert!(!marker.exists()); + } + + #[tokio::test] + async fn malformed_stream_output_is_a_typed_backend_error() { + let work_dir = temp_dir("claude-malformed-stream-work"); + let image = temp_path("claude-malformed-stream.png"); + std::fs::write(&image, b"\x89PNG\r\n\x1a\nbody").unwrap(); + let cli = FakeCli::new( + "claude", + "#!/bin/sh\ncat >/dev/null\nprintf 'not-json\\n'\n", + ); + let runner = Runner { bin: cli.bin() }; + + let error = runner + .run( + Request { + images: std::slice::from_ref(&image), + ..request(work_dir.to_str().unwrap()) + }, + Duration::from_secs(5), + ) + .await + .unwrap_err(); + + assert_failed(error, "Claude returned malformed streaming JSON output"); + } + + #[tokio::test] + async fn drains_large_startup_output_while_writing_image_input() { + let work_dir = temp_dir("claude-concurrent-stream-work"); + let image = temp_path("claude-concurrent-stream.png"); + let mut bytes = vec![0; 1024 * 1024]; + bytes[..8].copy_from_slice(b"\x89PNG\r\n\x1a\n"); + std::fs::write(&image, bytes).unwrap(); + let cli = FakeCli::new( + "claude", + r#"#!/bin/sh +i=0 +while [ "$i" -lt 10000 ]; do + printf '%s\n' '{"type":"system"}' + i=$((i + 1)) +done +cat >/dev/null +printf '%s\n' '{"type":"result","subtype":"success","is_error":false,"result":"done","session_id":"stream-session"}' +"#, + ); + let runner = Runner { bin: cli.bin() }; + + let output = runner + .run( + Request { + images: std::slice::from_ref(&image), + ..request(work_dir.to_str().unwrap()) + }, + Duration::from_secs(5), + ) + .await + .unwrap(); + + assert_eq!(output.reply, "done"); + assert_eq!(output.session_id.as_deref(), Some("stream-session")); + } + + #[tokio::test] + async fn image_failure_does_not_expose_base64_content() { + let work_dir = temp_dir("claude-image-failure-work"); + let image = temp_path("claude-image-failure.png"); + let bytes = b"\x89PNG\r\n\x1a\nprivate-image-data"; + std::fs::write(&image, bytes).unwrap(); + let cli = FakeCli::new("claude", "#!/bin/sh\ncat >&2\nprintf '\\n' >&2\nexit 1\n"); + let runner = Runner { bin: cli.bin() }; + + let error = runner + .run( + Request { + images: std::slice::from_ref(&image), + ..request(work_dir.to_str().unwrap()) + }, + Duration::from_secs(5), + ) + .await + .unwrap_err(); + + let encoded = base64::engine::general_purpose::STANDARD.encode(bytes); + match error { + RunError::Failed(message) => { + assert!(message.contains("Claude image request failed")); + assert!(!message.contains(&encoded)); + } + other => panic!("expected failed error, got {other:?}"), + } + } + + #[tokio::test] + async fn image_resume_lookup_failure_remains_typed() { + let work_dir = temp_dir("claude-image-missing-resume-work"); + let image = temp_path("claude-image-missing-resume.png"); + std::fs::write(&image, b"\x89PNG\r\n\x1a\nbody").unwrap(); + let cli = FakeCli::new( + "claude", + "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' '{\"type\":\"result\",\"is_error\":true,\"result\":\"No conversation found with session ID missing\"}'\nexit 1\n", + ); + let runner = Runner { bin: cli.bin() }; + + let error = runner + .run( + Request { + session_id: "missing", + is_new: false, + images: std::slice::from_ref(&image), + ..request(work_dir.to_str().unwrap()) + }, + Duration::from_secs(5), + ) + .await + .unwrap_err(); + + assert!(matches!(error, RunError::SessionMissing(_))); } #[tokio::test] diff --git a/src/gateway/tests.rs b/src/gateway/tests.rs index 86a7755..50e4ce5 100644 --- a/src/gateway/tests.rs +++ b/src/gateway/tests.rs @@ -1965,10 +1965,10 @@ async fn invalid_and_oversized_telegram_images_fall_back_without_running_the_age } #[tokio::test(flavor = "current_thread")] -async fn telegram_image_on_unsupported_backend_falls_back_without_downloading() { +async fn telegram_image_reaches_claude_and_is_removed_after_the_turn() { let state_path = temp_state_path(); - let sessions_dir = temp_path("unsupported-image-sessions"); - let assistant_dir = temp_path("unsupported-image-assistant"); + let sessions_dir = temp_path("claude-image-sessions"); + let assistant_dir = temp_path("claude-image-assistant"); std::fs::create_dir_all(&assistant_dir).unwrap(); let calls = Arc::new(Mutex::new(Vec::new())); let mut cfg = test_config( @@ -1993,16 +1993,20 @@ async fn telegram_image_on_unsupported_backend_falls_back_without_downloading() resume_missing_once: None, }), )])); - let mut message = telegram_image_message(1, 7, 7, "inspect"); - message.images[0].data = Some(b"not an image".to_vec()); - - run_messages(&mut gateway, vec![message]).await; + run_messages( + &mut gateway, + vec![telegram_image_message(1, 7, 7, "inspect")], + ) + .await; - assert!(calls.lock().unwrap().is_empty()); + let calls = calls.lock().unwrap(); + assert_eq!(calls.len(), 1); assert_eq!( - gateway.ctx.sent_replies.lock().unwrap()[0].1, - "Image messages are currently supported only when this conversation routes to Codex or Pi." + crate::prompt::current_message(&calls[0].prompt).as_deref(), + Some("inspect") ); + assert_eq!(calls[0].images.len(), 1); + assert!(!calls[0].images[0].exists()); let _ = std::fs::remove_file(&state_path); let _ = std::fs::remove_file(format!("{state_path}.db")); diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index c6ff4e1..ecba8df 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -144,20 +144,6 @@ where return; }; - if !job.image_attachments.is_empty() && !runner.supports_images() { - let reply = "Image messages are currently supported only when this conversation routes to Codex or Pi."; - let delivery = record_and_deliver(ctx, &job, OutboundOrigin::Gateway, reply).await; - report_delivery( - ctx, - &job, - delivery, - reply, - "image_backend_unsupported", - "deliver image backend fallback", - ); - return; - } - let session_result = { let initial_session_id = runner.initial_session_id(); ctx.store.lock().unwrap().session_for( diff --git a/src/image.rs b/src/image.rs index ddc030d..de5dd3b 100644 --- a/src/image.rs +++ b/src/image.rs @@ -90,19 +90,28 @@ impl Drop for PreparedImages { } } -fn image_extension(bytes: &[u8]) -> Result<&'static str> { +pub(crate) fn media_type(bytes: &[u8]) -> Result<&'static str> { if bytes.starts_with(&[0xff, 0xd8, 0xff]) { - return Ok("jpg"); + return Ok("image/jpeg"); } if bytes.starts_with(b"\x89PNG\r\n\x1a\n") { - return Ok("png"); + return Ok("image/png"); } if bytes.len() >= 12 && bytes.starts_with(b"RIFF") && &bytes[8..12] == b"WEBP" { - return Ok("webp"); + return Ok("image/webp"); } bail!("image attachment is not a supported JPEG, PNG, or WebP file") } +fn image_extension(bytes: &[u8]) -> Result<&'static str> { + match media_type(bytes)? { + "image/jpeg" => Ok("jpg"), + "image/png" => Ok("png"), + "image/webp" => Ok("webp"), + _ => unreachable!("media_type returns only supported image types"), + } +} + #[cfg(test)] mod tests { use super::*;