Record a steer where the model read it, not where it was typed

A message sent while an answer was streaming was recorded in the middle
of that answer and above the tool call it ended with. The model had
committed to that call in the same message it was already writing, so it
had read none of it -- and on screen the tool result underneath read as
something the steer had asked for. The answer also split into two
bubbles around a message that was not part of it.

The driver announced a steer at "the next assistant text or tool call",
on the reasoning that anything the CLI says next is proof it has been
round the model again. With --include-partial-messages that is not true:
the deltas and the tool_use block of a message already in flight keep
arriving afterwards, and none of them saw the steer.

`message_start` is what actually proves it. The CLI sends the previous
call's tool results back before it opens the next assistant message, so
that line is the first moment anything written since can have been read
-- and it carries no events of its own, which is what makes it a place
to put one. Verified against 2.1.237: message_start, the blocks, the
tool_result, then the next message_start.

The end of the turn stays as the other half, and is the case that must
not be lost: a message typed after the final model call has no later
message_start, and one that is only recorded when announced would
otherwise vanish while a phone drew it as still waiting.

Checked live on haiku, before and after. Before: the steer landed at
seq 37 among the essay's deltas, with the tool call at 45 and its result
at 46. After: essay whole, tool call 42, result 43, steer 44. Also
checked the case this had no reason to touch -- a steer sent during a
30-second Bash call, which was already correct -- and it still records
after the result. The two tests fail on the old rule; the failure prints
the old order, which is the bug.

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:
+/// line goes out immediately and the *announcement* waits here instead,
-/// the next assistant text or tool call is proof another model call has
+/// until the CLI opens the next model call -- see
-/// happened, and the message was in it. That keeps a phone's held bubble
+/// [`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<String> = {
-                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`]
+/// was asked to do. Deliberately a wider set than what announces a steer
-/// with the ends swapped: that one takes the `Idle` that closes a turn
+/// (see [`announce_steers`]): any sign of work proves a turn is running,
-/// and this one takes the states that open one. `Idle` is the pair to
+/// while only a `message_start` proves a line written a moment ago has
-/// this -- it is where `running` goes back to false, a few lines above
+/// been read. `Idle` is the pair to this -- it is where `running` goes
-/// where it is set here.
+/// 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
+/// Records every message written since the last announcement, in the
-/// since the last one did.
+/// order it was written. False means the session has been torn down.
 ///
-/// Assistant output and a tool call both mean another model call happened;
+/// Called from the two places that prove the CLI has consumed them: the
-/// an idle means the turn is over and nothing further is coming. Status
+/// start of a new model call, and the end of the turn. Both are in
-/// changes that are not idle prove nothing -- a turn can go `running`
+/// [`translate_line`], and the pair is the whole of the rule -- a steer
-/// without having read a line written a moment ago.
+/// announced anywhere else lands above output that predates it.
-fn announces_a_steer(event: &Event) -> bool {
+fn announce_steers(queue: &Arc<Mutex<Queue>>, sink: &EventSink) -> bool {
-    matches!(
+    let taken: Vec<String> = {
-        event,
+        let mut queue = queue.lock().unwrap();
-        Event::AssistantText { .. }
+        queue.awaiting.drain(..).collect()
-            | Event::ToolStart { .. }
+    };
-            | Event::Status {
+    for text in taken {
-                state: SessionStatus::Idle
+        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<Mutex<Queue>>),
+    ) -> Vec<Event> {
+        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::<Event>();
+        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.
     ///

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.