diff --git a/server/src/cursor/conversation/output.rs b/server/src/cursor/conversation/output.rs index 7fe905f..f86a2a3 100644 --- a/server/src/cursor/conversation/output.rs +++ b/server/src/cursor/conversation/output.rs @@ -431,21 +431,28 @@ impl ConversationOutput { streams.clear(); } if let CommitCause::RuntimeEvent { event_id } = &state.cause { - if let Some(injection_id) = event_id.strip_prefix("inject-context:") { - if let Some(pending) = self.pending_injections.remove(injection_id) - { - let delivered_at_ms = crate::cursor::tools::runtime::now_ms() - .min(i64::MAX as u64) - as i64; - self.handle.emit(&events::context_injection_delivered( - injection_id.to_owned(), - pending.delivery_batch_id.clone(), - delivered_at_ms, - ))?; - if let Some(user_message) = pending.user_message { - self.handle - .emit(&events::user_message_appended(user_message))?; - } + // Injections key `pending_injections` by their raw + // injection id and commit under `inject-context:{id}`, + // while runtime user messages key it by (and commit + // under) the full `user-message:{id}` event id. Strip + // the injection prefix when present and otherwise use + // the event id verbatim so both are cleared and emit + // their delivered/appended events. + let injection_id = event_id + .strip_prefix("inject-context:") + .unwrap_or(event_id.as_str()); + if let Some(pending) = self.pending_injections.remove(injection_id) { + let delivered_at_ms = crate::cursor::tools::runtime::now_ms() + .min(i64::MAX as u64) + as i64; + self.handle.emit(&events::context_injection_delivered( + injection_id.to_owned(), + pending.delivery_batch_id.clone(), + delivered_at_ms, + ))?; + if let Some(user_message) = pending.user_message { + self.handle + .emit(&events::user_message_appended(user_message))?; } } } diff --git a/server/src/cursor/tools/codec/render.rs b/server/src/cursor/tools/codec/render.rs index 06f3857..d212576 100644 --- a/server/src/cursor/tools/codec/render.rs +++ b/server/src/cursor/tools/codec/render.rs @@ -210,7 +210,7 @@ pub fn tool_completed(call: &ToolCall, completion: &ToolCompletion) -> pb::Agent pub fn tool_placeholder(name: &str, call_id: &str) -> Result { use pb::tool_call::Tool; let tool = match normalized(name).as_str() { - "shell" => Tool::ShellToolCall(pb::ShellToolCall::default()), + "shell" | "bash" => Tool::ShellToolCall(pb::ShellToolCall::default()), "delete" => Tool::DeleteToolCall(pb::DeleteToolCall::default()), "glob" => Tool::GlobToolCall(pb::GlobToolCall::default()), "grep" => Tool::GrepToolCall(pb::GrepToolCall::default()), @@ -570,3 +570,23 @@ fn now_ms() -> u64 { .unwrap_or_default() .as_millis() as u64 } + +#[cfg(test)] +mod tests { + use super::tool_placeholder; + use crate::cursor::protocol::proto::agent::v1 as pb; + + #[test] + fn bash_renders_as_a_shell_placeholder() { + // The dispatcher treats `bash`/`Bash` as a Shell alias, so the streaming + // placeholder must too; otherwise a `Bash` tool call aborts the turn with + // `unsupported tool: bash` before it ever runs. + for name in ["shell", "Shell", "bash", "Bash"] { + let tool = tool_placeholder(name, "call-1").unwrap().tool; + assert!( + matches!(tool, Some(pb::tool_call::Tool::ShellToolCall(_))), + "{name} should render as a Shell tool" + ); + } + } +} diff --git a/server/src/cursor/tools/codec/request.rs b/server/src/cursor/tools/codec/request.rs index e16dfab..60ec9f0 100644 --- a/server/src/cursor/tools/codec/request.rs +++ b/server/src/cursor/tools/codec/request.rs @@ -35,7 +35,7 @@ pub fn request(id: u32, call: &ToolCall, context: &ExecContext) -> Result { + "shell" | "bash" => { let command = string("command")?; let (simple_commands, parsing_result) = shell_command_metadata(&command); Message::ShellStreamArgs(pb::ShellArgs { @@ -520,3 +520,37 @@ fn prost_value(value: &Value) -> prost_types::Value { }; ProstValue { kind: Some(kind) } } + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::request; + use crate::cursor::protocol::proto::agent::v1 as pb; + use crate::cursor::tools::runtime::ExecContext; + use crate::model::ToolCall; + + #[test] + fn bash_is_encoded_as_a_shell_exec_request() { + // The dispatcher routes `bash`/`Bash` to the shell executor, so the + // request codec must encode it as a Shell stream instead of erroring + // with `tool bash is not executed through ExecServerMessage`. + let call = ToolCall { + index: 0, + call_id: "call-1".into(), + model_call_id: "model-1".into(), + name: "Bash".into(), + arguments_text: String::new(), + arguments: json!({ "command": "ls -la" }), + }; + let message = request(1, &call, &ExecContext::default()).unwrap(); + let Some(pb::agent_server_message::Message::ExecServerMessage(exec)) = message.message + else { + panic!("expected an ExecServerMessage"); + }; + let Some(pb::exec_server_message::Message::ShellStreamArgs(args)) = exec.message else { + panic!("expected ShellStreamArgs"); + }; + assert_eq!(args.command, "ls -la"); + } +} diff --git a/server/src/cursor/tools/codec/response.rs b/server/src/cursor/tools/codec/response.rs index 74e127a..0cf889a 100644 --- a/server/src/cursor/tools/codec/response.rs +++ b/server/src/cursor/tools/codec/response.rs @@ -144,7 +144,8 @@ pub async fn stream_closed(id: u32, pending: &CursorToolRuntime) -> Result std::result::Result return Err("old_string was not found in the notebook cell".into()), @@ -241,3 +244,49 @@ fn normalized(value: &str) -> String { .flat_map(char::to_lowercase) .collect() } + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::edit_notebook; + use crate::model::ToolCall; + + fn notebook_call(old_string: &str) -> ToolCall { + ToolCall { + index: 0, + call_id: "call".into(), + model_call_id: "model".into(), + name: "EditNotebook".into(), + arguments_text: String::new(), + arguments: json!({ + "target_notebook": "/notebook.ipynb", + "cell_idx": 0, + "old_string": old_string, + "new_string": "replacement", + }), + } + } + + fn single_cell_notebook() -> String { + json!({ + "cells": [{"cell_type": "code", "source": ["print('hi')\n"]}], + }) + .to_string() + } + + #[test] + fn edit_notebook_rejects_empty_old_string() { + // StrReplace rejects an empty old_string; EditNotebook must do the same + // instead of prepending new_string (empty cell) or reporting a + // misleading "not unique" error (non-empty cell). + let error = edit_notebook(¬ebook_call(""), &single_cell_notebook()).unwrap_err(); + assert_eq!(error, "old_string must not be empty"); + } + + #[test] + fn edit_notebook_replaces_a_unique_old_string() { + let edited = edit_notebook(¬ebook_call("hi"), &single_cell_notebook()).unwrap(); + assert!(edited.contains("print('replacement')")); + } +} diff --git a/server/tests/interrupt.rs b/server/tests/interrupt.rs index b1260f8..0573e43 100644 --- a/server/tests/interrupt.rs +++ b/server/tests/interrupt.rs @@ -506,6 +506,122 @@ async fn runtime_user_message_action_interrupts_and_continues_with_new_message() assert!(history.contains("queued follow-up")); } +#[tokio::test] +async fn runtime_user_message_reports_delivered_and_appended() { + // A runtime user message is queued into `pending_injections` under a + // `user-message:{id}` key, but the commit correlation only handled the + // `inject-context:` prefix, so the entry was never cleared: the client + // never saw Delivered/UserMessageAppended and every later tool round was + // detached (a hang). This asserts the full delivery sequence. + let (_directory, store) = fixtures::temp_store().await; + let provider = fake_provider::FakeProvider::default(); + provider.push_pending(); + provider.push(text_response("continued after user interruption")); + let assets = PromptAssets::load( + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("prompt/cursor") + .as_path(), + ) + .unwrap(); + let registry = TransportRegistry::new( + store, + Arc::new(provider.clone()), + PromptCompiler::new(assets), + ); + let handle = registry + .get_or_create("user-message-events-request") + .await + .unwrap(); + let mut output = handle.subscribe(); + handle + .command(TransportCommand::Append { + seqno: 0, + message: Box::new(client_run_for( + "user-message-events-request", + "user-message-events-conversation", + )), + }) + .await + .unwrap(); + + let mut append_seqno = 1; + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); + while provider.requests().is_empty() { + assert!( + tokio::time::Instant::now() < deadline, + "provider did not start" + ); + if let Ok(Some(frame)) = + tokio::time::timeout(std::time::Duration::from_millis(20), output.recv()).await + { + acknowledge_kv(&handle, &mut append_seqno, &frame).await; + } + } + handle + .command(TransportCommand::Append { + seqno: append_seqno, + message: Box::new(runtime_user_message()), + }) + .await + .unwrap(); + append_seqno += 1; + + let mut protocol_events = Vec::new(); + loop { + let frame = tokio::time::timeout(std::time::Duration::from_secs(5), output.recv()) + .await + .unwrap() + .expect("RunSSE closed before successful EndStream"); + let (flags, payload) = connect::decode_frames(&frame).unwrap().pop().unwrap(); + if flags & connect::END_STREAM_FLAG != 0 { + assert_eq!(payload.as_ref(), b"{}"); + break; + } + let server = pb::AgentServerMessage::decode(payload).unwrap(); + if let Some(pb::agent_server_message::Message::InteractionUpdate(update)) = server.message { + match update.message { + Some(pb::interaction_update::Message::ContextInjectionState(update)) => { + assert_eq!(update.injection_id, "user-message:queued-user"); + match update.state.and_then(|state| state.state) { + Some(pb::context_injection_state::State::Queued(_)) => { + protocol_events.push("queued") + } + Some(pb::context_injection_state::State::Delivered(delivered)) => { + assert!(!delivered.delivery_batch_id.is_empty()); + assert!(delivered.delivered_at_ms > 0); + protocol_events.push("delivered"); + } + _ => {} + } + } + Some(pb::interaction_update::Message::UserMessageAppended(update)) => { + let user = update.user_message.expect("appended user message"); + assert_eq!(user.message_id, "queued-user"); + assert_eq!(user.text, "queued follow-up"); + protocol_events.push("user_message_appended"); + } + Some(pb::interaction_update::Message::TextDelta(update)) + if update.text.contains("continued after user interruption") => + { + protocol_events.push("continued_output"); + } + _ => {} + } + } + acknowledge_kv(&handle, &mut append_seqno, &frame).await; + } + + assert_eq!( + protocol_events, + [ + "queued", + "delivered", + "user_message_appended", + "continued_output" + ] + ); +} + #[tokio::test] async fn tool_call_with_empty_arguments_does_not_fail_the_run() { // A tool call that carries no arguments streams no argument text. Parsing it