diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index a4b8f57..1106f35 100644 --- a/server/src/session/claude.rs +++ b/server/src/session/claude.rs @@ -1,6 +1,15 @@ //! The Claude Code driver: `claude -p` speaking stream-json on stdio, //! translated into the common event model. //! +//! This half owns the process -- spawning it (locally or through `ssh`), +//! resuming it after a crash, writing lines to it, and shutting it down. +//! Turning a line into [`Event`]s is [`translate`], which changes when the +//! CLI's wire format does rather than when any of the above does. +//! +//! The probing record below stays here, since it is the provenance for +//! both halves: the flags are this file's, the message catalogue is what +//! `translate` implements. +//! //! Wire format pinned against CLI 2.1.237 by probing (2026-08-24; scripts //! summarized here since they live outside the repo): //! @@ -22,7 +31,6 @@ //! - `control_request{subtype:set_model}` answers success; //! `{subtype:interrupt}` stops the turn. -use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; @@ -33,12 +41,15 @@ use tokio::sync::{mpsc, oneshot}; use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus}; use crate::config::{HostConfig, ProviderConfig, SessionConfig}; +use translate::{AnswerOutcome, Translator}; /// Where the driver remembers its CLI session id between backend runs -- /// the whole crash-recovery story: respawning with `--resume ` picks /// the conversation back up from Claude's own session files. Kept in the /// session directory rather than config.ron so the shared schema stays /// free of per-driver state. +mod translate; + const RESUME_FILE: &str = "claude-session.json"; /// Grace period between closing stdin (the polite exit) and SIGKILL. @@ -362,611 +373,3 @@ fn attachment_block(session_dir: &Path, id: &str) -> Result { } })) } - -/// What answering a question produced. -enum AnswerOutcome { - /// Send this control_response line to the CLI. - Respond(Value), - /// Part of a multi-question request; more answers still needed. - Pending, - Unknown, -} - -/// A `can_use_tool` request we've surfaced to the phone and not yet -/// answered. For plain permissions there is one implicit question -/// (Allow/Deny); for AskUserQuestion, one per entry in `questions`. -struct PendingRequest { - request_id: String, - input: Value, - /// Question text per sub-question, in order -- the keys the answers - /// map uses. Empty for a plain permission request. - questions: Vec, - answers: HashMap, -} - -/// Translation state: stream-json lines in, common events out. The one -/// side effect is saving images a tool result carries into the session -/// dir (they'd bloat the transcript as base64); everything else is pure, -/// so the dialect mapping is unit-testable from recorded lines. -struct Translator { - session_id: Option, - pending: HashMap, - session_dir: PathBuf, -} - -impl Translator { - fn new(session_dir: PathBuf) -> Self { - Self { - session_id: None, - pending: HashMap::new(), - session_dir, - } - } - fn translate(&mut self, message: &Value) -> Vec { - // Events from subagents (Task tool internals) carry a - // parent_tool_use_id; the transcript shows the Task tool's own - // start/end instead of every nested step. - if message - .get("parent_tool_use_id") - .is_some_and(|id| !id.is_null()) - { - return Vec::new(); - } - match message.get("type").and_then(Value::as_str) { - Some("system") => { - if message.get("subtype").and_then(Value::as_str) == Some("init") - && let Some(id) = message.get("session_id").and_then(Value::as_str) - { - self.session_id = Some(id.to_string()); - } - Vec::new() - } - Some("stream_event") => self.translate_stream_event(&message["event"]), - Some("assistant") => self.translate_assistant(&message["message"]), - Some("user") => self.translate_user(message), - Some("control_request") => self.translate_control_request(message), - Some("control_response") => { - let response = &message["response"]; - if response.get("subtype").and_then(Value::as_str) == Some("error") { - let error = response - .get("error") - .and_then(Value::as_str) - .unwrap_or("unknown"); - vec![Event::Error { - message: format!("claude rejected a request: {error}"), - }] - } else { - Vec::new() - } - } - Some("result") => { - let usage = &message["usage"]; - let tokens = usage - .get("input_tokens") - .and_then(Value::as_u64) - .unwrap_or(0) - + usage - .get("output_tokens") - .and_then(Value::as_u64) - .unwrap_or(0); - let mut events = Vec::new(); - if message - .get("is_error") - .and_then(Value::as_bool) - .unwrap_or(false) - { - events.push(Event::Error { - message: message - .get("result") - .and_then(Value::as_str) - .unwrap_or("the turn ended with an error") - .to_string(), - }); - } - if tokens > 0 { - events.push(Event::UsageDelta { tokens }); - } - events.push(Event::Status { - state: SessionStatus::Idle, - }); - events - } - _ => Vec::new(), - } - } - - /// Raw API streaming: only text deltas become events. Consolidated - /// blocks arriving later re-carry the same text, so those are skipped - /// in `translate_assistant` -- one source per fact. - fn translate_stream_event(&mut self, event: &Value) -> Vec { - if event.get("type").and_then(Value::as_str) == Some("content_block_delta") - && let Some(delta) = event["delta"].get("text") - && event["delta"].get("type").and_then(Value::as_str) == Some("text_delta") - && let Some(text) = delta.as_str() - { - return vec![Event::AssistantText { - delta: text.to_string(), - }]; - } - Vec::new() - } - - fn translate_assistant(&mut self, message: &Value) -> Vec { - let Some(content) = message.get("content").and_then(Value::as_array) else { - return Vec::new(); - }; - content - .iter() - .filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use")) - .map(|block| Event::ToolStart { - id: block - .get("id") - .and_then(Value::as_str) - .unwrap_or_default() - .to_string(), - tool: block - .get("name") - .and_then(Value::as_str) - .unwrap_or_default() - .to_string(), - input: block.get("input").cloned().unwrap_or(Value::Null), - }) - .collect() - } - - fn translate_control_request(&mut self, message: &Value) -> Vec { - let request = &message["request"]; - if request.get("subtype").and_then(Value::as_str) != Some("can_use_tool") { - return Vec::new(); - } - let request_id = message - .get("request_id") - .and_then(Value::as_str) - .unwrap_or_default() - .to_string(); - let tool_name = request - .get("tool_name") - .and_then(Value::as_str) - .unwrap_or("a tool"); - let input = request.get("input").cloned().unwrap_or(Value::Null); - - let mut events = Vec::new(); - let mut questions = Vec::new(); - if tool_name == "AskUserQuestion" { - for (i, question) in input - .get("questions") - .and_then(Value::as_array) - .into_iter() - .flatten() - .enumerate() - { - let text = question - .get("question") - .and_then(Value::as_str) - .unwrap_or("(question)") - .to_string(); - let options = question - .get("options") - .and_then(Value::as_array) - .into_iter() - .flatten() - .filter_map(|option| option.get("label").and_then(Value::as_str)) - .map(String::from) - .collect(); - events.push(Event::Question { - id: format!("{request_id}#{i}"), - prompt: text.clone(), - options, - }); - questions.push(text); - } - } else { - let summary = serde_json::to_string_pretty(&input).unwrap_or_default(); - let summary: String = summary.chars().take(600).collect(); - events.push(Event::Question { - id: request_id.clone(), - prompt: format!("Allow {tool_name}?\n{summary}"), - options: vec!["Allow".to_string(), "Deny".to_string()], - }); - } - self.pending.insert( - request_id.clone(), - PendingRequest { - request_id, - input, - questions, - answers: HashMap::new(), - }, - ); - events.push(Event::Status { - state: SessionStatus::AwaitingInput, - }); - events - } - - /// Applies one answer from the phone. Question ids are the control - /// request id, suffixed `#i` for AskUserQuestion sub-questions. - fn answer(&mut self, question_id: &str, answer: &str) -> AnswerOutcome { - let (request_id, sub) = match question_id.split_once('#') { - Some((request_id, index)) => (request_id, index.parse::().ok()), - None => (question_id, None), - }; - let Some(pending) = self.pending.get_mut(request_id) else { - return AnswerOutcome::Unknown; - }; - - let response = if let Some(index) = sub { - let Some(question) = pending.questions.get(index) else { - return AnswerOutcome::Unknown; - }; - pending.answers.insert(question.clone(), answer.to_string()); - if pending.answers.len() < pending.questions.len() { - return AnswerOutcome::Pending; - } - let mut updated = pending.input.clone(); - updated["answers"] = serde_json::to_value(&pending.answers).expect("string map"); - json!({"behavior": "allow", "updatedInput": updated}) - } else if answer.eq_ignore_ascii_case("deny") { - json!({"behavior": "deny", "message": "The user denied this from the phone."}) - } else { - json!({"behavior": "allow", "updatedInput": pending.input}) - }; - - let request_id = pending.request_id.clone(); - self.pending.remove(&request_id); - AnswerOutcome::Respond(json!({ - "type": "control_response", - "response": {"subtype": "success", "request_id": request_id, "response": response}, - })) - } - - /// `user` messages: tool results become ToolEnd, with any image parts - /// saved into the session dir and referenced by an Image event (the - /// phone fetches them from `/sessions/{id}/files/{ref}`). Replayed and - /// synthetic user text is skipped -- the manager already recorded the - /// user's side. - fn translate_user(&self, message: &Value) -> Vec { - let Some(content) = message["message"].get("content").and_then(Value::as_array) else { - return Vec::new(); - }; - let mut events = Vec::new(); - for block in content { - if block.get("type").and_then(Value::as_str) != Some("tool_result") { - continue; - } - let mut texts = Vec::new(); - match block.get("content") { - Some(Value::String(text)) => texts.push(text.clone()), - Some(Value::Array(parts)) => { - for part in parts { - match part.get("type").and_then(Value::as_str) { - Some("text") => { - if let Some(text) = part.get("text").and_then(Value::as_str) { - texts.push(text.to_string()); - } - } - Some("image") => { - if let Some(name) = self.save_image(part) { - events.push(Event::Image { image: name }); - } - } - _ => {} - } - } - } - _ => {} - } - events.push(Event::ToolEnd { - id: block - .get("tool_use_id") - .and_then(Value::as_str) - .unwrap_or_default() - .to_string(), - output: texts.join("\n"), - }); - } - events - } - - /// Decodes one base64 image block into `files/` and returns its ref. - fn save_image(&self, part: &Value) -> Option { - let source = part.get("source")?; - let data = source.get("data")?.as_str()?; - use base64::Engine; - let bytes = base64::engine::general_purpose::STANDARD - .decode(data) - .ok()?; - // Screenshots are the overwhelming case, and they are PNG; an - // unrecognized type is more likely a dialect change than a JPEG. - let extension = source - .get("media_type") - .and_then(Value::as_str) - .and_then(crate::media::extension_for) - .unwrap_or("png"); - let name = format!("{}.{extension}", super::random_hex()); - let dir = self.session_dir.join("files"); - if let Err(err) = crate::private::create_dir(&dir) - .map_err(std::io::Error::other) - .and_then(|()| std::fs::write(dir.join(&name), bytes)) - { - tracing::error!("couldn't save produced image: {err}"); - return None; - } - Some(name) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn translate_lines(translator: &mut Translator, lines: &[&str]) -> Vec { - lines - .iter() - .flat_map(|line| translator.translate(&serde_json::from_str(line).expect("json"))) - .collect() - } - - #[test] - fn captures_the_resume_token_from_init() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"system","subtype":"init","cwd":"/x","session_id":"5ecf21da-d53f","tools":[],"model":"claude-haiku-4-5-20251001"}"#, - ], - ); - assert!(events.is_empty()); - assert_eq!(translator.session_id.as_deref(), Some("5ecf21da-d53f")); - } - - #[test] - fn streams_text_deltas_and_skips_the_consolidated_copy() { - // Real lines (trimmed) from the 2.1.237 probe. - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Done."}},"session_id":"s","parent_tool_use_id":null}"#, - r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"Done."}]},"parent_tool_use_id":null,"session_id":"s"}"#, - r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hmm"}},"session_id":"s","parent_tool_use_id":null}"#, - ], - ); - assert_eq!( - events, - vec![Event::AssistantText { - delta: "Done.".to_string() - }] - ); - } - - #[test] - fn tool_use_and_result_become_tool_events() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_01","name":"Bash","input":{"command":"echo probe-ok"}}]},"parent_tool_use_id":null}"#, - r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01","type":"tool_result","content":"probe-ok","is_error":false}]},"parent_tool_use_id":null}"#, - ], - ); - assert_eq!( - events, - vec![ - Event::ToolStart { - id: "toolu_01".to_string(), - tool: "Bash".to_string(), - input: serde_json::json!({"command": "echo probe-ok"}), - }, - Event::ToolEnd { - id: "toolu_01".to_string(), - output: "probe-ok".to_string() - }, - ] - ); - } - - #[test] - fn subagent_events_are_not_duplicated_into_the_transcript() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_02","name":"Bash","input":{}}]},"parent_tool_use_id":"toolu_parent"}"#, - ], - ); - assert!(events.is_empty()); - } - - #[test] - fn a_permission_request_becomes_an_allow_deny_question() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"control_request","request_id":"req-1","request":{"subtype":"can_use_tool","tool_name":"Bash","input":{"command":"rm -rf /tmp/x"},"tool_use_id":"toolu_03"}}"#, - ], - ); - let Event::Question { - id, - prompt, - options, - } = &events[0] - else { - panic!("expected a question, got {events:?}"); - }; - assert_eq!(id, "req-1"); - assert!(prompt.contains("Bash") && prompt.contains("rm -rf /tmp/x")); - assert_eq!(options, &["Allow", "Deny"]); - assert_eq!( - events[1], - Event::Status { - state: SessionStatus::AwaitingInput - } - ); - - // Allowing echoes the input back; the request is then gone. - let AnswerOutcome::Respond(response) = translator.answer("req-1", "Allow") else { - panic!("expected a control response"); - }; - assert_eq!(response["response"]["request_id"], "req-1"); - assert_eq!(response["response"]["response"]["behavior"], "allow"); - assert_eq!( - response["response"]["response"]["updatedInput"]["command"], - "rm -rf /tmp/x" - ); - assert!(matches!( - translator.answer("req-1", "Allow"), - AnswerOutcome::Unknown - )); - } - - #[test] - fn denying_a_permission_sends_deny() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - translate_lines( - &mut translator, - &[ - r#"{"type":"control_request","request_id":"req-2","request":{"subtype":"can_use_tool","tool_name":"Write","input":{"file_path":"/etc/passwd"}}}"#, - ], - ); - let AnswerOutcome::Respond(response) = translator.answer("req-2", "Deny") else { - panic!("expected a control response"); - }; - assert_eq!(response["response"]["response"]["behavior"], "deny"); - } - - #[test] - fn ask_user_question_rides_the_same_flow_with_answers_keyed_by_question() { - // The real 2.1.237 shape, verified live: answers go back inside - // updatedInput, keyed by the question text. - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"control_request","request_id":"req-3","request":{"subtype":"can_use_tool","tool_name":"AskUserQuestion","input":{"questions":[{"question":"Which color?","header":"Color","options":[{"label":"Red"},{"label":"Blue"}],"multiSelect":false},{"question":"Which size?","header":"Size","options":[{"label":"S"},{"label":"L"}],"multiSelect":false}]},"tool_use_id":"toolu_04","requires_user_interaction":true}}"#, - ], - ); - let questions: Vec<_> = events - .iter() - .filter_map(|event| match event { - Event::Question { - id, - prompt, - options, - } => Some((id.clone(), prompt.clone(), options.clone())), - _ => None, - }) - .collect(); - assert_eq!(questions.len(), 2); - assert_eq!(questions[0].0, "req-3#0"); - assert_eq!(questions[0].1, "Which color?"); - assert_eq!(questions[0].2, vec!["Red", "Blue"]); - - // First answer alone isn't enough; the response goes out when the - // last sub-question is answered, with all answers aboard. - assert!(matches!( - translator.answer("req-3#0", "Blue"), - AnswerOutcome::Pending - )); - let AnswerOutcome::Respond(response) = translator.answer("req-3#1", "L") else { - panic!("expected a control response"); - }; - let updated = &response["response"]["response"]["updatedInput"]; - assert_eq!(updated["answers"]["Which color?"], "Blue"); - assert_eq!(updated["answers"]["Which size?"], "L"); - assert_eq!(updated["questions"][0]["question"], "Which color?"); - } - - #[test] - fn images_in_tool_results_are_saved_and_referenced() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - // A 1x1 PNG, the smallest real payload worth round-tripping. - let png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="; - let line = format!( - r#"{{"type":"user","message":{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"toolu_05","content":[{{"type":"text","text":"took a screenshot"}},{{"type":"image","source":{{"type":"base64","media_type":"image/png","data":"{png}"}}}}]}}]}},"parent_tool_use_id":null}}"# - ); - let events = translator.translate(&serde_json::from_str(&line).expect("json")); - - let Event::Image { image } = &events[0] else { - panic!("expected an image event, got {events:?}"); - }; - assert!(image.ends_with(".png")); - let saved = dir.path().join("files").join(image); - assert!(saved.is_file(), "image not saved at {}", saved.display()); - assert_eq!( - events[1], - Event::ToolEnd { - id: "toolu_05".to_string(), - output: "took a screenshot".to_string(), - } - ); - } - - #[test] - fn a_turn_result_reports_usage_and_returns_to_idle() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"result","subtype":"success","is_error":false,"num_turns":2,"stop_reason":"end_turn","session_id":"s","total_cost_usd":0.0149,"usage":{"input_tokens":18,"output_tokens":164}}"#, - ], - ); - assert_eq!( - events, - vec![ - Event::UsageDelta { tokens: 182 }, - Event::Status { - state: SessionStatus::Idle - }, - ] - ); - } - - #[test] - fn an_error_result_surfaces_the_message() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"something broke","usage":{}}"#, - ], - ); - assert_eq!( - events[0], - Event::Error { - message: "something broke".to_string() - } - ); - assert_eq!( - *events.last().unwrap(), - Event::Status { - state: SessionStatus::Idle - } - ); - } - - #[test] - fn replayed_and_synthetic_user_text_is_skipped() { - let dir = tempfile::tempdir().expect("tempdir"); - let mut translator = Translator::new(dir.path().to_path_buf()); - let events = translate_lines( - &mut translator, - &[ - r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"hi"}]},"isReplay":true,"parent_tool_use_id":null}"#, - r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"[continue]"}]},"isSynthetic":true,"parent_tool_use_id":null}"#, - ], - ); - assert!(events.is_empty()); - } -} diff --git a/server/src/session/claude/translate.rs b/server/src/session/claude/translate.rs new file mode 100644 index 0000000..3ac8baa --- /dev/null +++ b/server/src/session/claude/translate.rs @@ -0,0 +1,628 @@ +//! The stream-json dialect: CLI lines in, common [`Event`]s out. +//! +//! Split from the driver beside it because the two change for unrelated +//! reasons. This half moves when the CLI's wire format does -- a new +//! message subtype, a field that changed shape -- and that is what the +//! tests at the bottom pin, replaying recorded lines. The driver half +//! moves when spawning, resuming or shutting down changes, and never +//! reads a line itself. +//! +//! The one side effect here is saving images a tool result carries into +//! the session directory (they would bloat the transcript as base64); +//! everything else is pure, which is what makes the mapping testable +//! without a process. + +use std::collections::HashMap; +use std::path::PathBuf; + +use serde_json::{Value, json}; + +use super::super::driver::{Event, SessionStatus}; + +/// What answering a question produced. +pub(super) enum AnswerOutcome { + /// Send this control_response line to the CLI. + Respond(Value), + /// Part of a multi-question request; more answers still needed. + Pending, + Unknown, +} + +/// A `can_use_tool` request we've surfaced to the phone and not yet +/// answered. For plain permissions there is one implicit question +/// (Allow/Deny); for AskUserQuestion, one per entry in `questions`. +struct PendingRequest { + request_id: String, + input: Value, + /// Question text per sub-question, in order -- the keys the answers + /// map uses. Empty for a plain permission request. + questions: Vec, + answers: HashMap, +} + +/// Translation state: stream-json lines in, common events out. The one +/// side effect is saving images a tool result carries into the session +/// dir (they'd bloat the transcript as base64); everything else is pure, +/// so the dialect mapping is unit-testable from recorded lines. +pub(super) struct Translator { + pub(super) session_id: Option, + pending: HashMap, + session_dir: PathBuf, +} + +impl Translator { + pub(super) fn new(session_dir: PathBuf) -> Self { + Self { + session_id: None, + pending: HashMap::new(), + session_dir, + } + } + pub(super) fn translate(&mut self, message: &Value) -> Vec { + // Events from subagents (Task tool internals) carry a + // parent_tool_use_id; the transcript shows the Task tool's own + // start/end instead of every nested step. + if message + .get("parent_tool_use_id") + .is_some_and(|id| !id.is_null()) + { + return Vec::new(); + } + match message.get("type").and_then(Value::as_str) { + Some("system") => { + if message.get("subtype").and_then(Value::as_str) == Some("init") + && let Some(id) = message.get("session_id").and_then(Value::as_str) + { + self.session_id = Some(id.to_string()); + } + Vec::new() + } + Some("stream_event") => self.translate_stream_event(&message["event"]), + Some("assistant") => self.translate_assistant(&message["message"]), + Some("user") => self.translate_user(message), + Some("control_request") => self.translate_control_request(message), + Some("control_response") => { + let response = &message["response"]; + if response.get("subtype").and_then(Value::as_str) == Some("error") { + let error = response + .get("error") + .and_then(Value::as_str) + .unwrap_or("unknown"); + vec![Event::Error { + message: format!("claude rejected a request: {error}"), + }] + } else { + Vec::new() + } + } + Some("result") => { + let usage = &message["usage"]; + let tokens = usage + .get("input_tokens") + .and_then(Value::as_u64) + .unwrap_or(0) + + usage + .get("output_tokens") + .and_then(Value::as_u64) + .unwrap_or(0); + let mut events = Vec::new(); + if message + .get("is_error") + .and_then(Value::as_bool) + .unwrap_or(false) + { + events.push(Event::Error { + message: message + .get("result") + .and_then(Value::as_str) + .unwrap_or("the turn ended with an error") + .to_string(), + }); + } + if tokens > 0 { + events.push(Event::UsageDelta { tokens }); + } + events.push(Event::Status { + state: SessionStatus::Idle, + }); + events + } + _ => Vec::new(), + } + } + + /// Raw API streaming: only text deltas become events. Consolidated + /// blocks arriving later re-carry the same text, so those are skipped + /// in `translate_assistant` -- one source per fact. + fn translate_stream_event(&mut self, event: &Value) -> Vec { + if event.get("type").and_then(Value::as_str) == Some("content_block_delta") + && let Some(delta) = event["delta"].get("text") + && event["delta"].get("type").and_then(Value::as_str) == Some("text_delta") + && let Some(text) = delta.as_str() + { + return vec![Event::AssistantText { + delta: text.to_string(), + }]; + } + Vec::new() + } + + fn translate_assistant(&mut self, message: &Value) -> Vec { + let Some(content) = message.get("content").and_then(Value::as_array) else { + return Vec::new(); + }; + content + .iter() + .filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use")) + .map(|block| Event::ToolStart { + id: block + .get("id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + tool: block + .get("name") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + input: block.get("input").cloned().unwrap_or(Value::Null), + }) + .collect() + } + + fn translate_control_request(&mut self, message: &Value) -> Vec { + let request = &message["request"]; + if request.get("subtype").and_then(Value::as_str) != Some("can_use_tool") { + return Vec::new(); + } + let request_id = message + .get("request_id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let tool_name = request + .get("tool_name") + .and_then(Value::as_str) + .unwrap_or("a tool"); + let input = request.get("input").cloned().unwrap_or(Value::Null); + + let mut events = Vec::new(); + let mut questions = Vec::new(); + if tool_name == "AskUserQuestion" { + for (i, question) in input + .get("questions") + .and_then(Value::as_array) + .into_iter() + .flatten() + .enumerate() + { + let text = question + .get("question") + .and_then(Value::as_str) + .unwrap_or("(question)") + .to_string(); + let options = question + .get("options") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|option| option.get("label").and_then(Value::as_str)) + .map(String::from) + .collect(); + events.push(Event::Question { + id: format!("{request_id}#{i}"), + prompt: text.clone(), + options, + }); + questions.push(text); + } + } else { + let summary = serde_json::to_string_pretty(&input).unwrap_or_default(); + let summary: String = summary.chars().take(600).collect(); + events.push(Event::Question { + id: request_id.clone(), + prompt: format!("Allow {tool_name}?\n{summary}"), + options: vec!["Allow".to_string(), "Deny".to_string()], + }); + } + self.pending.insert( + request_id.clone(), + PendingRequest { + request_id, + input, + questions, + answers: HashMap::new(), + }, + ); + events.push(Event::Status { + state: SessionStatus::AwaitingInput, + }); + events + } + + /// Applies one answer from the phone. Question ids are the control + /// request id, suffixed `#i` for AskUserQuestion sub-questions. + pub(super) fn answer(&mut self, question_id: &str, answer: &str) -> AnswerOutcome { + let (request_id, sub) = match question_id.split_once('#') { + Some((request_id, index)) => (request_id, index.parse::().ok()), + None => (question_id, None), + }; + let Some(pending) = self.pending.get_mut(request_id) else { + return AnswerOutcome::Unknown; + }; + + let response = if let Some(index) = sub { + let Some(question) = pending.questions.get(index) else { + return AnswerOutcome::Unknown; + }; + pending.answers.insert(question.clone(), answer.to_string()); + if pending.answers.len() < pending.questions.len() { + return AnswerOutcome::Pending; + } + let mut updated = pending.input.clone(); + updated["answers"] = serde_json::to_value(&pending.answers).expect("string map"); + json!({"behavior": "allow", "updatedInput": updated}) + } else if answer.eq_ignore_ascii_case("deny") { + json!({"behavior": "deny", "message": "The user denied this from the phone."}) + } else { + json!({"behavior": "allow", "updatedInput": pending.input}) + }; + + let request_id = pending.request_id.clone(); + self.pending.remove(&request_id); + AnswerOutcome::Respond(json!({ + "type": "control_response", + "response": {"subtype": "success", "request_id": request_id, "response": response}, + })) + } + + /// `user` messages: tool results become ToolEnd, with any image parts + /// saved into the session dir and referenced by an Image event (the + /// phone fetches them from `/sessions/{id}/files/{ref}`). Replayed and + /// synthetic user text is skipped -- the manager already recorded the + /// user's side. + fn translate_user(&self, message: &Value) -> Vec { + let Some(content) = message["message"].get("content").and_then(Value::as_array) else { + return Vec::new(); + }; + let mut events = Vec::new(); + for block in content { + if block.get("type").and_then(Value::as_str) != Some("tool_result") { + continue; + } + let mut texts = Vec::new(); + match block.get("content") { + Some(Value::String(text)) => texts.push(text.clone()), + Some(Value::Array(parts)) => { + for part in parts { + match part.get("type").and_then(Value::as_str) { + Some("text") => { + if let Some(text) = part.get("text").and_then(Value::as_str) { + texts.push(text.to_string()); + } + } + Some("image") => { + if let Some(name) = self.save_image(part) { + events.push(Event::Image { image: name }); + } + } + _ => {} + } + } + } + _ => {} + } + events.push(Event::ToolEnd { + id: block + .get("tool_use_id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + output: texts.join("\n"), + }); + } + events + } + + /// Decodes one base64 image block into `files/` and returns its ref. + fn save_image(&self, part: &Value) -> Option { + let source = part.get("source")?; + let data = source.get("data")?.as_str()?; + use base64::Engine; + let bytes = base64::engine::general_purpose::STANDARD + .decode(data) + .ok()?; + // Screenshots are the overwhelming case, and they are PNG; an + // unrecognized type is more likely a dialect change than a JPEG. + let extension = source + .get("media_type") + .and_then(Value::as_str) + .and_then(crate::media::extension_for) + .unwrap_or("png"); + let name = format!("{}.{extension}", super::super::random_hex()); + let dir = self.session_dir.join("files"); + if let Err(err) = crate::private::create_dir(&dir) + .map_err(std::io::Error::other) + .and_then(|()| std::fs::write(dir.join(&name), bytes)) + { + tracing::error!("couldn't save produced image: {err}"); + return None; + } + Some(name) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn translate_lines(translator: &mut Translator, lines: &[&str]) -> Vec { + lines + .iter() + .flat_map(|line| translator.translate(&serde_json::from_str(line).expect("json"))) + .collect() + } + + #[test] + fn captures_the_resume_token_from_init() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"system","subtype":"init","cwd":"/x","session_id":"5ecf21da-d53f","tools":[],"model":"claude-haiku-4-5-20251001"}"#, + ], + ); + assert!(events.is_empty()); + assert_eq!(translator.session_id.as_deref(), Some("5ecf21da-d53f")); + } + + #[test] + fn streams_text_deltas_and_skips_the_consolidated_copy() { + // Real lines (trimmed) from the 2.1.237 probe. + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Done."}},"session_id":"s","parent_tool_use_id":null}"#, + r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"Done."}]},"parent_tool_use_id":null,"session_id":"s"}"#, + r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hmm"}},"session_id":"s","parent_tool_use_id":null}"#, + ], + ); + assert_eq!( + events, + vec![Event::AssistantText { + delta: "Done.".to_string() + }] + ); + } + + #[test] + fn tool_use_and_result_become_tool_events() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_01","name":"Bash","input":{"command":"echo probe-ok"}}]},"parent_tool_use_id":null}"#, + r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01","type":"tool_result","content":"probe-ok","is_error":false}]},"parent_tool_use_id":null}"#, + ], + ); + assert_eq!( + events, + vec![ + Event::ToolStart { + id: "toolu_01".to_string(), + tool: "Bash".to_string(), + input: serde_json::json!({"command": "echo probe-ok"}), + }, + Event::ToolEnd { + id: "toolu_01".to_string(), + output: "probe-ok".to_string() + }, + ] + ); + } + + #[test] + fn subagent_events_are_not_duplicated_into_the_transcript() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_02","name":"Bash","input":{}}]},"parent_tool_use_id":"toolu_parent"}"#, + ], + ); + assert!(events.is_empty()); + } + + #[test] + fn a_permission_request_becomes_an_allow_deny_question() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"control_request","request_id":"req-1","request":{"subtype":"can_use_tool","tool_name":"Bash","input":{"command":"rm -rf /tmp/x"},"tool_use_id":"toolu_03"}}"#, + ], + ); + let Event::Question { + id, + prompt, + options, + } = &events[0] + else { + panic!("expected a question, got {events:?}"); + }; + assert_eq!(id, "req-1"); + assert!(prompt.contains("Bash") && prompt.contains("rm -rf /tmp/x")); + assert_eq!(options, &["Allow", "Deny"]); + assert_eq!( + events[1], + Event::Status { + state: SessionStatus::AwaitingInput + } + ); + + // Allowing echoes the input back; the request is then gone. + let AnswerOutcome::Respond(response) = translator.answer("req-1", "Allow") else { + panic!("expected a control response"); + }; + assert_eq!(response["response"]["request_id"], "req-1"); + assert_eq!(response["response"]["response"]["behavior"], "allow"); + assert_eq!( + response["response"]["response"]["updatedInput"]["command"], + "rm -rf /tmp/x" + ); + assert!(matches!( + translator.answer("req-1", "Allow"), + AnswerOutcome::Unknown + )); + } + + #[test] + fn denying_a_permission_sends_deny() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + translate_lines( + &mut translator, + &[ + r#"{"type":"control_request","request_id":"req-2","request":{"subtype":"can_use_tool","tool_name":"Write","input":{"file_path":"/etc/passwd"}}}"#, + ], + ); + let AnswerOutcome::Respond(response) = translator.answer("req-2", "Deny") else { + panic!("expected a control response"); + }; + assert_eq!(response["response"]["response"]["behavior"], "deny"); + } + + #[test] + fn ask_user_question_rides_the_same_flow_with_answers_keyed_by_question() { + // The real 2.1.237 shape, verified live: answers go back inside + // updatedInput, keyed by the question text. + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"control_request","request_id":"req-3","request":{"subtype":"can_use_tool","tool_name":"AskUserQuestion","input":{"questions":[{"question":"Which color?","header":"Color","options":[{"label":"Red"},{"label":"Blue"}],"multiSelect":false},{"question":"Which size?","header":"Size","options":[{"label":"S"},{"label":"L"}],"multiSelect":false}]},"tool_use_id":"toolu_04","requires_user_interaction":true}}"#, + ], + ); + let questions: Vec<_> = events + .iter() + .filter_map(|event| match event { + Event::Question { + id, + prompt, + options, + } => Some((id.clone(), prompt.clone(), options.clone())), + _ => None, + }) + .collect(); + assert_eq!(questions.len(), 2); + assert_eq!(questions[0].0, "req-3#0"); + assert_eq!(questions[0].1, "Which color?"); + assert_eq!(questions[0].2, vec!["Red", "Blue"]); + + // First answer alone isn't enough; the response goes out when the + // last sub-question is answered, with all answers aboard. + assert!(matches!( + translator.answer("req-3#0", "Blue"), + AnswerOutcome::Pending + )); + let AnswerOutcome::Respond(response) = translator.answer("req-3#1", "L") else { + panic!("expected a control response"); + }; + let updated = &response["response"]["response"]["updatedInput"]; + assert_eq!(updated["answers"]["Which color?"], "Blue"); + assert_eq!(updated["answers"]["Which size?"], "L"); + assert_eq!(updated["questions"][0]["question"], "Which color?"); + } + + #[test] + fn images_in_tool_results_are_saved_and_referenced() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + // A 1x1 PNG, the smallest real payload worth round-tripping. + let png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="; + let line = format!( + r#"{{"type":"user","message":{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"toolu_05","content":[{{"type":"text","text":"took a screenshot"}},{{"type":"image","source":{{"type":"base64","media_type":"image/png","data":"{png}"}}}}]}}]}},"parent_tool_use_id":null}}"# + ); + let events = translator.translate(&serde_json::from_str(&line).expect("json")); + + let Event::Image { image } = &events[0] else { + panic!("expected an image event, got {events:?}"); + }; + assert!(image.ends_with(".png")); + let saved = dir.path().join("files").join(image); + assert!(saved.is_file(), "image not saved at {}", saved.display()); + assert_eq!( + events[1], + Event::ToolEnd { + id: "toolu_05".to_string(), + output: "took a screenshot".to_string(), + } + ); + } + + #[test] + fn a_turn_result_reports_usage_and_returns_to_idle() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"result","subtype":"success","is_error":false,"num_turns":2,"stop_reason":"end_turn","session_id":"s","total_cost_usd":0.0149,"usage":{"input_tokens":18,"output_tokens":164}}"#, + ], + ); + assert_eq!( + events, + vec![ + Event::UsageDelta { tokens: 182 }, + Event::Status { + state: SessionStatus::Idle + }, + ] + ); + } + + #[test] + fn an_error_result_surfaces_the_message() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"something broke","usage":{}}"#, + ], + ); + assert_eq!( + events[0], + Event::Error { + message: "something broke".to_string() + } + ); + assert_eq!( + *events.last().unwrap(), + Event::Status { + state: SessionStatus::Idle + } + ); + } + + #[test] + fn replayed_and_synthetic_user_text_is_skipped() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"hi"}]},"isReplay":true,"parent_tool_use_id":null}"#, + r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"[continue]"}]},"isSynthetic":true,"parent_tool_use_id":null}"#, + ], + ); + assert!(events.is_empty()); + } +}