diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index 897e354..42d4896 100644 --- a/server/src/session/claude.rs +++ b/server/src/session/claude.rs @@ -61,7 +61,7 @@ use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus}; use super::process; use super::transport::{Launch, Streams, Transport}; use crate::config::{ProviderConfig, SessionConfig}; -use translate::{AnswerOutcome, Setting, Translator}; +use translate::{AnswerOutcome, Setting, Translator, starts_a_model_call}; /// How much of a failing process's stderr the exit report carries. /// @@ -118,11 +118,20 @@ const STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5); /// inside the same turn, with one `result` for the whole thing. /// /// What the CLI does not do is say on stdout that it has read one. So the -/// line goes out immediately and the *announcement* waits here instead: -/// the next assistant text or tool call is proof another model call has -/// happened, and the message was in it. That keeps a phone's held bubble +/// line goes out immediately and the *announcement* waits here instead, +/// until the CLI opens the next model call -- see +/// [`translate::starts_a_model_call`]. That keeps a phone's held bubble /// where it belongs -- below the working indicator until the session has /// actually taken it -- without delaying the message itself to get it. +/// +/// The proof has to be the model call and not the output, which is what +/// an earlier version took it to be. Assistant text and a tool call both +/// keep arriving from a message that was *already in flight* when the +/// steer was written, and that message saw none of it: a steer sent +/// while an answer was streaming was recorded in the middle of it, above +/// tool calls the model had already committed to. On screen the answer +/// split into two bubbles around a message it had not read, and the tool +/// results that followed read as things the steer had asked for. #[derive(Default)] struct Queue { /// A turn is in flight, so a message sent now is a steer into it. @@ -790,6 +799,7 @@ fn translate_line( tracing::warn!("unparseable claude output line: {shown}"); return true; }; + let opens_a_model_call = starts_a_model_call(&message); let (events, new_session_id) = { let mut state = state.lock().unwrap(); let before = state.session_id.clone(); @@ -800,29 +810,17 @@ fn translate_line( if let Some(session_id) = new_session_id { write_resume_token(session_dir, &session_id); } + // The steer is announced where the CLI opens the model call that read + // it, and the announcement goes out *before* that call's output, so + // the message sits above what it produced and below what it did not. + // + // This line carries no events of its own, which is what makes it the + // right place: everything the previous call produced -- its text, its + // tool calls, their results -- is already recorded above. + if opens_a_model_call && !announce_steers(queue, sink) { + return false; + } for event in events { - // Anything the CLI says after a steer was written is proof it has - // been round the model again, and the steer went with it -- so - // that is the moment it is announced, and the moment a phone can - // stop drawing it as still waiting. The announcement goes out - // *before* the event that proves it, so the message is above the - // output it produced rather than below it. - // - // A turn ending counts too, and is the case that must not be - // missed: a message written after the last model call of a turn - // has no later output to prove anything, and without this it would - // never be announced at all. - if announces_a_steer(&event) { - let taken: Vec = { - let mut queue = queue.lock().unwrap(); - queue.awaiting.drain(..).collect() - }; - for text in taken { - if sink.send(Event::MessageTaken { text }).is_err() { - return false; - } - } - } // A turn nobody here started -- see `proves_a_turn`. Said before // the event that proves it, for the same reason a steer is: the // session was already working when it produced this. @@ -849,6 +847,14 @@ fn translate_line( state: SessionStatus::Idle } ) { + // The case that must not be missed: a message written after + // the final model call of a turn has no later `message_start` + // to prove anything, so without this it would never be + // announced at all. The end of the turn is where it belongs + // anyway -- nothing above it came after the message. + if !announce_steers(queue, sink) { + return false; + } queue.lock().unwrap().running = false; } if sink.send(event).is_err() { @@ -871,11 +877,11 @@ fn translate_line( /// reading as idle until the turn ends. /// /// So the driver says it from what it observes rather than from what it -/// was asked to do, and this is the same set as [`announces_a_steer`] -/// with the ends swapped: that one takes the `Idle` that closes a turn -/// and this one takes the states that open one. `Idle` is the pair to -/// this -- it is where `running` goes back to false, a few lines above -/// where it is set here. +/// was asked to do. Deliberately a wider set than what announces a steer +/// (see [`announce_steers`]): any sign of work proves a turn is running, +/// while only a `message_start` proves a line written a moment ago has +/// been read. `Idle` is the pair to this -- it is where `running` goes +/// back to false, a few lines above where it is set here. fn proves_a_turn(event: &Event) -> bool { matches!( event, @@ -891,22 +897,24 @@ fn proves_a_turn(event: &Event) -> bool { ) } -/// Whether this event proves the CLI has consumed anything written to it -/// since the last one did. +/// Records every message written since the last announcement, in the +/// order it was written. False means the session has been torn down. /// -/// Assistant output and a tool call both mean another model call happened; -/// an idle means the turn is over and nothing further is coming. Status -/// changes that are not idle prove nothing -- a turn can go `running` -/// without having read a line written a moment ago. -fn announces_a_steer(event: &Event) -> bool { - matches!( - event, - Event::AssistantText { .. } - | Event::ToolStart { .. } - | Event::Status { - state: SessionStatus::Idle - } - ) +/// Called from the two places that prove the CLI has consumed them: the +/// start of a new model call, and the end of the turn. Both are in +/// [`translate_line`], and the pair is the whole of the rule -- a steer +/// announced anywhere else lands above output that predates it. +fn announce_steers(queue: &Arc>, sink: &EventSink) -> bool { + let taken: Vec = { + let mut queue = queue.lock().unwrap(); + queue.awaiting.drain(..).collect() + }; + for text in taken { + if sink.send(Event::MessageTaken { text }).is_err() { + return false; + } + } + true } /// The end of the stderr log, for an exit report a person reads. @@ -1036,6 +1044,145 @@ mod tests { events } + /// Feeds lines through the reader, running `interject` between two of + /// them, and returns what came out. + /// + /// The hook is what makes a steer testable at all: what matters is + /// not which events a line produces but *where* a message written + /// part-way through the stream ends up among them. + fn events_with_interjection( + lines: &[&str], + after: usize, + interject: impl FnOnce(&Arc>), + ) -> Vec { + let dir = tempfile::tempdir().expect("temp dir"); + let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf()))); + let queue = Arc::new(Mutex::new(Queue::default())); + let (sink, mut out) = mpsc::unbounded_channel::(); + let mut interject = Some(interject); + for (i, line) in lines.iter().enumerate() { + assert!(translate_line(line, dir.path(), &state, &sink, &queue)); + if i == after { + interject.take().expect("one interjection")(&queue); + } + } + drop(sink); + let mut events = Vec::new(); + while let Ok(event) = out.try_recv() { + events.push(event); + } + events + } + + /// One assistant message, streamed: two text deltas, then the + /// `tool_use` it ends with, then that call's result. + /// + /// Written out rather than shortened because the point of both tests + /// below is the *order*, and the shape of a real turn is what makes + /// the order mean anything. Recorded from 2.1.237. + const STREAMED_CALL: &[&str] = &[ + r#"{"type":"stream_event","event":{"type":"message_start"},"session_id":"s","parent_tool_use_id":null}"#, + r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me "}},"session_id":"s","parent_tool_use_id":null}"#, + r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"check that."}},"session_id":"s","parent_tool_use_id":null}"#, + r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_01","name":"Bash","input":{"command":"echo one"}}]},"parent_tool_use_id":null}"#, + r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01","type":"tool_result","content":"one","is_error":false}]},"parent_tool_use_id":null}"#, + ]; + + /// A steer typed while an answer is streaming is recorded below that + /// answer's tool call and its result, not among them. + /// + /// The message reaches the CLI immediately; what waits is saying so. + /// Everything the CLI emits after it was typed still belongs to a + /// model call that had not read it -- the rest of the text, the + /// `tool_use` the model had already committed to, the result that + /// came back. `message_start` is the first line that proves the next + /// call has it, so that is where the announcement goes. + #[test] + fn a_steer_is_recorded_below_the_call_that_had_not_read_it() { + let mut lines = STREAMED_CALL.to_vec(); + lines.push( + r#"{"type":"stream_event","event":{"type":"message_start"},"session_id":"s","parent_tool_use_id":null}"#, + ); + lines.push( + r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Doing that instead."}},"session_id":"s","parent_tool_use_id":null}"#, + ); + // Typed after the first delta, with the answer still arriving. + let events = events_with_interjection(&lines, 1, |queue| { + queue + .lock() + .unwrap() + .awaiting + .push_back("do the other one instead".into()) + }); + + let at = |find: fn(&Event) -> bool| { + events + .iter() + .position(find) + .unwrap_or_else(|| panic!("nothing matched in {events:?}")) + }; + let taken = at(|e| matches!(e, Event::MessageTaken { .. })); + assert!( + taken > at(|e| matches!(e, Event::ToolStart { .. })), + "a steer must not sit above a call the model had already made: {events:?}" + ); + assert!( + taken > at(|e| matches!(e, Event::ToolEnd { .. })), + "a steer must not sit above the result of that call: {events:?}" + ); + assert_eq!( + events + .iter() + .filter(|e| matches!(e, Event::AssistantText { .. })) + .count(), + 3, + "the streamed answer must stay whole: {events:?}" + ); + } + + /// A steer written after the turn's last model call is still recorded. + /// + /// Nothing further is coming, so no `message_start` will ever prove + /// it was read -- and a message that is only recorded when announced + /// would otherwise vanish, leaving a phone drawing it as still + /// waiting forever. The end of the turn is also where it belongs: + /// nothing above it happened after it was typed. + #[test] + fn a_steer_with_no_model_call_left_is_recorded_at_the_end_of_the_turn() { + let mut lines = STREAMED_CALL.to_vec(); + lines.push( + r#"{"type":"result","subtype":"success","usage":{"input_tokens":1,"output_tokens":1}}"#, + ); + // Typed after the tool result, with only the turn's end to come. + let events = events_with_interjection(&lines, 4, |queue| { + queue + .lock() + .unwrap() + .awaiting + .push_back("never mind".into()) + }); + + let taken = events + .iter() + .position(|e| matches!(e, Event::MessageTaken { .. })) + .unwrap_or_else(|| panic!("a steer must never be dropped: {events:?}")); + let idle = events + .iter() + .position(|e| { + matches!( + e, + Event::Status { + state: SessionStatus::Idle + } + ) + }) + .unwrap_or_else(|| panic!("expected the turn to end: {events:?}")); + assert!( + taken < idle, + "the steer belongs inside the turn it was typed into: {events:?}" + ); + } + /// The divider comes from the CLI announcing the reset, not from an /// `init` arriving. /// diff --git a/server/src/session/claude/translate.rs b/server/src/session/claude/translate.rs index 11d396b..b677ed0 100644 --- a/server/src/session/claude/translate.rs +++ b/server/src/session/claude/translate.rs @@ -19,6 +19,24 @@ use serde_json::{Value, json}; use super::super::driver::{Event, QuestionOption, SessionStatus}; +/// Whether this line is the CLI opening a fresh model call. +/// +/// `message_start` begins one assistant message, and the CLI sends the +/// previous call's tool results back before it opens the next -- so this +/// is the first moment at which anything written since the last one can +/// have been read. Nothing earlier will do: the text deltas and the +/// `tool_use` block of a message *already in flight* keep arriving after +/// a steer is written, and none of them saw it. +/// +/// Only present because the driver passes `--include-partial-messages`. +/// Without it there are no `stream_event` lines at all and this is never +/// true, which is why the caller keeps a fallback that does not depend on +/// it. +pub(super) fn starts_a_model_call(message: &Value) -> bool { + message.get("type").and_then(Value::as_str) == Some("stream_event") + && message["event"].get("type").and_then(Value::as_str) == Some("message_start") +} + /// What answering a question produced. pub(super) enum AnswerOutcome { /// Send this control_response line to the CLI.