Merge branch 'main' into acp-mode-indicator

Author: CSRessel

.github/workflows/docs.md

@@ -58,6 +58,7 @@ validate -> test -> build-native (matrix: 4 targets) -> stage-npm -> create-next
 - Main branch `@next` publishes reuse the exact same version detection, tag creation, npm publish, and GitHub Release creation code paths as manual `publish_next` dispatches
 - The `dry_run` input only applies to `workflow_dispatch` -- tag pushes and main branch pushes always publish for real
 - Build runners use Blacksmith (e.g., `blacksmith-4vcpu-ubuntu-2404`) for most jobs, with standard `ubuntu-24.04` for npm publish (which needs the `npm-publish` environment)
+- Rust CI caches Cargo registry and git dependency data, but intentionally does not cache `~/.cargo/bin/`; restoring cached binaries after toolchain setup can overwrite rustup-managed `cargo` shims on hosted runners.
 - Git tags are the source of truth for all version numbering -- both the `validate` job (via `create_nori_release --get-next-version`) and the `create_nori_release` script's `determine_version()` function use `list_tags()` to enumerate existing versions; GitHub Releases are not consulted for version counting
 
 Created and maintained by Nori.

.github/workflows/rust-ci.yml

@@ -63,7 +63,9 @@ jobs:
           echo "hash=${{ hashFiles('nori-rs/Cargo.lock') }}" >> "$GITHUB_OUTPUT"
           echo "toolchain_hash=${{ hashFiles('nori-rs/rust-toolchain.toml') }}" >> "$GITHUB_OUTPUT"
 
-      # Cache cargo home directory (registry, git deps, etc.)
+      # Cache registry and git deps only. Do not cache ~/.cargo/bin because
+      # restoring tool binaries after rust-toolchain setup can overwrite
+      # rustup-managed cargo shims on hosted runners.
       - name: Restore cargo home cache
         id: cache_cargo_home_restore
         uses: actions/cache/restore@v4

nori-rs/acp/docs.md

@@ -79,7 +79,7 @@ ACP session-domain state now flows through a single serialized reducer. `Session
 `SessionRuntime` is the authoritative model for:
 - whether the ACP session is idle, loading, or in a prompt turn
 - queued user prompts and compact prompts waiting behind an active request
-- request-local message assembly for assistant/reasoning streams
+- request-local message assembly for user/assistant/reasoning streams, including flushing the prior open text buffer when the ACP session update type changes
 - tool snapshot ownership via `owner_request_id`
 - pending permission request ownership and cancellation cleanup
 - final assistant message extraction used for `PromptCompleted { last_agent_message, .. }`
@@ -89,7 +89,7 @@ The live backend path in `user_input.rs`, `submit_and_ops.rs`, `spawn_and_relay.
 
 Metadata notifications that ACP permits while idle are treated as session-owned rather than request-owned. `AvailableCommandsUpdate`, `CurrentModeUpdate`, `ConfigOptionUpdate`, `SessionInfoUpdate`, and `UsageUpdate` no longer produce "no request is active" warnings; instead the reducer persists the latest values and forwards normalized `ClientEvent`s downstream.
 
-`session/load` replay also preserves more session context than before. User-side `MessageDelta { stream: User, .. }` values are reassembled into `ReplayEntry::UserMessage`, while `SessionUpdateInfo` notes pass through unchanged. For usage updates, that replay path now restores the structured footer context state without needing to re-render the verbose message in history.
+`session/load` replay also preserves more session context than before. User-side `MessageDelta { stream: User, .. }` values are reassembled into `ReplayEntry::UserMessage`, while `SessionUpdateInfo` notes pass through unchanged. Message replay preserves chronological stream-kind boundaries: an answer -> reasoning -> answer sequence becomes three replay entries, while adjacent deltas of the same stream are still coalesced. For usage updates, that replay path now restores the structured footer context state without needing to re-render the verbose message in history.
 
 **Custom Agent TOML Schema** (`config/types/mod.rs`):
 
@@ -740,6 +740,8 @@ Approval request         → client_event entry
 
 Older `tool_call`, `tool_result`, and `patch_apply` transcript entry types remain in the schema for legacy read compatibility, but ACP live recording now uses normalized `ClientEvent` entries so transcript persistence matches the live TUI path.
 
+Reducer-owned transcript assembly preserves ACP session update type boundaries. When text changes from assistant to reasoning, reasoning to assistant, or user to either agent stream, the previous open message is flushed before the new stream is accumulated. Consecutive chunks with the same stream are still treated as one message because stable ACP does not provide a durable same-type message boundary.
+
 Tool output for non-patch `tool_result` entries is truncated to 10,000 bytes when recording to transcript. All string truncation helpers in the crate -- `truncate_for_log()` in `tool_display.rs` (tracing previews), `truncate_str()` in `translator.rs` (tool-call display labels like "Execute: ..."), and the transcript byte truncation -- use `codex_utils_string::take_bytes_at_char_boundary()` to avoid slicing inside multi-byte UTF-8 characters.
 
 Configuration:

nori-rs/acp/src/backend/session_reducer.rs

@@ -598,6 +598,7 @@ fn reduce_permission_request(
 // Message assembly
 // ---------------------------------------------------------------------------
 
+#[derive(Copy, Clone, PartialEq, Eq)]
 enum MessageKind {
     Agent,
     Thought,
@@ -609,15 +610,17 @@ fn append_chunk_to_open_message(
     chunk: &acp::ContentChunk,
     kind: MessageKind,
 ) {
-    let Some(active) = &mut runtime.active else {
-        return;
-    };
-
     let text = match &chunk.content {
         acp::ContentBlock::Text(t) => &t.text,
         _ => return,
     };
 
+    flush_open_messages_except(runtime, kind);
+
+    let Some(active) = &mut runtime.active else {
+        return;
+    };
+
     let open = match kind {
         MessageKind::Agent => active
             .open_agent_message
@@ -633,6 +636,33 @@ fn append_chunk_to_open_message(
     open.chunks.push(text.clone());
 }
 
+fn flush_open_messages_except(runtime: &mut SessionRuntime, keep: MessageKind) {
+    let (user, thought, agent) = {
+        let Some(active) = &mut runtime.active else {
+            return;
+        };
+        (
+            (keep != MessageKind::User)
+                .then(|| active.open_user_message.take())
+                .flatten(),
+            (keep != MessageKind::Thought)
+                .then(|| active.open_thought_message.take())
+                .flatten(),
+            (keep != MessageKind::Agent)
+                .then(|| active.open_agent_message.take())
+                .flatten(),
+        )
+    };
+
+    push_open_transcript_message(runtime, TranscriptRole::User, user);
+    push_open_transcript_message(runtime, TranscriptRole::Thought, thought);
+    if let Some(text) = push_open_transcript_message(runtime, TranscriptRole::Agent, agent)
+        && let Some(active) = &mut runtime.active
+    {
+        active.last_agent_message = Some(text);
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Active request finalization
 // ---------------------------------------------------------------------------
@@ -641,41 +671,43 @@ fn append_chunk_to_open_message(
 /// transcript, clear active, and return the last agent message text.
 fn finalize_active(runtime: &mut SessionRuntime) -> Option<String> {
     let active = runtime.active.take()?;
-    let mut last_agent_message = None;
-
-    // Finalize open messages in order: user, thought, agent.
-    if let Some(open) = active.open_user_message {
-        let text = open.text();
-        if !text.is_empty() {
-            runtime.persisted.transcript.push(TranscriptMessage {
-                role: TranscriptRole::User,
-                content: text,
-            });
-        }
-    }
-    if let Some(open) = active.open_thought_message {
-        let text = open.text();
-        if !text.is_empty() {
-            runtime.persisted.transcript.push(TranscriptMessage {
-                role: TranscriptRole::Thought,
-                content: text,
-            });
-        }
-    }
-    if let Some(open) = active.open_agent_message {
-        let text = open.text();
-        if !text.is_empty() {
-            last_agent_message = Some(text.clone());
-            runtime.persisted.transcript.push(TranscriptMessage {
-                role: TranscriptRole::Agent,
-                content: text,
-            });
-        }
+    let mut last_agent_message = active.last_agent_message;
+
+    // At most one text kind should still be open because kind switches flush
+    // the previous buffer as chunks arrive.
+    push_open_transcript_message(runtime, TranscriptRole::User, active.open_user_message);
+    push_open_transcript_message(
+        runtime,
+        TranscriptRole::Thought,
+        active.open_thought_message,
+    );
+    if let Some(text) =
+        push_open_transcript_message(runtime, TranscriptRole::Agent, active.open_agent_message)
+    {
+        last_agent_message = Some(text);
     }
 
     last_agent_message
 }
 
+fn push_open_transcript_message(
+    runtime: &mut SessionRuntime,
+    role: TranscriptRole,
+    open: Option<OpenMessage>,
+) -> Option<String> {
+    let text = open?.text();
+    if text.is_empty() {
+        return None;
+    }
+
+    runtime.persisted.transcript.push(TranscriptMessage {
+        role,
+        content: text.clone(),
+    });
+
+    (role == TranscriptRole::Agent).then_some(text)
+}
+
 // ---------------------------------------------------------------------------
 // Helpers
 // ---------------------------------------------------------------------------

nori-rs/acp/src/backend/session_reducer/tests.rs

@@ -615,6 +615,71 @@ fn multiple_chunks_assembled_into_one_transcript_entry() {
     assert_eq!(agent_messages[0].content, "hello world!");
 }
 
+#[test]
+fn mixed_agent_and_thought_chunks_preserve_transcript_order() {
+    let mut rt = new_runtime();
+    let mut norm = new_normalizer();
+
+    reduce(
+        &mut rt,
+        InboundEvent::PromptSubmit(simple_prompt()),
+        &mut norm,
+    );
+
+    let updates = [
+        acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(acp::ContentBlock::Text(
+            acp::TextContent::new("CI is green."),
+        ))),
+        acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new(acp::ContentBlock::Text(
+            acp::TextContent::new("Preparing PR."),
+        ))),
+        acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(acp::ContentBlock::Text(
+            acp::TextContent::new("The PR is up."),
+        ))),
+    ];
+
+    for update in updates {
+        reduce(&mut rt, notification(update), &mut norm);
+    }
+
+    reduce(
+        &mut rt,
+        InboundEvent::PromptResponse {
+            stop_reason: acp::StopReason::EndTurn,
+        },
+        &mut norm,
+    );
+
+    let transcript: Vec<_> = rt
+        .persisted
+        .transcript
+        .iter()
+        .map(|message| (message.role, message.content.as_str()))
+        .collect();
+
+    assert_eq!(
+        transcript,
+        vec![
+            (
+                nori_protocol::session_runtime::TranscriptRole::User,
+                "hello",
+            ),
+            (
+                nori_protocol::session_runtime::TranscriptRole::Agent,
+                "CI is green.",
+            ),
+            (
+                nori_protocol::session_runtime::TranscriptRole::Thought,
+                "Preparing PR.",
+            ),
+            (
+                nori_protocol::session_runtime::TranscriptRole::Agent,
+                "The PR is up.",
+            ),
+        ]
+    );
+}
+
 // =========================================================================
 // 9. Load lifecycle
 // =========================================================================

nori-rs/acp/src/backend/transcript.rs

@@ -68,46 +68,43 @@ pub fn client_events_to_replay_client_events(
     client_events: Vec<nori_protocol::ClientEvent>,
 ) -> Vec<nori_protocol::ClientEvent> {
     let mut replay = Vec::new();
-    let mut user = String::new();
-    let mut assistant = String::new();
-    let mut reasoning = String::new();
-
-    let flush_buffers = |replay: &mut Vec<nori_protocol::ClientEvent>,
-                         user: &mut String,
-                         assistant: &mut String,
-                         reasoning: &mut String| {
-        if !user.is_empty() {
-            replay.push(nori_protocol::ClientEvent::ReplayEntry(
-                nori_protocol::ReplayEntry::UserMessage {
-                    text: std::mem::take(user),
-                },
-            ));
-        }
-        if !reasoning.is_empty() {
-            replay.push(nori_protocol::ClientEvent::ReplayEntry(
-                nori_protocol::ReplayEntry::ReasoningMessage {
-                    text: std::mem::take(reasoning),
-                },
-            ));
-        }
-        if !assistant.is_empty() {
-            replay.push(nori_protocol::ClientEvent::ReplayEntry(
-                nori_protocol::ReplayEntry::AssistantMessage {
-                    text: std::mem::take(assistant),
-                },
-            ));
+    let mut current_stream: Option<nori_protocol::MessageStream> = None;
+    let mut current_text = String::new();
+
+    let flush_message = |replay: &mut Vec<nori_protocol::ClientEvent>,
+                         stream: &mut Option<nori_protocol::MessageStream>,
+                         text: &mut String| {
+        let Some(stream) = stream.take() else {
+            return;
+        };
+        if text.is_empty() {
+            return;
         }
+
+        let text = std::mem::take(text);
+        let entry = match stream {
+            nori_protocol::MessageStream::User => nori_protocol::ReplayEntry::UserMessage { text },
+            nori_protocol::MessageStream::Answer => {
+                nori_protocol::ReplayEntry::AssistantMessage { text }
+            }
+            nori_protocol::MessageStream::Reasoning => {
+                nori_protocol::ReplayEntry::ReasoningMessage { text }
+            }
+        };
+        replay.push(nori_protocol::ClientEvent::ReplayEntry(entry));
     };
 
     for event in client_events {
         match event {
-            nori_protocol::ClientEvent::MessageDelta(message_delta) => match message_delta.stream {
-                nori_protocol::MessageStream::User => user.push_str(&message_delta.delta),
-                nori_protocol::MessageStream::Answer => assistant.push_str(&message_delta.delta),
-                nori_protocol::MessageStream::Reasoning => reasoning.push_str(&message_delta.delta),
-            },
+            nori_protocol::ClientEvent::MessageDelta(message_delta) => {
+                if current_stream.as_ref() != Some(&message_delta.stream) {
+                    flush_message(&mut replay, &mut current_stream, &mut current_text);
+                    current_stream = Some(message_delta.stream);
+                }
+                current_text.push_str(&message_delta.delta);
+            }
             other => {
-                flush_buffers(&mut replay, &mut user, &mut assistant, &mut reasoning);
+                flush_message(&mut replay, &mut current_stream, &mut current_text);
                 if let Some(replay_entry) = replay_entry_from_client_event(&other) {
                     replay.push(nori_protocol::ClientEvent::ReplayEntry(replay_entry));
                 } else if should_pass_through_replay_client_event(&other) {
@@ -117,7 +114,7 @@ pub fn client_events_to_replay_client_events(
         }
     }
 
-    flush_buffers(&mut replay, &mut user, &mut assistant, &mut reasoning);
+    flush_message(&mut replay, &mut current_stream, &mut current_text);
     replay
 }
 
@@ -386,6 +383,45 @@ mod tests {
         );
     }
 
+    #[test]
+    fn client_events_to_replay_client_events_preserves_mixed_message_delta_order() {
+        let replay = client_events_to_replay_client_events(vec![
+            nori_protocol::ClientEvent::MessageDelta(nori_protocol::MessageDelta {
+                stream: nori_protocol::MessageStream::Answer,
+                delta: "CI is green.".into(),
+            }),
+            nori_protocol::ClientEvent::MessageDelta(nori_protocol::MessageDelta {
+                stream: nori_protocol::MessageStream::Reasoning,
+                delta: "Preparing PR.".into(),
+            }),
+            nori_protocol::ClientEvent::MessageDelta(nori_protocol::MessageDelta {
+                stream: nori_protocol::MessageStream::Answer,
+                delta: "The PR is up.".into(),
+            }),
+        ]);
+
+        assert_eq!(
+            replay,
+            vec![
+                nori_protocol::ClientEvent::ReplayEntry(
+                    nori_protocol::ReplayEntry::AssistantMessage {
+                        text: "CI is green.".into(),
+                    },
+                ),
+                nori_protocol::ClientEvent::ReplayEntry(
+                    nori_protocol::ReplayEntry::ReasoningMessage {
+                        text: "Preparing PR.".into(),
+                    },
+                ),
+                nori_protocol::ClientEvent::ReplayEntry(
+                    nori_protocol::ReplayEntry::AssistantMessage {
+                        text: "The PR is up.".into(),
+                    },
+                ),
+            ]
+        );
+    }
+
     #[test]
     fn transcript_to_replay_client_events_preserves_session_update_info() {
         let transcript = make_transcript(vec![TranscriptEntry::ClientEvent(ClientEventEntry {