@sigitcodereview / sigit / commits / 40a5a74

Put a messageId on every message chunk

Agent message and thought chunks now carry a messageId, so a client can tell where one message ends and the next begins instead of guessing. Each inference round is one message: its reply text and its reasoning share an id, and the round after a tool call starts a new one. A notice sent on its own gets its own id, and replayed user and agent messages get fresh ones on session/load. Ids are random, so nothing new has to be stored in the history. Closes #195 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Seto Elkahfi committed Oct 9, 2026 at 00:16 UTC 40a5a746928c364852812b279c393fe4b001f8c0

3 files changed +172 -15

.agents/AGENTS.md

@@ -118,6 +118,10 @@ feeds results back. Neither the loop nor ACP/TUI surfaces depend on a concrete b

118 `SessionState::remote`), so its turn can wait on the endpoint while another thread is

119 installed. Unknown session ids are rejected instead of silently borrowing the active thread's

120 cwd. Prompt cancellation takes no lock, which lets a client cancel a turn whatever it holds.

121 + Message chunks carry a `messageId`: each inference round is one message, so every chunk of

122 + it, reasoning included, shares `StreamedReply::message_id` and a tool round starts a new id.

123 + A standalone notice (`send_assistant_message`) gets an id of its own, and replayed messages

124 + get fresh ones; ids are random, so none is stored in history.

125 `session/load` and `session/resume` share `restore_session`; the only difference is that

126 resume must not replay the history as `session/update`. `session/close` is the one place a

127 `SessionState` is dropped. It runs in two halves: `begin_close` signals the session's turn

src/main.rs

+82 -15

@@ -81,7 +81,7 @@ use agent_client_protocol::schema::v1::{

81 DeleteSessionResponse, Diff, EmbeddedResourceResource, ForkSessionRequest, ForkSessionResponse,

82 Implementation, InitializeRequest, InitializeResponse, KillTerminalRequest,

83 ListSessionsRequest, ListSessionsResponse, LoadSessionRequest, LoadSessionResponse,

84 - LogoutCapabilities, LogoutRequest, LogoutResponse, McpCapabilities, McpServer, Meta,

84 + LogoutCapabilities, LogoutRequest, LogoutResponse, McpCapabilities, McpServer, MessageId, Meta,

85 NewSessionRequest, NewSessionResponse, PermissionOption, PermissionOptionKind, Plan, PlanEntry,

86 PlanEntryPriority, PlanEntryStatus, PromptCapabilities, PromptRequest, PromptResponse,

87 ReadTextFileRequest, ReleaseTerminalRequest, RequestPermissionOutcome,

@@ -375,6 +375,10 @@ struct StreamedReply {

375 streamed_any: bool,

376 /// Whether the next visible text should open a new paragraph.

377 paragraph_pending: bool,

378 + /// The ACP `messageId` of the model message being streamed. Each

379 + /// inference round is its own message in history, so a tool round

380 + /// interrupting the prose starts a new one.

381 + message_id: Option<MessageId>,

382 }

383

384 /// What a streamed fragment newly revealed once `<think>` blocks were split:

@@ -392,6 +396,13 @@ impl StreamedReply {

396 /// no sentence to break away from, hence the `streamed_any` guard.

397 fn interrupt(&mut self) {

398 self.paragraph_pending = self.streamed_any;

399 + self.message_id = None;

400 + }

401 +

402 + /// The id every chunk of the current model message carries, its reply

403 + /// text and its reasoning alike.

404 + fn message_id(&mut self) -> MessageId {

405 + self.message_id.get_or_insert_with(new_message_id).clone()

406 }

407

408 /// Fold one streamed fragment into the reply and return the text it newly

@@ -449,6 +460,12 @@ impl StreamedReply {

460 }

461 }

462

463 +/// A fresh ACP `messageId`. Ids are opaque and only have to be unique within

464 +/// a session, so a random one serves both live chunks and replayed ones.

465 +fn new_message_id() -> MessageId {

466 + MessageId::new(format!("msg_{}", uuid::Uuid::new_v4().simple()))

467 +}

468 +

469 #[derive(Default)]

470 struct PromptCancellation {

471 cancelled: AtomicBool,

@@ -671,17 +688,17 @@ fn history_replay_updates(history: &[serde_json::Value]) -> Vec<SessionUpdate> {

688 "user" => {

689 let text = history_message_text(message);

690 if !text.trim().is_empty() {

674 - updates.push(SessionUpdate::UserMessageChunk(ContentChunk::new(

675 - ContentBlock::from(text),

676 - )));

691 + updates.push(SessionUpdate::UserMessageChunk(

692 + ContentChunk::new(ContentBlock::from(text)).message_id(new_message_id()),

693 + ));

694 }

695 }

696 "assistant" => {

697 let (_think, visible) = chat::strip_think_blocks(&history_message_text(message));

698 if !visible.trim().is_empty() {

682 - updates.push(SessionUpdate::AgentMessageChunk(ContentChunk::new(

683 - ContentBlock::from(visible),

684 - )));

699 + updates.push(SessionUpdate::AgentMessageChunk(

700 + ContentChunk::new(ContentBlock::from(visible)).message_id(new_message_id()),

701 + ));

702 }

703 for call in message["tool_calls"].as_array().into_iter().flatten() {

704 let name = call["function"]["name"].as_str().unwrap_or("tool");

@@ -1569,29 +1586,50 @@ impl SiGitAgent {

1586 Ok(())

1587 }

1588

1589 + /// Send a message of its own: a notice, a slash command's answer. It gets

1590 + /// a fresh `messageId`, so a client never folds it into the reply around

1591 + /// it. Text that belongs to the model's reply goes through

1592 + /// [`Self::send_reply_chunk`] instead.

1593 fn send_assistant_message(

1594 &self,

1595 cx: &ConnectionTo<Client>,

1596 session_id: SessionId,

1597 text: impl Into<String>,

1598 + ) -> agent_client_protocol::Result<()> {

1599 + self.send_reply_chunk(cx, session_id, new_message_id(), text)

1600 + }

1601 +

1602 + /// Send one chunk of the model message `message_id`.

1603 + fn send_reply_chunk(

1604 + &self,

1605 + cx: &ConnectionTo<Client>,

1606 + session_id: SessionId,

1607 + message_id: MessageId,

1608 + text: impl Into<String>,

1609 ) -> agent_client_protocol::Result<()> {

1610 cx.send_notification(SessionNotification::new(

1611 session_id,

1580 - SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::from(text.into()))),

1612 + SessionUpdate::AgentMessageChunk(

1613 + ContentChunk::new(ContentBlock::from(text.into())).message_id(message_id),

1614 + ),

1615 ))

1616 }

1617

1618 /// Stream a fragment of the model's reasoning as a thought chunk. Clients

1619 /// render these in a collapsed "thinking" section, separate from the reply.

1620 + /// It carries the id of the model message it came with.

1621 fn send_agent_thought(

1622 &self,

1623 cx: &ConnectionTo<Client>,

1624 session_id: SessionId,

1625 + message_id: MessageId,

1626 text: impl Into<String>,

1627 ) -> agent_client_protocol::Result<()> {

1628 cx.send_notification(SessionNotification::new(

1629 session_id,

1594 - SessionUpdate::AgentThoughtChunk(ContentChunk::new(ContentBlock::from(text.into()))),

1630 + SessionUpdate::AgentThoughtChunk(

1631 + ContentChunk::new(ContentBlock::from(text.into())).message_id(message_id),

1632 + ),

1633 ))

1634 }

1635

@@ -1671,16 +1709,20 @@ impl SiGitAgent {

1709 ) {

1710 match chunk {

1711 TokenChunk::Reasoning(text) => {

1674 - self.send_agent_thought(cx, session_id.clone(), text).ok();

1712 + let message_id = reply.message_id();

1713 + self.send_agent_thought(cx, session_id.clone(), message_id, text)

1714 + .ok();

1715 }

1716 TokenChunk::Visible(piece) => {

1717 let revealed = reply.push(&piece);

1718 if let Some(reasoning) = revealed.reasoning {

1679 - self.send_agent_thought(cx, session_id.clone(), reasoning)

1719 + let message_id = reply.message_id();

1720 + self.send_agent_thought(cx, session_id.clone(), message_id, reasoning)

1721 .ok();

1722 }

1723 if let Some(visible) = revealed.visible {

1683 - self.send_assistant_message(cx, session_id.clone(), visible)

1724 + let message_id = reply.message_id();

1725 + self.send_reply_chunk(cx, session_id.clone(), message_id, visible)

1726 .ok();

1727 }

1728 }

@@ -3894,7 +3936,8 @@ impl SiGitAgent {

3936 };

3937

3938 if !final_text.is_empty() {

3897 - self.send_assistant_message(cx, session_id.clone(), final_text)

3939 + let message_id = reply.message_id();

3940 + self.send_reply_chunk(cx, session_id.clone(), message_id, final_text)

3941 .ok();

3942 }

3943 } else if round > 0 && !round_cap_reached && reply.sent.len() == sent_before_last_round {

@@ -3907,9 +3950,11 @@ impl SiGitAgent {

3950 session_id,

3951 round

3952 );

3910 - self.send_assistant_message(

3953 + let message_id = reply.message_id();

3954 + self.send_reply_chunk(

3955 cx,

3956 session_id.clone(),

3957 + message_id,

3958 format!(

3959 "{PARAGRAPH_BREAK}{}",

3960 silent_stop_message(stopped_repeating.as_deref(), round)

@@ -3924,9 +3969,11 @@ impl SiGitAgent {

3969 session_id,

3970 max_tool_rounds

3971 );

3927 - self.send_assistant_message(

3972 + let message_id = reply.message_id();

3973 + self.send_reply_chunk(

3974 cx,

3975 session_id.clone(),

3976 + message_id,

3977 format!(

3978 "{PARAGRAPH_BREAK}{}",

3979 round_cap_stop_message(max_tool_rounds)

@@ -7186,6 +7233,26 @@ mod tests {

7233 other => panic!("expected the tool call, got {other:?}"),

7234 }

7235 assert!(matches!(updates[3], SessionUpdate::AgentMessageChunk(_)));

7236 +

7237 + // Each replayed message has an id of its own.

7238 + let ids: Vec<_> = updates

7239 + .iter()

7240 + .filter_map(|update| match update {

7241 + SessionUpdate::UserMessageChunk(chunk)

7242 + | SessionUpdate::AgentMessageChunk(chunk) => Some(

7243 + chunk

7244 + .message_id

7245 + .clone()

7246 + .expect("replayed chunk has a messageId"),

7247 + ),

7248 + _ => None,

7249 + })

7250 + .collect();

7251 + assert_eq!(ids.len(), 3);

7252 + assert!(

7253 + ids[0] != ids[1] && ids[1] != ids[2] && ids[0] != ids[2],

7254 + "{ids:?}"

7255 + );

7256 }

7257

7258 #[test]

tests/acp_permissions.rs

+86

@@ -1139,6 +1139,92 @@ fn text_from_two_tool_rounds_reaches_the_client_as_separate_paragraphs() {

1139 let _ = std::fs::remove_dir_all(&scratch);

1140 }

1141

1142 +/// Every chunk of one model message carries the same `messageId`, its

1143 +/// reasoning included, and the message after a tool round gets a new one

1144 +/// (issue #195).

1145 +#[test]

1146 +fn chunks_of_one_model_message_share_a_message_id() {

1147 + let endpoint = start_fake_endpoint(vec![

1148 + sse_body(&[

1149 + json!({"choices": [{"delta": {"reasoning_content": "Where is it?"}}]}),

1150 + json!({"choices": [{"delta": {"content": "I'll look "}}]}),

1151 + json!({"choices": [{"delta": {"content": "around."}}]}),

1152 + json!({

1153 + "choices": [{"delta": {"tool_calls": [{

1154 + "index": 0,

1155 + "id": "call_1",

1156 + "function": {"name": "list_directory", "arguments": r#"{"path":"."}"#},

1157 + }]}}]

1158 + }),

1159 + ]),

1160 + sse_text("Found it."),

1161 + ]);

1162 +

1163 + let scratch = std::env::temp_dir().join(format!("sigit_acp_msg_ids_{}", std::process::id()));

1164 + let config_dir = scratch.join("config");

1165 + let cwd = scratch.join("cwd");

1166 + std::fs::create_dir_all(&config_dir).unwrap();

1167 + std::fs::create_dir_all(&cwd).unwrap();

1168 +

1169 + let mut agent = spawn_agent(endpoint.port, &config_dir);

1170 +

1171 + let id = agent.request(

1172 + "initialize",

1173 + json!({"protocolVersion": 1, "clientCapabilities": {}}),

1174 + );

1175 + agent.wait_for_response(id);

1176 +

1177 + let id = agent.request("session/new", json!({"cwd": cwd, "mcpServers": []}));

1178 + let session_id = agent.wait_for_response(id)["result"]["sessionId"]

1179 + .as_str()

1180 + .expect("session id")

1181 + .to_string();

1182 +

1183 + let prompt_id = agent.request(

1184 + "session/prompt",

1185 + json!({

1186 + "sessionId": session_id,

1187 + "prompt": [{"type": "text", "text": "find it"}],

1188 + }),

1189 + );

1190 + let (_, updates) = agent.wait_for_response_with_updates(prompt_id);

1191 +

1192 + let chunks: Vec<(&str, &str)> = updates

1193 + .iter()

1194 + .filter(|update| {

1195 + update["sessionUpdate"] == "agent_message_chunk"

1196 + || update["sessionUpdate"] == "agent_thought_chunk"

1197 + })

1198 + .map(|update| {

1199 + (

1200 + update["content"]["text"].as_str().unwrap_or_default(),

1201 + update["messageId"]

1202 + .as_str()

1203 + .unwrap_or_else(|| panic!("chunk without a messageId: {update}")),

1204 + )

1205 + })

1206 + .collect();

1207 + let id_of = |text: &str| {

1208 + chunks

1209 + .iter()

1210 + .find(|(chunk, _)| chunk.contains(text))

1211 + .unwrap_or_else(|| panic!("no chunk with {text:?} in {chunks:?}"))

1212 + .1

1213 + };

1214 +

1215 + let first = id_of("Where is it?");

1216 + assert_eq!(id_of("I'll look"), first, "{chunks:?}");

1217 + assert_eq!(id_of("around."), first, "{chunks:?}");

1218 + assert_ne!(

1219 + id_of("Found it."),

1220 + first,

1221 + "the round after the tool call is a new message: {chunks:?}"

1222 + );

1223 +

1224 + drop(agent);

1225 + let _ = std::fs::remove_dir_all(&scratch);

1226 +}

1227 +

1228 /// A model that writes its tool call out as literal `<tool_call>` text instead

1229 /// of using the structured field must still drive the loop. Before recovery

1230 /// the tag was streamed to the client as prose and the turn ended with no tool