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

@@ -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;

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);

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();

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(())
 }

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;

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<Option<ToolCall>> {
-    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"
-        );
-    }
-}

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<dyn std::error::Error + Send + Sync>> {
+                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::<Vec<_>>(),
                 "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::<Vec<_>>());
-            }
             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(

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<dyn std::error::Error + Send + Sync>> {
+                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)?;
                 }