From 80604bdf97c1ce1bbfbfeffd6d7381c952512d71 Mon Sep 17 00:00:00 2001 From: joel01-dev Date: Fri, 28 Aug 2026 00:47:53 -0300 Subject: [PATCH 1/4] fix: prevent silent SSE stream hangs, add Bash alias, debug logging - Actor: use lifecycle::cancel() instead of handle.cancel() on channel close and Abort, ensuring end-stream frame is always emitted - Lifecycle: use match instead of ? on encode_error_end_stream with fallback end-stream frame to prevent silent hangs - Dispatch: add Bash tool alias for Shell - Context/Blob sync: increase timeouts from 15s to 60s - Providers: add detailed debug logging to openai_chat and router --- server/src/cursor/actor.rs | 4 +- server/src/cursor/blob_sync.rs | 4 +- server/src/cursor/context_sync.rs | 2 +- server/src/cursor/lifecycle.rs | 14 +- server/src/cursor/sessions.rs | 2 + server/src/cursor/tools/dispatch/mod.rs | 114 +--------------- server/src/provider/openai_chat.rs | 172 ++++++++++++++---------- server/src/provider/router.rs | 73 ++++++++++ 8 files changed, 193 insertions(+), 192 deletions(-) diff --git a/server/src/cursor/actor.rs b/server/src/cursor/actor.rs index 3df974d..cb03b5c 100644 --- a/server/src/cursor/actor.rs +++ b/server/src/cursor/actor.rs @@ -58,14 +58,14 @@ impl CursorActor { let command = match receiver.recv().await { Some(command) => command, None => { - handle.cancel(); + lifecycle::cancel(&handle).ok(); break; } }; match command { CursorCommand::Abort => { handle.mark_conversation_cancelled(); - handle.cancel(); + lifecycle::cancel(&handle).ok(); } CursorCommand::Finished => { break; diff --git a/server/src/cursor/blob_sync.rs b/server/src/cursor/blob_sync.rs index be4c9dc..ce8acd8 100644 --- a/server/src/cursor/blob_sync.rs +++ b/server/src/cursor/blob_sync.rs @@ -129,7 +129,7 @@ impl BlobSynchronizer { let result = tokio::select! { result = receiver => result.map_err(|_| Error::Protocol("KV SET response channel closed".into()))?, _ = cancellation.cancelled() => Err(Error::Cancelled), - _ = tokio::time::sleep(Duration::from_secs(15)) => Err(Error::Protocol(format!("KV SET timed out: {}", blob_id.to_base64()))), + _ = tokio::time::sleep(Duration::from_secs(60)) => Err(Error::Protocol(format!("KV SET timed out: {}", blob_id.to_base64()))), }; if result.is_err() { self.inner.set_requests.lock().await.remove(&id); @@ -182,7 +182,7 @@ impl BlobSynchronizer { let result = tokio::select! { result = receiver => result.map_err(|_| Error::Protocol("KV GET response channel closed".into()))?, _ = cancellation.cancelled() => Err(Error::Cancelled), - _ = tokio::time::sleep(Duration::from_secs(15)) => Err(Error::Protocol(format!("KV GET timed out: {}", blob_id.to_base64()))), + _ = tokio::time::sleep(Duration::from_secs(60)) => Err(Error::Protocol(format!("KV GET timed out: {}", blob_id.to_base64()))), }; if result.is_err() { self.inner.get_requests.lock().await.remove(&id); diff --git a/server/src/cursor/context_sync.rs b/server/src/cursor/context_sync.rs index 88caa00..b373e9d 100644 --- a/server/src/cursor/context_sync.rs +++ b/server/src/cursor/context_sync.rs @@ -84,7 +84,7 @@ impl RequestContextSynchronizer { let result = tokio::select! { result = receiver => result.map_err(|_| Error::Protocol("request context response channel closed".into()))?, _ = cancellation.cancelled() => Err(Error::Cancelled), - _ = tokio::time::sleep(Duration::from_secs(15)) => Err(Error::Protocol("request context timed out".into())), + _ = tokio::time::sleep(Duration::from_secs(60)) => Err(Error::Protocol("request context timed out".into())), }; if result.is_err() { self.pending.lock().await.take(); diff --git a/server/src/cursor/lifecycle.rs b/server/src/cursor/lifecycle.rs index 6195a41..a9d65b8 100644 --- a/server/src/cursor/lifecycle.rs +++ b/server/src/cursor/lifecycle.rs @@ -32,17 +32,25 @@ pub fn fail(handle: &CursorSessionHandle, error: &Error) -> Result<()> { | Error::Encode(_) | Error::Io(_) => plain_error(ConnectCode::Internal, error), }; - handle.emit_frame(encode_error_end_stream(&stream_error)?); + // Always close the output even if encoding fails, to prevent silent hangs. + match encode_error_end_stream(&stream_error) { + Ok(frame) => handle.emit_frame(frame), + Err(_) => handle.emit_frame(encode_end_stream()), + } handle.close_output(); Ok(()) } pub fn cancel(handle: &CursorSessionHandle) -> Result<()> { - handle.emit_frame(encode_error_end_stream(&ConnectStreamError { + // Always close the output even if encoding fails, to prevent silent hangs. + match encode_error_end_stream(&ConnectStreamError { code: ConnectCode::Canceled, message: "run was cancelled".into(), details: Vec::new(), - })?); + }) { + Ok(frame) => handle.emit_frame(frame), + Err(_) => handle.emit_frame(encode_end_stream()), + } handle.close_output(); Ok(()) } diff --git a/server/src/cursor/sessions.rs b/server/src/cursor/sessions.rs index 5e3b456..6b4b8a8 100644 --- a/server/src/cursor/sessions.rs +++ b/server/src/cursor/sessions.rs @@ -313,6 +313,8 @@ impl CursorSessionRegistry { pub(crate) async fn wait_route(&self, request_id: &str) -> CursorRoute { loop { + // Create the notification future BEFORE checking state to avoid + // a race where a notification fires between state check and await. let changed = self.inner.route_changed.notified(); if self.inner.runs.lock().await.contains_key(request_id) { return CursorRoute::Local; diff --git a/server/src/cursor/tools/dispatch/mod.rs b/server/src/cursor/tools/dispatch/mod.rs index 39957ba..fc4266d 100644 --- a/server/src/cursor/tools/dispatch/mod.rs +++ b/server/src/cursor/tools/dispatch/mod.rs @@ -1,3 +1,4 @@ +mod await_shell; mod edit; mod exec; mod interaction; @@ -50,70 +51,19 @@ pub(super) async fn start( return local::subagents_disabled(call); } - let normalized_call = normalize_block_until_ms(call)?; - let call = normalized_call.as_ref().unwrap_or(call); - match normalized(&call.name).as_str() { - "shell" | "read" | "delete" | "grep" | "glob" | "readlints" | "task" | "callmcptool" + "shell" | "bash" | "read" | "delete" | "grep" | "glob" | "readlints" | "task" | "callmcptool" | "fetchmcpresource" | "getmcptools" => exec::start(runtime, call, context).await, "write" | "strreplace" | "editnotebook" => edit::start(runtime, call, context).await, "askquestion" | "websearch" | "webfetch" | "switchmode" | "createplan" | "generateimage" => interaction::start(runtime, call).await, "todowrite" | "updatecurrentstep" => local::start(call, message_index), + "awaitshell" => await_shell::start(runtime, results, call, context).await, "semblesearch" | "semblefindrelated" => semble::start(results, call, store.cloned()), _ => Err(Error::Protocol(format!("unsupported tool: {}", call.name))), } } -fn normalize_block_until_ms(call: &ToolCall) -> Result> { - if normalized(&call.name) != "shell" { - return Ok(None); - } - let Some(value) = call.arguments.get("block_until_ms") else { - return Ok(None); - }; - - let integer = if let Some(value) = value.as_i64() { - value - } else { - let value = value.as_f64().ok_or_else(|| { - Error::Protocol(format!("{} block_until_ms must be an integer", call.name)) - })?; - if !value.is_finite() || value.fract() != 0.0 { - return Err(Error::Protocol(format!( - "{} block_until_ms must be an integer", - call.name - ))); - } - if value < i64::MIN as f64 || value > i64::MAX as f64 { - return Err(Error::Protocol(format!( - "{} block_until_ms is out of range", - call.name - ))); - } - value as i64 - }; - - if integer < 0 { - return Err(Error::Protocol(format!( - "{} block_until_ms is out of range", - call.name - ))); - } - - if value.as_i64().is_some() { - return Ok(None); - } - - let mut normalized_call = call.clone(); - normalized_call - .arguments - .as_object_mut() - .ok_or_else(|| Error::Protocol(format!("{} arguments must be a JSON object", call.name)))? - .insert("block_until_ms".into(), serde_json::Value::from(integer)); - Ok(Some(normalized_call)) -} - fn is_mcp_auth(call: &ToolCall) -> bool { normalized(&call.name) == "callmcptool" && call @@ -139,61 +89,3 @@ pub(super) fn normalized(name: &str) -> String { .flat_map(char::to_lowercase) .collect() } - -#[cfg(test)] -mod tests { - use super::*; - - fn tool(name: &str, arguments: serde_json::Value) -> ToolCall { - ToolCall { - index: 0, - call_id: "call-1".into(), - model_call_id: "model-call-1".into(), - name: name.into(), - arguments_text: arguments.to_string(), - arguments, - } - } - - #[test] - fn shell_accepts_integer_valued_float_timeout() { - let call = tool( - "Shell", - serde_json::json!({"command": "echo ok", "block_until_ms": 45_000.0}), - ); - - let call = normalize_block_until_ms(&call).unwrap().unwrap(); - - assert_eq!(call.arguments["block_until_ms"].as_i64(), Some(45_000)); - } - - #[test] - fn shell_rejects_fractional_timeout() { - let call = tool( - "Shell", - serde_json::json!({"command": "echo ok", "block_until_ms": 30_000.5}), - ); - - let error = normalize_block_until_ms(&call).unwrap_err(); - - assert_eq!( - error.to_string(), - "protocol error: Shell block_until_ms must be an integer" - ); - } - - #[test] - fn shell_rejects_negative_timeout_instead_of_defaulting() { - let call = tool( - "Shell", - serde_json::json!({"command": "echo ok", "block_until_ms": -1}), - ); - - let error = normalize_block_until_ms(&call).unwrap_err(); - - assert_eq!( - error.to_string(), - "protocol error: Shell block_until_ms is out of range" - ); - } -} diff --git a/server/src/provider/openai_chat.rs b/server/src/provider/openai_chat.rs index 75827e7..acf9614 100644 --- a/server/src/provider/openai_chat.rs +++ b/server/src/provider/openai_chat.rs @@ -15,6 +15,29 @@ use crate::{ Error, Result, }; +macro_rules! dbg_log { + ($loc:expr, $msg:expr, $data:expr) => { + { + let _ = (|| -> std::result::Result<(), Box> { + use std::io::Write; + let payload = serde_json::json!({ + "sessionId": "216d24", + "location": $loc, + "message": $msg, + "data": $data, + "timestamp": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_millis() as u64, + }); + let mut f = std::fs::OpenOptions::new().create(true).append(true).open( + concat!(env!("CARGO_MANIFEST_DIR"), "/../.cursor/debug-216d24.log") + )?; + writeln!(f, "{}", payload)?; + f.flush()?; + Ok(()) + })(); + } + }; +} + use super::{ apply_openai_prompt_cache_key, merge_extra_params, recorder::recorded_headers, CallRecorder, FinishReason, ModelEvent, Provider, ProviderStream, @@ -61,18 +84,23 @@ impl Provider for OpenAiChatProvider { let recorder = self.recorder.clone(); Box::pin(try_stream! { let ModelInvocation { call_id, request, .. } = invocation; + dbg_log!("openai_chat.rs:stream", "Provider stream started", serde_json::json!({ + "model": request.model.model_id, + "url": config.request_url, + "call_id": call_id.clone(), + "history_len": request.history.len(), + "tools_count": request.prompt.tools.len() + })); let messages = openai_chat_messages(&request.prompt.instructions, &request.history)?; let mut body = json!({ "model": request.model.model_id, "messages": messages, + "tools": request.prompt.tools.iter().map(|tool| json!({"type":"function","function":{ + "name": tool.name, "description": tool.description, "parameters": tool.parameters + }})).collect::>(), "stream": true, "stream_options": {"include_usage": true} }); - if !request.prompt.tools.is_empty() { - body["tools"] = json!(request.prompt.tools.iter().map(|tool| json!({"type":"function","function":{ - "name": tool.name, "description": tool.description, "parameters": tool.parameters - }})).collect::>()); - } apply_model(&mut body, &request.model, config.max_output_tokens)?; merge_extra_params(&mut body, &request.model.extra_params)?; apply_openai_prompt_cache_key(&mut body, &request.model.model_id)?; @@ -85,7 +113,22 @@ impl Provider for OpenAiChatProvider { _ = cancellation.cancelled() => return, response = request => response, }; - let response = response?; + let response = match response { + Ok(r) => { + dbg_log!("openai_chat.rs:stream", "HTTP response received", serde_json::json!({ + "status": r.status().as_u16(), + "url": config.request_url + })); + r + } + Err(e) => { + dbg_log!("openai_chat.rs:stream", "HTTP request FAILED", serde_json::json!({ + "error": e.to_string(), + "url": config.request_url + })); + Err(Error::from(e))? + } + }; if let Some(recorder) = &recorder { recorder.response_headers(response.status().as_u16()).await?; } @@ -118,15 +161,50 @@ impl Provider for OpenAiChatProvider { let mut final_usage = None; let mut finish = None; let mut saw_done_marker = false; + let mut loop_iteration: u64 = 0; loop { + loop_iteration += 1; + dbg_log!("openai_chat.rs:stream:poll", "Polling SSE event", serde_json::json!({ + "iteration": loop_iteration, + "saw_done": saw_done_marker, + "has_finish": finish.is_some(), + "tool_count": tools.len(), + "text_open": text_open, + "thinking_open": thinking_open, + "reasoning_len": reasoning.len() + })); let event = tokio::select! { _ = cancellation.cancelled() => { + dbg_log!("openai_chat.rs:stream", "Stream cancelled by token", serde_json::json!({ + "iteration": loop_iteration, + "saw_done_marker": saw_done_marker, + "tool_count": tools.len(), + "text_open": text_open + })); return; } event = source.next() => event, }; - let Some(event) = event else { break }; - let event = event.map_err(|error| Error::Provider(format!("OpenAI Chat SSE: {error}")))?; + let Some(event) = event else { + dbg_log!("openai_chat.rs:stream:poll", "SSE stream ended (None from source)", serde_json::json!({ + "iteration": loop_iteration, + "saw_done_marker": saw_done_marker, + "has_finish": finish.is_some(), + "tool_count": tools.len(), + "text_open": text_open, + "thinking_open": thinking_open, + "reasoning_len": reasoning.len() + })); + break; + }; + let event = event.map_err(|error| { + let err_msg = error.to_string(); + dbg_log!("openai_chat.rs:stream:poll", "SSE event error", serde_json::json!({ + "iteration": loop_iteration, + "error": err_msg.clone() + })); + Error::Provider(format!("OpenAI Chat SSE: {err_msg}")) + })?; if event.data == "[DONE]" { saw_done_marker = true; break; } let value: Value = serde_json::from_str(&event.data)?; if let Some(usage) = value.get("usage").filter(|value| !value.is_null()) { @@ -134,7 +212,7 @@ impl Provider for OpenAiChatProvider { } let Some(choice) = value.get("choices").and_then(Value::as_array).and_then(|values| values.first()) else { continue; }; let delta = choice.get("delta").unwrap_or(&Value::Null); - if let Some(reasoning_delta) = delta.get("reasoning_content").or_else(|| delta.get("reasoning")).and_then(Value::as_str).filter(|text| !text.is_empty()) { + if let Some(reasoning_delta) = delta.get("reasoning_content").and_then(Value::as_str).filter(|text| !text.is_empty()) { if !thinking_open { thinking_open = true; yield ModelEvent::ThinkingStart; } reasoning.push_str(reasoning_delta); yield ModelEvent::ThinkingDelta(reasoning_delta.into()); @@ -158,6 +236,15 @@ impl Provider for OpenAiChatProvider { finish = Some(map_finish(reason, !tools.is_empty())); } } + dbg_log!("openai_chat.rs:stream", "SSE loop exited", serde_json::json!({ + "total_iterations": loop_iteration, + "saw_done_marker": saw_done_marker, + "finish_reason": finish.as_ref().map(|f| format!("{:?}", f)), + "reasoning_len": reasoning.len(), + "tool_count": tools.len(), + "text_open": text_open, + "thinking_open": thinking_open + })); if thinking_open { yield ModelEvent::ThinkingEnd; } if text_open { yield ModelEvent::TextEnd; } for (index, tool) in &mut tools { @@ -232,27 +319,12 @@ fn openai_chat_messages(instructions: &str, messages: &[ProjectedMessage]) -> Re calls, .. } => { + value.insert("content".into(), Value::String(text.clone())); let replay_reasoning = replay_state .as_ref() .filter(|state| state.provider_kind == "openai_chat") .and_then(|state| state.value.get("reasoning_content")) - .and_then(Value::as_str) - .filter(|reasoning| !reasoning.is_empty()); - - // Chat Completions rejects an empty assistant content string. Tool-call - // assistant messages use null content, while an assistant with no visible - // content at all does not need to be sent. - if text.is_empty() && calls.is_empty() && replay_reasoning.is_none() { - continue; - } - value.insert( - "content".into(), - if text.is_empty() { - Value::Null - } else { - Value::String(text.clone()) - }, - ); + .and_then(Value::as_str); if let Some(reasoning) = replay_reasoning { value.insert("reasoning_content".into(), Value::String(reasoning.into())); } @@ -415,7 +487,7 @@ mod tests { model::{ContentPart, ProjectedContent, ProjectedMessage, ToolResultContent}, model::{ProviderReplayState, Role, ToolCallContent}, }; - use serde_json::{json, Value}; + use serde_json::json; #[test] fn chat_replay_state_is_encoded_as_reasoning_content() { @@ -447,52 +519,6 @@ mod tests { assert_eq!(messages[0]["tool_calls"][0]["id"], "call-1"); } - #[test] - fn chat_tool_call_assistant_uses_null_content() { - let messages = openai_chat_messages( - "", - &[ProjectedMessage { - message_id: "test".into(), - role: Role::Assistant, - content: ProjectedContent::Assistant { - text: String::new(), - thinking: String::new(), - replay_state: None, - calls: vec![ToolCallContent { - index: 0, - call_id: "call-1".into(), - name: "Read".into(), - arguments: json!({"path": "README.md"}), - }], - }, - }], - ) - .unwrap(); - - assert_eq!(messages[0]["content"], Value::Null); - assert!(messages[0]["tool_calls"].is_array()); - } - - #[test] - fn chat_contentless_assistant_is_omitted() { - let messages = openai_chat_messages( - "", - &[ProjectedMessage { - message_id: "test".into(), - role: Role::Assistant, - content: ProjectedContent::Assistant { - text: String::new(), - thinking: String::new(), - replay_state: None, - calls: vec![], - }, - }], - ) - .unwrap(); - - assert!(messages.is_empty()); - } - #[test] fn another_provider_replay_does_not_invent_chat_reasoning_content() { let messages = openai_chat_messages( diff --git a/server/src/provider/router.rs b/server/src/provider/router.rs index eeec35f..5a29b0d 100644 --- a/server/src/provider/router.rs +++ b/server/src/provider/router.rs @@ -11,6 +11,29 @@ use crate::{ Error, Result, }; +macro_rules! dbg_log { + ($loc:expr, $msg:expr, $data:expr) => { + { + let _ = (|| -> std::result::Result<(), Box> { + use std::io::Write; + let payload = serde_json::json!({ + "sessionId": "216d24", + "location": $loc, + "message": $msg, + "data": $data, + "timestamp": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_millis() as u64, + }); + let mut f = std::fs::OpenOptions::new().create(true).append(true).open( + concat!(env!("CARGO_MANIFEST_DIR"), "/../.cursor/debug-216d24.log") + )?; + writeln!(f, "{}", payload)?; + f.flush()?; + Ok(()) + })(); + } + }; +} + use super::{ normalize::NormalizedProvider, AnthropicProvider, CallRecorder, OpenAiChatProvider, OpenAiResponsesProvider, Provider, ProviderStream, @@ -90,23 +113,73 @@ impl Provider for ProviderRouter { let provider = build_observed(&config, recorder.clone(), client)?; let stream_cancellation = cancellation.clone(); let mut stream = provider.stream(invocation, cancellation); + let stream_started = std::time::Instant::now(); + dbg_log!("router.rs:stream", "provider stream created", serde_json::json!({ + "model": selected, + "provider_type": format!("{:?}", provider_type), + "url": config.request_url, + "timeout_ms": config.request_timeout.as_millis() as u64, + })); + let mut last_event_time = std::time::Instant::now(); + let mut event_count: u64 = 0; while let Some(event) = stream.next().await { + let now = std::time::Instant::now(); + let gap_ms = now.duration_since(last_event_time).as_millis() as u64; + let elapsed_ms = now.duration_since(stream_started).as_millis() as u64; + event_count += 1; match event { Ok(event) => { + let event_name = match &event { + super::ModelEvent::Start { .. } => "Start", + super::ModelEvent::TextStart => "TextStart", + super::ModelEvent::TextDelta(d) => { dbg_log!("router.rs:stream", "TextDelta", serde_json::json!({"gap_ms": gap_ms, "elapsed_ms": elapsed_ms, "delta_len": d.len(), "event_count": event_count})); "TextDelta" }, + super::ModelEvent::TextEnd => "TextEnd", + super::ModelEvent::ThinkingStart => "ThinkingStart", + super::ModelEvent::ThinkingDelta(d) => { dbg_log!("router.rs:stream", "ThinkingDelta", serde_json::json!({"gap_ms": gap_ms, "elapsed_ms": elapsed_ms, "delta_len": d.len(), "event_count": event_count})); "ThinkingDelta" }, + super::ModelEvent::ThinkingEnd => "ThinkingEnd", + super::ModelEvent::ToolCallStart { name, .. } => { dbg_log!("router.rs:stream", "ToolCallStart", serde_json::json!({"name": name, "elapsed_ms": elapsed_ms})); "ToolCallStart" }, + super::ModelEvent::ToolCallArgumentsDelta { .. } => "ToolCallArgsDelta", + super::ModelEvent::ToolCallEnd { .. } => "ToolCallEnd", + super::ModelEvent::ProviderReplayState(_) => "ReplayState", + super::ModelEvent::Usage(_) => "Usage", + super::ModelEvent::Done(reason) => { dbg_log!("router.rs:stream", "Done", serde_json::json!({"reason": format!("{:?}", reason), "elapsed_ms": elapsed_ms, "event_count": event_count})); "Done" }, + }; + if gap_ms > 5000 { + dbg_log!("router.rs:stream", "SLOW GAP detected between events", serde_json::json!({ + "gap_ms": gap_ms, + "elapsed_ms": elapsed_ms, + "event": event_name, + "event_count": event_count, + })); + } recorder.event(&event).await?; + last_event_time = now; yield event; } Err(error) => { + dbg_log!("router.rs:stream", "PROVIDER ERROR", serde_json::json!({ + "error": error.to_string(), + "elapsed_ms": elapsed_ms, + "gap_ms": gap_ms, + "event_count": event_count, + })); recorder.failed(&error).await?; Err(error)?; } } } if !recorder.is_finished() { + let elapsed_ms = stream_started.elapsed().as_millis() as u64; if stream_cancellation.is_cancelled() { + dbg_log!("router.rs:stream", "stream ended: cancelled", serde_json::json!({"elapsed_ms": elapsed_ms, "event_count": event_count})); recorder.cancelled().await?; } else { let error = Error::Provider("provider stream ended without Done".into()); + dbg_log!("router.rs:stream", "stream ended: WITHOUT DONE (fatal)", serde_json::json!({ + "elapsed_ms": elapsed_ms, + "event_count": event_count, + "error": "provider stream ended without Done", + })); recorder.failed(&error).await?; Err(error)?; } From 2369d101ec5bd956285503d1401e76d633cfacb0 Mon Sep 17 00:00:00 2001 From: joel01-dev Date: Fri, 28 Aug 2026 00:48:43 -0300 Subject: [PATCH 2/4] docs: add PR description --- pr-body.md | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 pr-body.md diff --git a/pr-body.md b/pr-body.md new file mode 100644 index 0000000..8c85d25 --- /dev/null +++ b/pr-body.md @@ -0,0 +1,35 @@ +## Problem + +During normal Agent usage, the flow silently stops mid-conversation. The AI says "now I'll read the files" and then never continues. No tool call, no more text, no error. It just freezes. + +## Root Causes + +### 1. Actor not emitting end-stream on channel close (actor.rs) +When the command channel closes or receives Abort, the actor only called handle.cancel() which sets the cancellation token but never emits an end-stream frame. The SSE stream hangs indefinitely. + +### 2. lifecycle encoding failure skips close_output() (lifecycle.rs) +Both fail() and cancel() used the ? operator on encode_error_end_stream(), meaning if serialization fails, close_output() is never called. The stream stays open silently. + +### 3. Context/blob sync timeouts too aggressive (context_sync.rs, blob_sync.rs) +Hardcoded 15-second timeouts kill sessions silently on slow networks or with proxy configurations. + +### 4. Bash tool not recognized (dispatch/mod.rs) +Cursor IDE sends "Bash" tool calls but the server only recognized "Shell", causing "unsupported tool: Bash" errors. + +## Changes + +- actor.rs: Use lifecycle::cancel() instead of handle.cancel() +- lifecycle.rs: Use match instead of ? with fallback end-stream frame +- sessions.rs: Document correct Notify pattern +- context_sync.rs: 15s to 60s timeout +- blob_sync.rs: 15s to 60s timeout (SET and GET) +- dispatch/mod.rs: Add Bash tool alias for Shell +- openai_chat.rs: Add detailed debug logging +- router.rs: Add detailed debug logging + +## Testing + +- All existing tests pass +- cargo check compiles cleanly +- Both debug and release builds succeed +- Tested with Cursor BYOK desktop app \ No newline at end of file From 318545cebdfabc6c34c857cb2e0e9e7cdf4d7315 Mon Sep 17 00:00:00 2001 From: joel01-dev Date: Fri, 28 Aug 2026 01:07:41 -0300 Subject: [PATCH 3/4] fix: restore upstream dispatch/mod.rs, add only Bash alias --- server/src/cursor/tools/dispatch/mod.rs | 112 +++++++++++++++++++++++- 1 file changed, 110 insertions(+), 2 deletions(-) diff --git a/server/src/cursor/tools/dispatch/mod.rs b/server/src/cursor/tools/dispatch/mod.rs index fc4266d..2e6b3ce 100644 --- a/server/src/cursor/tools/dispatch/mod.rs +++ b/server/src/cursor/tools/dispatch/mod.rs @@ -1,4 +1,3 @@ -mod await_shell; mod edit; mod exec; mod interaction; @@ -51,6 +50,9 @@ pub(super) async fn start( return local::subagents_disabled(call); } + let normalized_call = normalize_block_until_ms(call)?; + let call = normalized_call.as_ref().unwrap_or(call); + match normalized(&call.name).as_str() { "shell" | "bash" | "read" | "delete" | "grep" | "glob" | "readlints" | "task" | "callmcptool" | "fetchmcpresource" | "getmcptools" => exec::start(runtime, call, context).await, @@ -58,12 +60,60 @@ pub(super) async fn start( "askquestion" | "websearch" | "webfetch" | "switchmode" | "createplan" | "generateimage" => interaction::start(runtime, call).await, "todowrite" | "updatecurrentstep" => local::start(call, message_index), - "awaitshell" => await_shell::start(runtime, results, call, context).await, "semblesearch" | "semblefindrelated" => semble::start(results, call, store.cloned()), _ => Err(Error::Protocol(format!("unsupported tool: {}", call.name))), } } +fn normalize_block_until_ms(call: &ToolCall) -> Result> { + if normalized(&call.name) != "shell" { + return Ok(None); + } + let Some(value) = call.arguments.get("block_until_ms") else { + return Ok(None); + }; + + let integer = if let Some(value) = value.as_i64() { + value + } else { + let value = value.as_f64().ok_or_else(|| { + Error::Protocol(format!("{} block_until_ms must be an integer", call.name)) + })?; + if !value.is_finite() || value.fract() != 0.0 { + return Err(Error::Protocol(format!( + "{} block_until_ms must be an integer", + call.name + ))); + } + if value < i64::MIN as f64 || value > i64::MAX as f64 { + return Err(Error::Protocol(format!( + "{} block_until_ms is out of range", + call.name + ))); + } + value as i64 + }; + + if integer < 0 { + return Err(Error::Protocol(format!( + "{} block_until_ms is out of range", + call.name + ))); + } + + if value.as_i64().is_some() { + return Ok(None); + } + + let mut normalized_call = call.clone(); + normalized_call + .arguments + .as_object_mut() + .ok_or_else(|| Error::Protocol(format!("{} arguments must be a JSON object", call.name)))? + .insert("block_until_ms".into(), serde_json::Value::from(integer)); + Ok(Some(normalized_call)) +} + fn is_mcp_auth(call: &ToolCall) -> bool { normalized(&call.name) == "callmcptool" && call @@ -89,3 +139,61 @@ pub(super) fn normalized(name: &str) -> String { .flat_map(char::to_lowercase) .collect() } + +#[cfg(test)] +mod tests { + use super::*; + + fn tool(name: &str, arguments: serde_json::Value) -> ToolCall { + ToolCall { + index: 0, + call_id: "call-1".into(), + model_call_id: "model-call-1".into(), + name: name.into(), + arguments_text: arguments.to_string(), + arguments, + } + } + + #[test] + fn shell_accepts_integer_valued_float_timeout() { + let call = tool( + "Shell", + serde_json::json!({"command": "echo ok", "block_until_ms": 45_000.0}), + ); + + let call = normalize_block_until_ms(&call).unwrap().unwrap(); + + assert_eq!(call.arguments["block_until_ms"].as_i64(), Some(45_000)); + } + + #[test] + fn shell_rejects_fractional_timeout() { + let call = tool( + "Shell", + serde_json::json!({"command": "echo ok", "block_until_ms": 30_000.5}), + ); + + let error = normalize_block_until_ms(&call).unwrap_err(); + + assert_eq!( + error.to_string(), + "protocol error: Shell block_until_ms must be an integer" + ); + } + + #[test] + fn shell_rejects_negative_timeout_instead_of_defaulting() { + let call = tool( + "Shell", + serde_json::json!({"command": "echo ok", "block_until_ms": -1}), + ); + + let error = normalize_block_until_ms(&call).unwrap_err(); + + assert_eq!( + error.to_string(), + "protocol error: Shell block_until_ms is out of range" + ); + } +} From b2610c562e792f93ff6773f4636fb6612bc77025 Mon Sep 17 00:00:00 2001 From: joel01-dev Date: Fri, 28 Aug 2026 01:12:31 -0300 Subject: [PATCH 4/4] fix: restore upstream openai_chat.rs, add only debug logging --- server/src/provider/openai_chat.rs | 109 ++++++++++++++++++++--------- 1 file changed, 74 insertions(+), 35 deletions(-) diff --git a/server/src/provider/openai_chat.rs b/server/src/provider/openai_chat.rs index acf9614..996fa8e 100644 --- a/server/src/provider/openai_chat.rs +++ b/server/src/provider/openai_chat.rs @@ -95,12 +95,14 @@ impl Provider for OpenAiChatProvider { let mut body = json!({ "model": request.model.model_id, "messages": messages, - "tools": request.prompt.tools.iter().map(|tool| json!({"type":"function","function":{ - "name": tool.name, "description": tool.description, "parameters": tool.parameters - }})).collect::>(), "stream": true, "stream_options": {"include_usage": true} }); + if !request.prompt.tools.is_empty() { + body["tools"] = json!(request.prompt.tools.iter().map(|tool| json!({"type":"function","function":{ + "name": tool.name, "description": tool.description, "parameters": tool.parameters + }})).collect::>()); + } apply_model(&mut body, &request.model, config.max_output_tokens)?; merge_extra_params(&mut body, &request.model.extra_params)?; apply_openai_prompt_cache_key(&mut body, &request.model.model_id)?; @@ -164,42 +166,27 @@ impl Provider for OpenAiChatProvider { let mut loop_iteration: u64 = 0; loop { loop_iteration += 1; - dbg_log!("openai_chat.rs:stream:poll", "Polling SSE event", serde_json::json!({ - "iteration": loop_iteration, - "saw_done": saw_done_marker, - "has_finish": finish.is_some(), - "tool_count": tools.len(), - "text_open": text_open, - "thinking_open": thinking_open, - "reasoning_len": reasoning.len() - })); let event = tokio::select! { _ = cancellation.cancelled() => { dbg_log!("openai_chat.rs:stream", "Stream cancelled by token", serde_json::json!({ "iteration": loop_iteration, "saw_done_marker": saw_done_marker, - "tool_count": tools.len(), - "text_open": text_open + "tool_count": tools.len() })); return; } event = source.next() => event, }; let Some(event) = event else { - dbg_log!("openai_chat.rs:stream:poll", "SSE stream ended (None from source)", serde_json::json!({ + dbg_log!("openai_chat.rs:stream", "SSE stream ended (None)", serde_json::json!({ "iteration": loop_iteration, - "saw_done_marker": saw_done_marker, - "has_finish": finish.is_some(), - "tool_count": tools.len(), - "text_open": text_open, - "thinking_open": thinking_open, - "reasoning_len": reasoning.len() + "saw_done_marker": saw_done_marker })); break; }; let event = event.map_err(|error| { let err_msg = error.to_string(); - dbg_log!("openai_chat.rs:stream:poll", "SSE event error", serde_json::json!({ + dbg_log!("openai_chat.rs:stream", "SSE event error", serde_json::json!({ "iteration": loop_iteration, "error": err_msg.clone() })); @@ -212,7 +199,7 @@ impl Provider for OpenAiChatProvider { } let Some(choice) = value.get("choices").and_then(Value::as_array).and_then(|values| values.first()) else { continue; }; let delta = choice.get("delta").unwrap_or(&Value::Null); - if let Some(reasoning_delta) = delta.get("reasoning_content").and_then(Value::as_str).filter(|text| !text.is_empty()) { + if let Some(reasoning_delta) = delta.get("reasoning_content").or_else(|| delta.get("reasoning")).and_then(Value::as_str).filter(|text| !text.is_empty()) { if !thinking_open { thinking_open = true; yield ModelEvent::ThinkingStart; } reasoning.push_str(reasoning_delta); yield ModelEvent::ThinkingDelta(reasoning_delta.into()); @@ -236,15 +223,6 @@ impl Provider for OpenAiChatProvider { finish = Some(map_finish(reason, !tools.is_empty())); } } - dbg_log!("openai_chat.rs:stream", "SSE loop exited", serde_json::json!({ - "total_iterations": loop_iteration, - "saw_done_marker": saw_done_marker, - "finish_reason": finish.as_ref().map(|f| format!("{:?}", f)), - "reasoning_len": reasoning.len(), - "tool_count": tools.len(), - "text_open": text_open, - "thinking_open": thinking_open - })); if thinking_open { yield ModelEvent::ThinkingEnd; } if text_open { yield ModelEvent::TextEnd; } for (index, tool) in &mut tools { @@ -319,12 +297,27 @@ fn openai_chat_messages(instructions: &str, messages: &[ProjectedMessage]) -> Re calls, .. } => { - value.insert("content".into(), Value::String(text.clone())); let replay_reasoning = replay_state .as_ref() .filter(|state| state.provider_kind == "openai_chat") .and_then(|state| state.value.get("reasoning_content")) - .and_then(Value::as_str); + .and_then(Value::as_str) + .filter(|reasoning| !reasoning.is_empty()); + + // Chat Completions rejects an empty assistant content string. Tool-call + // assistant messages use null content, while an assistant with no visible + // content at all does not need to be sent. + if text.is_empty() && calls.is_empty() && replay_reasoning.is_none() { + continue; + } + value.insert( + "content".into(), + if text.is_empty() { + Value::Null + } else { + Value::String(text.clone()) + }, + ); if let Some(reasoning) = replay_reasoning { value.insert("reasoning_content".into(), Value::String(reasoning.into())); } @@ -487,7 +480,7 @@ mod tests { model::{ContentPart, ProjectedContent, ProjectedMessage, ToolResultContent}, model::{ProviderReplayState, Role, ToolCallContent}, }; - use serde_json::json; + use serde_json::{json, Value}; #[test] fn chat_replay_state_is_encoded_as_reasoning_content() { @@ -519,6 +512,52 @@ mod tests { assert_eq!(messages[0]["tool_calls"][0]["id"], "call-1"); } + #[test] + fn chat_tool_call_assistant_uses_null_content() { + let messages = openai_chat_messages( + "", + &[ProjectedMessage { + message_id: "test".into(), + role: Role::Assistant, + content: ProjectedContent::Assistant { + text: String::new(), + thinking: String::new(), + replay_state: None, + calls: vec![ToolCallContent { + index: 0, + call_id: "call-1".into(), + name: "Read".into(), + arguments: json!({"path": "README.md"}), + }], + }, + }], + ) + .unwrap(); + + assert_eq!(messages[0]["content"], Value::Null); + assert!(messages[0]["tool_calls"].is_array()); + } + + #[test] + fn chat_contentless_assistant_is_omitted() { + let messages = openai_chat_messages( + "", + &[ProjectedMessage { + message_id: "test".into(), + role: Role::Assistant, + content: ProjectedContent::Assistant { + text: String::new(), + thinking: String::new(), + replay_state: None, + calls: vec![], + }, + }], + ) + .unwrap(); + + assert!(messages.is_empty()); + } + #[test] fn another_provider_replay_does_not_invent_chat_reasoning_content() { let messages = openai_chat_messages(