From 6ac666da4f118e8f104d78e3b17cdf3d4dff695d Mon Sep 17 00:00:00 2001 From: kevin9327 Date: Sun, 30 Aug 2026 19:23:57 +0900 Subject: [PATCH] fix(conversation): clear pending runtime user-message injections Runtime user messages and context injections both queue into `pending_injections`, but with different keys: injections use the raw injection id (committed under `inject-context:{id}`) while user messages use the full `user-message:{id}` event id. The commit-correlation handler only stripped the `inject-context:` prefix, so a user message's entry was never removed. Consequences: - the client never received `ContextInjectionDelivered` / `UserMessageAppended` for the message; and - `pending_injections` stayed non-empty, so every later `ExecuteToolRound` was detached without dispatching its tools and `tool_round::execute` blocked forever -- a hung turn whenever the model made a tool call after the interruption. Derive the lookup key by stripping the injection prefix when present and otherwise using the event id verbatim, so both kinds are cleared and their delivered/appended events fire. Co-Authored-By: Claude Opus 4.8 --- server/src/cursor/conversation/output.rs | 37 +++++--- server/tests/interrupt.rs | 116 +++++++++++++++++++++++ 2 files changed, 138 insertions(+), 15 deletions(-) diff --git a/server/src/cursor/conversation/output.rs b/server/src/cursor/conversation/output.rs index 54c4126..5980b22 100644 --- a/server/src/cursor/conversation/output.rs +++ b/server/src/cursor/conversation/output.rs @@ -424,21 +424,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/tests/interrupt.rs b/server/tests/interrupt.rs index d480a1f..4ddeeef 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 injected_user_context_restarts_only_the_active_model_cycle() { let (_directory, store) = fixtures::temp_store().await;