iris/ai-app
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Separate the Claude dialect from the Claude process

Browse Source 
claude.rs held two things that change for unrelated reasons. One spawns the
CLI, resumes it with --resume after a crash, writes lines to it and shuts it
down; the other turns a stream-json line into common events. A CLI wire
format change touches only the second, a change to how sessions are
launched only the first, and at 830 lines a reader had to work out which
half they were in.

So the translator, the pending-request bookkeeping and the answer outcome
move to session/claude/translate.rs, and all twelve tests go with them --
every one was already a translation test, replaying recorded lines with no
process involved, which is the clearest evidence the seam was already
there. 366 lines and 628, from 830 plus tests in one file.

Pure code motion: no behaviour, no renames, and the only edits are the
visibility the split makes necessary. The probing record stays in the
driver file, since it is the provenance for both halves -- the flags are
that file's, the message catalogue is what translate implements.

Verified: 35 tests pass (the same 35), clippy clean, fmt clean, and cargo
doc resolves with broken_intra_doc_links denied.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017xn8nHw1tw1R6PtiY1eEtw
This commit is contained in:
irisandClaude Opus 5 committed 2026-08-28 03:43:01 -04:00
1 parent 585e4a0369
commit 4e760a4a72
2 files changed
+640 -609
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/session/claude.rs
+12 -609
View File
@@ -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 <id>` 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<Value> {
}
}))
}
/// 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<String>,
answers: HashMap<String, String>,
}
/// 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<String>,
pending: HashMap<String, PendingRequest>,
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<Event> {
// 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<Event> {
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<Event> {
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<Event> {
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::<usize>().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<Event> {
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<String> {
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<Event> {
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());
}
}
server/src/session/claude/translate.rs
+628
View File
@@ -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<String>,
answers: HashMap<String, String>,
}
/// 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<String>,
pending: HashMap<String, PendingRequest>,
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<Event> {
// 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<Event> {
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<Event> {
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<Event> {
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::<usize>().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<Event> {
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<String> {
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<Event> {
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());
}
}