all tools

cursor-server/src/cursor/bidi_append.rs

@@ -0,0 +1,48 @@
+use prost::Message;
+
+use crate::{
+    cursor::proto::{agent::v1 as agent, aiserver::v1 as ai},
+    run::{RunCommand, RunRegistry},
+    Error, Result,
+};
+
+pub async fn append(
+    registry: &RunRegistry,
+    request: ai::BidiAppendRequest,
+) -> Result<ai::BidiAppendResponse> {
+    let request_id = request
+        .request_id
+        .as_ref()
+        .map(|id| id.request_id.as_str())
+        .filter(|id| !id.is_empty())
+        .ok_or_else(|| Error::Protocol("BidiAppend request_id is required".into()))?;
+    if !request.data_binary.is_empty() {
+        return Err(Error::Protocol(
+            "BidiAppend data_binary is not part of the captured protocol".into(),
+        ));
+    }
+    if request.data.is_empty() {
+        return Err(Error::Protocol(
+            "BidiAppend contains no AgentClientMessage".into(),
+        ));
+    }
+    let payload = hex::decode(&request.data)
+        .map_err(|error| Error::Protocol(format!("invalid BidiAppend hex: {error}")))?;
+    let message = agent::AgentClientMessage::decode(payload.as_slice())?;
+    if let Some(agent::agent_client_message::Message::RunRequest(run)) = &message.message {
+        if let Some(conversation_id) = run.conversation_id.as_deref() {
+            registry
+                .bind_conversation(conversation_id, request_id)
+                .await;
+        }
+    }
+    registry
+        .get_or_create(request_id)
+        .await?
+        .command(RunCommand::Append {
+            seqno: request.append_seqno,
+            message: Box::new(message),
+        })
+        .await?;
+    Ok(ai::BidiAppendResponse {})
+}

cursor-server/src/cursor/blob_sync.rs

@@ -0,0 +1,248 @@
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicU32, Ordering},
+        Arc,
+    },
+    time::Duration,
+};
+
+use prost::Message;
+use tokio::sync::{oneshot, Mutex, Notify};
+
+use crate::{
+    cursor::proto::agent::v1 as pb,
+    run::RunHandle,
+    store::{BlobEdge, BlobId, Store},
+    Error, Result,
+};
+
+type BlobGetSender = oneshot::Sender<Result<Option<Vec<u8>>>>;
+
+#[derive(Clone)]
+pub struct BlobSynchronizer {
+    inner: Arc<Inner>,
+}
+
+struct Inner {
+    request_id: String,
+    store: Store,
+    handle: RunHandle,
+    next_id: AtomicU32,
+    set_requests: Mutex<HashMap<u32, BlobId>>,
+    get_requests: Mutex<HashMap<u32, BlobGetSender>>,
+    ack: Notify,
+}
+
+impl BlobSynchronizer {
+    pub fn new(request_id: String, store: Store, handle: RunHandle) -> Self {
+        Self {
+            inner: Arc::new(Inner {
+                request_id,
+                store,
+                handle,
+                next_id: AtomicU32::new(1),
+                set_requests: Mutex::new(HashMap::new()),
+                get_requests: Mutex::new(HashMap::new()),
+                ack: Notify::new(),
+            }),
+        }
+    }
+
+    pub fn request_id(&self) -> &str {
+        &self.inner.request_id
+    }
+
+    pub async fn recover(&self) -> Result<()> {
+        for item in self
+            .inner
+            .store
+            .pending_outbox(&self.inner.request_id)
+            .await?
+        {
+            if item.kind != "kv_set" {
+                continue;
+            }
+            let encoded = item
+                .key
+                .strip_prefix("blob:")
+                .ok_or_else(|| Error::Protocol(format!("invalid Blob outbox key: {}", item.key)))?;
+            let blob_id = BlobId::from_base64(encoded)?;
+            let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
+            self.inner
+                .set_requests
+                .lock()
+                .await
+                .insert(id, blob_id.clone());
+            self.inner.handle.emit(&pb::AgentServerMessage {
+                ttft_breakdown: None,
+                message: Some(pb::agent_server_message::Message::KvServerMessage(
+                    pb::KvServerMessage {
+                        id,
+                        span_context: None,
+                        message: Some(pb::kv_server_message::Message::SetBlobArgs(
+                            pb::SetBlobArgs {
+                                blob_id: blob_id.as_bytes().to_vec(),
+                                blob_data: item.payload,
+                            },
+                        )),
+                    },
+                )),
+            })?;
+            self.inner.store.mark_outbox_sent(item.id).await?;
+        }
+        self.publish_ready_checkpoints().await
+    }
+
+    pub async fn persist(&self, data: &[u8], edges: &[BlobEdge]) -> Result<BlobId> {
+        let id = self.inner.store.put_blob(data, edges).await?;
+        let key = format!("blob:{}", id.to_base64());
+        self.inner
+            .store
+            .enqueue_outbox(&self.inner.request_id, &key, "kv_set", data, &[])
+            .await?;
+        self.ensure_set(&id, data).await?;
+        Ok(id)
+    }
+
+    async fn ensure_set(&self, blob_id: &BlobId, data: &[u8]) -> Result<()> {
+        let dependency = [blob_id.clone()];
+        loop {
+            if self
+                .inner
+                .store
+                .dependencies_acked(&self.inner.request_id, &dependency)
+                .await?
+            {
+                return Ok(());
+            }
+            let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
+            self.inner
+                .set_requests
+                .lock()
+                .await
+                .insert(id, blob_id.clone());
+            self.inner.handle.emit(&pb::AgentServerMessage {
+                ttft_breakdown: None,
+                message: Some(pb::agent_server_message::Message::KvServerMessage(
+                    pb::KvServerMessage {
+                        id,
+                        span_context: None,
+                        message: Some(pb::kv_server_message::Message::SetBlobArgs(
+                            pb::SetBlobArgs {
+                                blob_id: blob_id.as_bytes().to_vec(),
+                                blob_data: data.to_vec(),
+                            },
+                        )),
+                    },
+                )),
+            })?;
+            let cancellation = self.inner.handle.cancellation();
+            tokio::select! {
+                _ = self.inner.ack.notified() => {}
+                _ = tokio::time::sleep(Duration::from_secs(2)) => {}
+                _ = cancellation.cancelled() => return Err(Error::Cancelled),
+            }
+        }
+    }
+
+    pub async fn get(&self, blob_id: &BlobId) -> Result<Option<Vec<u8>>> {
+        if let Some(data) = self.inner.store.get_blob(blob_id).await? {
+            return Ok(Some(data));
+        }
+        let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
+        let (sender, receiver) = oneshot::channel();
+        self.inner.get_requests.lock().await.insert(id, sender);
+        self.inner.handle.emit(&pb::AgentServerMessage {
+            ttft_breakdown: None,
+            message: Some(pb::agent_server_message::Message::KvServerMessage(
+                pb::KvServerMessage {
+                    id,
+                    span_context: None,
+                    message: Some(pb::kv_server_message::Message::GetBlobArgs(
+                        pb::GetBlobArgs {
+                            blob_id: blob_id.as_bytes().to_vec(),
+                        },
+                    )),
+                },
+            )),
+        })?;
+        let cancellation = self.inner.handle.cancellation();
+        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()))),
+        }
+    }
+
+    pub async fn handle_client(&self, message: pb::KvClientMessage) -> Result<()> {
+        match message.message {
+            Some(pb::kv_client_message::Message::SetBlobResult(result)) => {
+                if result.error.is_none() {
+                    if let Some(blob_id) = self.inner.set_requests.lock().await.remove(&message.id)
+                    {
+                        self.inner
+                            .store
+                            .ack_outbox(
+                                &self.inner.request_id,
+                                &format!("blob:{}", blob_id.to_base64()),
+                            )
+                            .await?;
+                        self.inner.ack.notify_waiters();
+                    }
+                }
+            }
+            Some(pb::kv_client_message::Message::GetBlobResult(result)) => {
+                if let Some(sender) = self.inner.get_requests.lock().await.remove(&message.id) {
+                    let value = if let Some(error) = result.error {
+                        Err(Error::Protocol(format!("KV GET: {}", error.message)))
+                    } else {
+                        Ok(result.blob_data)
+                    };
+                    let _ = sender.send(value);
+                }
+            }
+            None => {}
+        }
+        self.publish_ready_checkpoints().await?;
+        Ok(())
+    }
+
+    async fn publish_ready_checkpoints(&self) -> Result<()> {
+        for item in self
+            .inner
+            .store
+            .pending_outbox(&self.inner.request_id)
+            .await?
+        {
+            if item.kind != "checkpoint" {
+                continue;
+            }
+            let dependencies = item
+                .dependencies
+                .iter()
+                .map(|id| BlobId::from_base64(id))
+                .collect::<Result<Vec<_>>>()?;
+            if !self
+                .inner
+                .store
+                .dependencies_acked(&self.inner.request_id, &dependencies)
+                .await?
+            {
+                continue;
+            }
+            let checkpoint = pb::ConversationStateStructure::decode(item.payload.as_slice())?;
+            self.inner.handle.emit(&pb::AgentServerMessage {
+                ttft_breakdown: None,
+                message: Some(
+                    pb::agent_server_message::Message::ConversationCheckpointUpdate(checkpoint),
+                ),
+            })?;
+            self.inner
+                .store
+                .ack_outbox(&self.inner.request_id, &item.key)
+                .await?;
+        }
+        Ok(())
+    }
+}

cursor-server/src/cursor/checkpoint.rs

@@ -0,0 +1,384 @@
+use prost::Message;
+
+use std::collections::HashSet;
+
+use crate::{
+    cursor::{blob_sync::BlobSynchronizer, interaction::render_tool_call, proto::agent::v1 as pb},
+    model::{CanonicalMessage, MessageContent, Origin, Role, ToolCall},
+    prompting::fold_derived_state,
+    run::RunHandle,
+    store::{BlobEdge, BlobId, Store},
+    Error, Result,
+};
+
+pub struct CheckpointBuilder {
+    store: Store,
+    sync: BlobSynchronizer,
+}
+
+impl CheckpointBuilder {
+    pub fn new(store: Store, sync: BlobSynchronizer) -> Self {
+        Self { store, sync }
+    }
+
+    pub async fn build(
+        &self,
+        conversation_id: &str,
+        revision: i64,
+        messages: &[CanonicalMessage],
+        mode: i32,
+    ) -> Result<pb::ConversationStateStructure> {
+        self.build_with_tool_progress(
+            conversation_id,
+            revision,
+            messages,
+            mode,
+            &[],
+            &HashSet::new(),
+        )
+        .await
+    }
+
+    pub async fn build_with_tool_progress(
+        &self,
+        conversation_id: &str,
+        revision: i64,
+        messages: &[CanonicalMessage],
+        mode: i32,
+        active_calls: &[ToolCall],
+        completed: &HashSet<String>,
+    ) -> Result<pb::ConversationStateStructure> {
+        let mut root_ids = Vec::with_capacity(messages.len());
+        for message in messages {
+            root_ids.push(
+                self.sync
+                    .persist(&serde_json::to_vec(message)?, &[])
+                    .await?,
+            );
+        }
+        let turn_ids = self.build_turns(messages, mode, completed).await?;
+        let (todo_ids, plan_id) = self.build_derived_state(messages).await?;
+        let checkpoint = pb::ConversationStateStructure {
+            root_prompt_messages_json: root_ids.iter().map(|id| id.as_bytes().to_vec()).collect(),
+            turns: turn_ids.iter().map(|id| id.as_bytes().to_vec()).collect(),
+            todos: todo_ids.iter().map(|id| id.as_bytes().to_vec()).collect(),
+            plan: plan_id.as_ref().map(|id| id.as_bytes().to_vec()),
+            pending_tool_calls: active_calls
+                .iter()
+                .filter(|call| !completed.contains(&call.call_id))
+                .map(|call| call.call_id.clone())
+                .collect(),
+            mode: Some(mode),
+            ..Default::default()
+        };
+        let mut encoded = Vec::new();
+        checkpoint.encode(&mut encoded)?;
+        let mut edges = root_ids
+            .iter()
+            .enumerate()
+            .map(|(index, child)| BlobEdge {
+                child: child.clone(),
+                field_name: format!("root_prompt_messages_json[{index}]"),
+            })
+            .collect::<Vec<_>>();
+        edges.extend(turn_ids.iter().enumerate().map(|(index, child)| BlobEdge {
+            child: child.clone(),
+            field_name: format!("turns[{index}]"),
+        }));
+        edges.extend(todo_ids.iter().enumerate().map(|(index, child)| BlobEdge {
+            child: child.clone(),
+            field_name: format!("todos[{index}]"),
+        }));
+        if let Some(child) = plan_id {
+            edges.push(BlobEdge {
+                child,
+                field_name: "plan".into(),
+            });
+        }
+        let head = self.sync.persist(&encoded, &edges).await?;
+        if !self
+            .store
+            .publish_head(conversation_id, revision, &head)
+            .await?
+        {
+            return Err(Error::Cancelled);
+        }
+        let dependencies = self.store.blob_closure(std::slice::from_ref(&head)).await?;
+        self.store
+            .enqueue_outbox(
+                self.sync.request_id(),
+                &format!("checkpoint:{}", head.to_base64()),
+                "checkpoint",
+                &encoded,
+                &dependencies,
+            )
+            .await?;
+        Ok(checkpoint)
+    }
+
+    pub async fn publish(
+        &self,
+        handle: &RunHandle,
+        checkpoint: &pb::ConversationStateStructure,
+    ) -> Result<()> {
+        let encoded = checkpoint.encode_to_vec();
+        let head = BlobId::digest(&encoded);
+        let dependencies = self.store.blob_closure(std::slice::from_ref(&head)).await?;
+        if !self
+            .store
+            .dependencies_acked(self.sync.request_id(), &dependencies)
+            .await?
+        {
+            return Err(Error::Protocol(format!(
+                "checkpoint {} published before Blob ACK barrier",
+                head.to_base64()
+            )));
+        }
+        handle.emit(&pb::AgentServerMessage {
+            ttft_breakdown: None,
+            message: Some(
+                pb::agent_server_message::Message::ConversationCheckpointUpdate(checkpoint.clone()),
+            ),
+        })?;
+        self.store
+            .ack_outbox(
+                self.sync.request_id(),
+                &format!("checkpoint:{}", head.to_base64()),
+            )
+            .await?;
+        Ok(())
+    }
+
+    async fn build_derived_state(
+        &self,
+        messages: &[CanonicalMessage],
+    ) -> Result<(Vec<BlobId>, Option<BlobId>)> {
+        let state = fold_derived_state(messages);
+        let todo_values = state
+            .todos
+            .as_ref()
+            .and_then(|value| value.get("todos").or(Some(value)))
+            .and_then(serde_json::Value::as_array);
+        let mut todo_ids = Vec::new();
+        for todo in todo_values.into_iter().flatten() {
+            let status = match todo
+                .get("status")
+                .and_then(serde_json::Value::as_str)
+                .unwrap_or("pending")
+            {
+                "in_progress" => pb::TodoStatus::InProgress,
+                "completed" => pb::TodoStatus::Completed,
+                "cancelled" => pb::TodoStatus::Cancelled,
+                _ => pb::TodoStatus::Pending,
+            };
+            let message = pb::TodoItem {
+                id: todo
+                    .get("id")
+                    .and_then(serde_json::Value::as_str)
+                    .unwrap_or_default()
+                    .into(),
+                content: todo
+                    .get("content")
+                    .and_then(serde_json::Value::as_str)
+                    .unwrap_or_default()
+                    .into(),
+                status: status as i32,
+                created_at: 0,
+                updated_at: 0,
+                dependencies: todo
+                    .get("dependencies")
+                    .and_then(serde_json::Value::as_array)
+                    .into_iter()
+                    .flatten()
+                    .filter_map(serde_json::Value::as_str)
+                    .map(str::to_string)
+                    .collect(),
+            };
+            let mut encoded = Vec::new();
+            message.encode(&mut encoded)?;
+            todo_ids.push(self.sync.persist(&encoded, &[]).await?);
+        }
+        let plan_id = if let Some(value) = state.plan {
+            let text = value
+                .get("plan")
+                .and_then(serde_json::Value::as_str)
+                .or_else(|| value.as_str())
+                .unwrap_or_else(|| {
+                    value
+                        .get("overview")
+                        .and_then(serde_json::Value::as_str)
+                        .unwrap_or_default()
+                });
+            let mut encoded = Vec::new();
+            pb::ConversationPlan { plan: text.into() }.encode(&mut encoded)?;
+            Some(self.sync.persist(&encoded, &[]).await?)
+        } else {
+            None
+        };
+        Ok((todo_ids, plan_id))
+    }
+
+    async fn build_turns(
+        &self,
+        messages: &[CanonicalMessage],
+        mode: i32,
+        completed_overlay: &HashSet<String>,
+    ) -> Result<Vec<BlobId>> {
+        let mut completed = completed_overlay.clone();
+        for message in messages {
+            if let MessageContent::ToolResult(result) = &message.content {
+                completed.insert(result.call_id.clone());
+            }
+        }
+        let mut turns = Vec::<(CanonicalMessage, Vec<&CanonicalMessage>)>::new();
+        for message in messages {
+            if message.role == Role::User && message.origin == Origin::User {
+                turns.push((message.clone(), Vec::new()));
+            } else if matches!(message.origin, Origin::Assistant | Origin::Tool) {
+                if let Some((_, steps)) = turns.last_mut() {
+                    steps.push(message);
+                }
+            }
+        }
+        let mut turn_ids = Vec::with_capacity(turns.len());
+        for (user, step_messages) in turns {
+            let text = match user.content {
+                MessageContent::Text { text } => text,
+                other => serde_json::to_string(&other)?,
+            };
+            let user_message = pb::UserMessage {
+                text,
+                message_id: user.message_id.clone(),
+                mode,
+                ..Default::default()
+            };
+            let mut encoded = Vec::new();
+            user_message.encode(&mut encoded)?;
+            let user_id = self.sync.persist(&encoded, &[]).await?;
+            let mut step_ids = Vec::new();
+            for message in step_messages {
+                for step in message_steps(message, &completed)? {
+                    let mut encoded = Vec::new();
+                    step.encode(&mut encoded)?;
+                    step_ids.push(self.sync.persist(&encoded, &[]).await?);
+                }
+            }
+            let turn = pb::ConversationTurnStructure {
+                turn: Some(
+                    pb::conversation_turn_structure::Turn::AgentConversationTurn(
+                        pb::AgentConversationTurnStructure {
+                            user_message: user_id.as_bytes().to_vec(),
+                            steps: step_ids.iter().map(|id| id.as_bytes().to_vec()).collect(),
+                            request_id: None,
+                            encrypted_model: None,
+                            dynamic_tool_count: None,
+                            send_message_step_indices: Vec::new(),
+                        },
+                    ),
+                ),
+            };
+            let mut encoded = Vec::new();
+            turn.encode(&mut encoded)?;
+            let mut edges = vec![BlobEdge {
+                child: user_id,
+                field_name: "agent_conversation_turn.user_message".into(),
+            }];
+            edges.extend(
+                step_ids
+                    .into_iter()
+                    .enumerate()
+                    .map(|(index, child)| BlobEdge {
+                        child,
+                        field_name: format!("agent_conversation_turn.steps[{index}]"),
+                    }),
+            );
+            turn_ids.push(self.sync.persist(&encoded, &edges).await?);
+        }
+        Ok(turn_ids)
+    }
+
+    pub async fn import_prefetched(&self, blobs: &[pb::PreFetchedBlob]) -> Result<()> {
+        for blob in blobs {
+            let expected = BlobId::from_bytes(&blob.id)?;
+            let actual = self.store.put_blob(&blob.value, &[]).await?;
+            if expected != actual {
+                return Err(Error::Protocol(format!(
+                    "prefetched Blob hash mismatch: {}",
+                    expected.to_base64()
+                )));
+            }
+        }
+        Ok(())
+    }
+
+    pub async fn hydrate_messages(
+        &self,
+        state: Option<&pb::ConversationStateStructure>,
+    ) -> Result<Vec<CanonicalMessage>> {
+        let mut messages = Vec::new();
+        let Some(state) = state else {
+            return Ok(messages);
+        };
+        for raw_id in &state.root_prompt_messages_json {
+            let id = BlobId::from_bytes(raw_id)?;
+            let Some(data) = self.sync.get(&id).await? else {
+                return Err(Error::Protocol(format!(
+                    "missing message Blob {}",
+                    id.to_base64()
+                )));
+            };
+            messages.push(serde_json::from_slice(&data)?);
+        }
+        Ok(messages)
+    }
+}
+
+fn message_steps(
+    message: &CanonicalMessage,
+    completed: &HashSet<String>,
+) -> Result<Vec<pb::ConversationStep>> {
+    use pb::conversation_step::Message;
+    match &message.content {
+        MessageContent::Assistant {
+            text,
+            thinking,
+            tool_calls,
+            ..
+        } => {
+            let mut steps = Vec::new();
+            if !thinking.is_empty() {
+                steps.push(pb::ConversationStep {
+                    message: Some(Message::ThinkingMessage(pb::ThinkingMessage {
+                        text: thinking.clone(),
+                        duration_ms: 0,
+                    })),
+                });
+            }
+            if !text.is_empty() {
+                steps.push(pb::ConversationStep {
+                    message: Some(Message::AssistantMessage(pb::AssistantMessage {
+                        text: text.clone(),
+                    })),
+                });
+            }
+            for call in tool_calls {
+                let tool = render_tool_call(
+                    &ToolCall {
+                        index: 0,
+                        call_id: call.call_id.clone(),
+                        model_call_id: String::new(),
+                        name: call.name.clone(),
+                        arguments_text: serde_json::to_string(&call.arguments).unwrap_or_default(),
+                        arguments: call.arguments.clone(),
+                    },
+                    completed.contains(&call.call_id),
+                )?;
+                steps.push(pb::ConversationStep {
+                    message: Some(Message::ToolCall(tool)),
+                });
+            }
+            Ok(steps)
+        }
+        _ => Ok(Vec::new()),
+    }
+}

cursor-server/src/cursor/connect.rs

@@ -0,0 +1,121 @@
+use bytes::{BufMut, Bytes, BytesMut};
+use prost::Message;
+use serde::Serialize;
+
+use crate::{Error, Result};
+
+pub const END_STREAM_FLAG: u8 = 0x02;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ConnectCode {
+    Canceled,
+    InvalidArgument,
+    NotFound,
+    Unavailable,
+    Internal,
+}
+
+impl ConnectCode {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Canceled => "canceled",
+            Self::InvalidArgument => "invalid_argument",
+            Self::NotFound => "not_found",
+            Self::Unavailable => "unavailable",
+            Self::Internal => "internal",
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+pub struct ConnectErrorDetail {
+    #[serde(rename = "type")]
+    pub type_name: String,
+    pub value: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ConnectStreamError {
+    pub code: ConnectCode,
+    pub message: String,
+    pub details: Vec<ConnectErrorDetail>,
+}
+
+#[derive(Serialize)]
+struct EndStreamResponse<'a> {
+    error: WireError<'a>,
+}
+
+#[derive(Serialize)]
+struct WireError<'a> {
+    code: &'static str,
+    #[serde(skip_serializing_if = "str::is_empty")]
+    message: &'a str,
+    #[serde(skip_serializing_if = "details_are_empty")]
+    details: &'a [ConnectErrorDetail],
+}
+
+fn details_are_empty(details: &&[ConnectErrorDetail]) -> bool {
+    details.is_empty()
+}
+
+pub fn encode_message<M: Message>(message: &M) -> Result<Bytes> {
+    let len = message.encoded_len();
+    let mut output = BytesMut::with_capacity(5 + len);
+    output.put_u8(0);
+    output.put_u32(len as u32);
+    message.encode(&mut output)?;
+    Ok(output.freeze())
+}
+
+pub fn encode_end_stream() -> Bytes {
+    encode_end_stream_payload(b"{}")
+}
+
+pub fn encode_error_end_stream(error: &ConnectStreamError) -> Result<Bytes> {
+    let payload = serde_json::to_vec(&EndStreamResponse {
+        error: WireError {
+            code: error.code.as_str(),
+            message: &error.message,
+            details: &error.details,
+        },
+    })?;
+    Ok(encode_end_stream_payload(&payload))
+}
+
+fn encode_end_stream_payload(payload: &[u8]) -> Bytes {
+    let mut output = BytesMut::with_capacity(5 + payload.len());
+    output.put_u8(END_STREAM_FLAG);
+    output.put_u32(payload.len() as u32);
+    output.extend_from_slice(payload);
+    output.freeze()
+}
+
+pub fn decode_unary<M: Message + Default>(body: &[u8]) -> Result<M> {
+    if body.len() >= 5 {
+        let flags = body[0];
+        let length = u32::from_be_bytes(body[1..5].try_into().expect("four bytes")) as usize;
+        if flags & END_STREAM_FLAG == 0 && length == body.len() - 5 {
+            return Ok(M::decode(&body[5..])?);
+        }
+    }
+    Ok(M::decode(body)?)
+}
+
+pub fn decode_frames(mut body: &[u8]) -> Result<Vec<(u8, Bytes)>> {
+    let mut frames = Vec::new();
+    while !body.is_empty() {
+        if body.len() < 5 {
+            return Err(Error::Protocol("truncated Connect envelope".into()));
+        }
+        let flags = body[0];
+        let length = u32::from_be_bytes(body[1..5].try_into().expect("four bytes")) as usize;
+        body = &body[5..];
+        if body.len() < length {
+            return Err(Error::Protocol("truncated Connect payload".into()));
+        }
+        frames.push((flags, Bytes::copy_from_slice(&body[..length])));
+        body = &body[length..];
+    }
+    Ok(frames)
+}

cursor-server/src/cursor/exec.rs

@@ -0,0 +1,535 @@
+use serde_json::{Map, Value};
+
+use crate::{
+    cursor::{
+        pending::{ExecContext, PendingExecRegistry},
+        proto::agent::v1 as pb,
+        tool_result::ToolCompletion,
+    },
+    model::ToolCall,
+    Error, Result,
+};
+
+pub fn request(id: u32, call: &ToolCall, context: &ExecContext) -> Result<pb::AgentServerMessage> {
+    use pb::exec_server_message::Message;
+    let string = |name: &str| {
+        call.arguments
+            .get(name)
+            .and_then(Value::as_str)
+            .map(str::to_string)
+            .ok_or_else(|| Error::Protocol(format!("{} is missing {name}", call.name)))
+    };
+    let optional_string = |name: &str| {
+        call.arguments
+            .get(name)
+            .and_then(Value::as_str)
+            .map(str::to_string)
+    };
+    let int = |name: &str| {
+        call.arguments
+            .get(name)
+            .and_then(Value::as_i64)
+            .map(|v| v as i32)
+    };
+    let message = match normalize(&call.name).as_str() {
+        "shell" => Message::ShellStreamArgs(pb::ShellArgs {
+            command: string("command")?,
+            working_directory: optional_string("working_directory").unwrap_or_default(),
+            timeout: shell_timeout(call)?,
+            tool_call_id: call.call_id.clone(),
+            file_output_threshold_bytes: Some(40_000),
+            timeout_behavior: pb::TimeoutBehavior::Background as i32,
+            hard_timeout: Some(86_400_000),
+            description: optional_string("description"),
+            close_stdin: true,
+            conversation_id: Some(context.conversation_id.clone()),
+            admin_command_denylist: context.admin_command_denylist.clone(),
+            ..Default::default()
+        }),
+        "forcebackgroundshell" => Message::ForceBackgroundShellArgs(pb::ForceBackgroundShellArgs {
+            tool_call_id: string("tool_call_id")?,
+        }),
+        "read" => Message::ReadArgs(pb::ReadArgs {
+            path: string("path")?,
+            tool_call_id: call.call_id.clone(),
+            offset: int("offset"),
+            limit: call
+                .arguments
+                .get("limit")
+                .and_then(Value::as_u64)
+                .map(|v| v as u32),
+            encoding_hint: optional_string("encoding_hint"),
+        }),
+        "write" => Message::WriteArgs(pb::WriteArgs {
+            path: string("path")?,
+            file_text: string("contents")?,
+            tool_call_id: call.call_id.clone(),
+            return_file_content_after_write: true,
+            file_bytes: Vec::new(),
+            encoding_hint: optional_string("encoding_hint"),
+        }),
+        "delete" => Message::DeleteArgs(pb::DeleteArgs {
+            path: string("path")?,
+            tool_call_id: call.call_id.clone(),
+        }),
+        "grep" => Message::GrepArgs(pb::GrepArgs {
+            pattern: string("pattern")?,
+            path: optional_string("path"),
+            glob: optional_string("glob"),
+            output_mode: optional_string("output_mode"),
+            context_before: int("context_before"),
+            context_after: int("context_after"),
+            context: int("context"),
+            case_insensitive: call
+                .arguments
+                .get("case_insensitive")
+                .and_then(Value::as_bool),
+            r#type: optional_string("type"),
+            head_limit: int("head_limit"),
+            multiline: call.arguments.get("multiline").and_then(Value::as_bool),
+            sort: optional_string("sort"),
+            sort_ascending: call
+                .arguments
+                .get("sort_ascending")
+                .and_then(Value::as_bool),
+            tool_call_id: call.call_id.clone(),
+            sandbox_policy: None,
+            offset: int("offset"),
+        }),
+        "glob" => Message::GrepArgs(pb::GrepArgs {
+            pattern: String::new(),
+            path: optional_string("target_directory"),
+            glob: optional_string("glob_pattern"),
+            output_mode: Some("files_with_matches".into()),
+            tool_call_id: call.call_id.clone(),
+            ..Default::default()
+        }),
+        "ls" => Message::LsArgs(pb::LsArgs {
+            path: string("path")?,
+            ignore: call
+                .arguments
+                .get("ignore")
+                .and_then(Value::as_array)
+                .map(|v| {
+                    v.iter()
+                        .filter_map(Value::as_str)
+                        .map(str::to_string)
+                        .collect()
+                })
+                .unwrap_or_default(),
+            tool_call_id: call.call_id.clone(),
+            sandbox_policy: None,
+            timeout_ms: call
+                .arguments
+                .get("timeout_ms")
+                .and_then(Value::as_u64)
+                .map(|v| v as u32),
+        }),
+        "readlints" => Message::DiagnosticsArgs(pb::DiagnosticsArgs {
+            path: call
+                .arguments
+                .get("paths")
+                .and_then(Value::as_array)
+                .and_then(|paths| paths.first())
+                .and_then(Value::as_str)
+                .unwrap_or_default()
+                .into(),
+            tool_call_id: call.call_id.clone(),
+        }),
+        "patchedit" => Message::PiEditArgs(pb::PiEditExecArgs {
+            path: string("path")?,
+            edits: vec![pb::PiEditReplacement {
+                old_text: string("old_string")?,
+                new_text: string("new_string")?,
+            }],
+        }),
+        "writeshellstdin" => Message::WriteShellStdinArgs(pb::WriteShellStdinArgs {
+            shell_id: call
+                .arguments
+                .get("shell_id")
+                .and_then(Value::as_u64)
+                .unwrap_or_default() as u32,
+            chars: string("chars")?,
+        }),
+        "task" => Message::SubagentArgs(pb::SubagentArgs {
+            tool_call_id: call.call_id.clone(),
+            subagent_type: string("subagent_type")?,
+            model_id: optional_string("model").unwrap_or_default(),
+            prompt: string("prompt")?,
+            readonly: call
+                .arguments
+                .get("readonly")
+                .and_then(Value::as_bool)
+                .unwrap_or(false),
+            resume_agent_id: optional_string("resume"),
+            run_in_background: Some(false),
+            continuation_config: None,
+            parent_conversation_id: None,
+            interrupt: None,
+            mode: 0,
+            fork_agent_id: None,
+            root_parent_conversation_id: None,
+            selected_context: None,
+            direct_meta_parent_child_subagent: None,
+            environment: 0,
+            cloud_base_branch: None,
+            credentials: None,
+        }),
+        "callmcptool" => Message::McpArgs(pb::McpArgs {
+            name: string("toolName")?,
+            args: call
+                .arguments
+                .get("arguments")
+                .and_then(Value::as_object)
+                .map(json_object_to_prost)
+                .unwrap_or_default(),
+            tool_call_id: call.call_id.clone(),
+            provider_identifier: optional_string("provider_identifier").unwrap_or_default(),
+            tool_name: string("toolName")?,
+            smart_mode_approval: None,
+            smart_mode_approval_only: false,
+            skip_approval: false,
+            server_identifier: string("server")?,
+        }),
+        "fetchmcpresource" => Message::ReadMcpResourceExecArgs(pb::ReadMcpResourceExecArgs {
+            server: string("server")?,
+            uri: string("uri")?,
+            download_path: optional_string("downloadPath"),
+            tool_call_id: call.call_id.clone(),
+            smart_mode_approval: None,
+        }),
+        "webfetch" => Message::FetchArgs(pb::FetchArgs {
+            url: string("url")?,
+            tool_call_id: call.call_id.clone(),
+        }),
+        other => {
+            return Err(Error::Protocol(format!(
+                "tool {other} is not executed through ExecServerMessage"
+            )))
+        }
+    };
+    Ok(pb::AgentServerMessage {
+        ttft_breakdown: None,
+        message: Some(pb::agent_server_message::Message::ExecServerMessage(
+            pb::ExecServerMessage {
+                id,
+                exec_id: call.call_id.clone(),
+                span_context: None,
+                accept_hook_additional_contexts: Some(true),
+                message: Some(message),
+            },
+        )),
+    })
+}
+
+pub fn mcp_request(
+    id: u32,
+    call: &ToolCall,
+    definition: &pb::McpToolDefinition,
+) -> Result<pb::AgentServerMessage> {
+    let args = call
+        .arguments
+        .as_object()
+        .map(json_object_to_prost)
+        .unwrap_or_default();
+    Ok(pb::AgentServerMessage {
+        ttft_breakdown: None,
+        message: Some(pb::agent_server_message::Message::ExecServerMessage(
+            pb::ExecServerMessage {
+                id,
+                exec_id: call.call_id.clone(),
+                span_context: None,
+                accept_hook_additional_contexts: None,
+                message: Some(pb::exec_server_message::Message::McpArgs(pb::McpArgs {
+                    name: definition.name.clone(),
+                    args,
+                    tool_call_id: call.call_id.clone(),
+                    provider_identifier: definition.provider_identifier.clone(),
+                    tool_name: if definition.tool_name.is_empty() {
+                        definition.name.clone()
+                    } else {
+                        definition.tool_name.clone()
+                    },
+                    smart_mode_approval: None,
+                    smart_mode_approval_only: false,
+                    skip_approval: false,
+                    server_identifier: String::new(),
+                })),
+            },
+        )),
+    })
+}
+
+pub fn abort(id: u32) -> pb::AgentServerMessage {
+    pb::AgentServerMessage {
+        ttft_breakdown: None,
+        message: Some(pb::agent_server_message::Message::ExecServerControlMessage(
+            pb::ExecServerControlMessage {
+                message: Some(pb::exec_server_control_message::Message::Abort(
+                    pb::ExecServerAbort { id },
+                )),
+            },
+        )),
+    }
+}
+
+pub enum ClientExecEvent {
+    Delta(Box<pb::AgentServerMessage>),
+    Completed(Box<ToolCompletion>),
+    Pending,
+}
+
+pub async fn client_event(
+    message: &pb::ExecClientMessage,
+    pending: &PendingExecRegistry,
+) -> Result<ClientExecEvent> {
+    let call = pending
+        .call(message.id)
+        .await
+        .ok_or_else(|| Error::Protocol(format!("unknown ExecClientMessage id: {}", message.id)))?;
+    let Some(wire_result) = &message.message else {
+        return Ok(ClientExecEvent::Pending);
+    };
+    let pb::exec_client_message::Message::ShellStream(stream) = wire_result else {
+        return complete(message.id, pending, wire_result.clone()).await;
+    };
+    use pb::shell_stream::Event;
+    let event = match &stream.event {
+        Some(Event::Stdout(stdout)) => {
+            if pending.append_stdout(message.id, &stdout.data).await {
+                ClientExecEvent::Delta(Box::new(shell_delta(&call, true, &stdout.data)))
+            } else {
+                ClientExecEvent::Pending
+            }
+        }
+        Some(Event::Stderr(stderr)) => {
+            if pending.append_stderr(message.id, &stderr.data).await {
+                ClientExecEvent::Delta(Box::new(shell_delta(&call, false, &stderr.data)))
+            } else {
+                ClientExecEvent::Pending
+            }
+        }
+        Some(Event::Start(_)) | Some(Event::HookContext(_)) => ClientExecEvent::Pending,
+        Some(Event::Exit(exit)) => {
+            let entry = take(message.id, pending).await?;
+            let result = shell_exit_result(message, exit, &entry.stdout, &entry.stderr);
+            completed(entry, pb::exec_client_message::Message::ShellResult(result))?
+        }
+        Some(Event::Backgrounded(backgrounded)) => {
+            let entry = take(message.id, pending).await?;
+            let result = shell_backgrounded_result(
+                backgrounded,
+                &entry.stdout,
+                &entry.stderr,
+                &entry.context.terminals_folder,
+            );
+            completed(entry, pb::exec_client_message::Message::ShellResult(result))?
+        }
+        Some(Event::Rejected(value)) => {
+            let result = pb::ShellResult {
+                result: Some(pb::shell_result::Result::Rejected(value.clone())),
+                ..Default::default()
+            };
+            complete(
+                message.id,
+                pending,
+                pb::exec_client_message::Message::ShellResult(result),
+            )
+            .await?
+        }
+        Some(Event::PermissionDenied(value)) => {
+            let result = pb::ShellResult {
+                result: Some(pb::shell_result::Result::PermissionDenied(value.clone())),
+                ..Default::default()
+            };
+            complete(
+                message.id,
+                pending,
+                pb::exec_client_message::Message::ShellResult(result),
+            )
+            .await?
+        }
+        Some(Event::SandboxUnsupported(value)) => {
+            let result = pb::ShellResult {
+                result: Some(pb::shell_result::Result::SpawnError(pb::ShellSpawnError {
+                    command: value.command.clone(),
+                    working_directory: value.working_directory.clone(),
+                    error: value.reason.clone(),
+                })),
+                ..Default::default()
+            };
+            complete(
+                message.id,
+                pending,
+                pb::exec_client_message::Message::ShellResult(result),
+            )
+            .await?
+        }
+        None => ClientExecEvent::Pending,
+    };
+    Ok(event)
+}
+
+async fn complete(
+    id: u32,
+    pending: &PendingExecRegistry,
+    result: pb::exec_client_message::Message,
+) -> Result<ClientExecEvent> {
+    completed(take(id, pending).await?, result)
+}
+
+async fn take(id: u32, pending: &PendingExecRegistry) -> Result<super::pending::PendingExec> {
+    pending
+        .take(id)
+        .await
+        .ok_or_else(|| Error::Protocol(format!("unknown terminal Exec id: {id}")))
+}
+
+fn completed(
+    pending: super::pending::PendingExec,
+    result: pb::exec_client_message::Message,
+) -> Result<ClientExecEvent> {
+    Ok(ClientExecEvent::Completed(Box::new(
+        super::tool_result::from_exec(pending, &result)?,
+    )))
+}
+
+fn shell_exit_result(
+    message: &pb::ExecClientMessage,
+    exit: &pb::ShellStreamExit,
+    stdout: &str,
+    stderr: &str,
+) -> pb::ShellResult {
+    let result = if exit.code == 0 && !exit.aborted {
+        pb::shell_result::Result::Success(pb::ShellSuccess {
+            working_directory: exit.cwd.clone(),
+            exit_code: exit.code as i32,
+            stdout: stdout.into(),
+            stderr: stderr.into(),
+            interleaved_output: Some(format!("{stdout}{stderr}")),
+            local_execution_time_ms: exit
+                .local_execution_time_ms
+                .or(message.local_execution_time_ms),
+            ..Default::default()
+        })
+    } else {
+        pb::shell_result::Result::Failure(pb::ShellFailure {
+            working_directory: exit.cwd.clone(),
+            exit_code: exit.code as i32,
+            stdout: stdout.into(),
+            stderr: stderr.into(),
+            interleaved_output: Some(format!("{stdout}{stderr}")),
+            abort_reason: exit.abort_reason,
+            aborted: exit.aborted,
+            local_execution_time_ms: exit
+                .local_execution_time_ms
+                .or(message.local_execution_time_ms),
+            ..Default::default()
+        })
+    };
+    pb::ShellResult {
+        result: Some(result),
+        is_background: Some(false),
+        ..Default::default()
+    }
+}
+
+fn shell_backgrounded_result(
+    backgrounded: &pb::ShellStreamBackgrounded,
+    stdout: &str,
+    stderr: &str,
+    terminals_folder: &str,
+) -> pb::ShellResult {
+    pb::ShellResult {
+        result: Some(pb::shell_result::Result::Success(pb::ShellSuccess {
+            command: backgrounded.command.clone(),
+            working_directory: backgrounded.working_directory.clone(),
+            stdout: stdout.into(),
+            stderr: stderr.into(),
+            shell_id: Some(backgrounded.shell_id),
+            pid: backgrounded.pid,
+            ms_to_wait: backgrounded.ms_to_wait,
+            background_reason: backgrounded.reason,
+            interleaved_output: Some(format!("{stdout}{stderr}")),
+            ..Default::default()
+        })),
+        is_background: Some(true),
+        terminals_folder: (!terminals_folder.is_empty()).then(|| terminals_folder.into()),
+        pid: backgrounded.pid,
+        ..Default::default()
+    }
+}
+
+fn shell_delta(call: &ToolCall, stdout: bool, content: &str) -> pb::AgentServerMessage {
+    let delta = if stdout {
+        pb::shell_tool_call_delta::Delta::Stdout(pb::ShellToolCallStdoutDelta {
+            content: content.into(),
+        })
+    } else {
+        pb::shell_tool_call_delta::Delta::Stderr(pb::ShellToolCallStderrDelta {
+            content: content.into(),
+        })
+    };
+    super::interaction::server_interaction(pb::interaction_update::Message::ToolCallDelta(
+        Box::new(pb::ToolCallDeltaUpdate {
+            call_id: call.call_id.clone(),
+            tool_call_delta: Some(Box::new(pb::ToolCallDelta {
+                delta: Some(pb::tool_call_delta::Delta::ShellToolCallDelta(
+                    pb::ShellToolCallDelta { delta: Some(delta) },
+                )),
+            })),
+            model_call_id: call.model_call_id.clone(),
+        }),
+    ))
+}
+
+fn shell_timeout(call: &ToolCall) -> Result<i32> {
+    let value = call
+        .arguments
+        .get("block_until_ms")
+        .map(|value| {
+            value
+                .as_i64()
+                .ok_or_else(|| Error::Protocol("Shell block_until_ms must be an integer".into()))
+        })
+        .transpose()?
+        .unwrap_or(30_000);
+    i32::try_from(value)
+        .ok()
+        .filter(|value| *value >= 0)
+        .ok_or_else(|| Error::Protocol("Shell block_until_ms is out of range".into()))
+}
+
+fn normalize(value: &str) -> String {
+    value
+        .chars()
+        .filter(|c| c.is_ascii_alphanumeric())
+        .flat_map(char::to_lowercase)
+        .collect()
+}
+
+pub(crate) fn json_object_to_prost(
+    value: &Map<String, Value>,
+) -> std::collections::HashMap<String, prost_types::Value> {
+    value
+        .iter()
+        .map(|(key, value)| (key.clone(), prost_value(value)))
+        .collect()
+}
+
+fn prost_value(value: &Value) -> prost_types::Value {
+    use prost_types::{value::Kind, ListValue, Struct, Value as ProstValue};
+    let kind = match value {
+        Value::Null => Kind::NullValue(0),
+        Value::Bool(v) => Kind::BoolValue(*v),
+        Value::Number(v) => Kind::NumberValue(v.as_f64().unwrap_or_default()),
+        Value::String(v) => Kind::StringValue(v.clone()),
+        Value::Array(v) => Kind::ListValue(ListValue {
+            values: v.iter().map(prost_value).collect(),
+        }),
+        Value::Object(v) => Kind::StructValue(Struct {
+            fields: json_object_to_prost(v).into_iter().collect(),
+        }),
+    };
+    ProstValue { kind: Some(kind) }
+}

cursor-server/src/cursor/handlers.rs

@@ -0,0 +1,53 @@
+use axum::{
+    body::Bytes,
+    extract::{DefaultBodyLimit, State},
+    http::{header, HeaderValue, Response, StatusCode},
+    routing::post,
+    Router,
+};
+use tower_http::decompression::RequestDecompressionLayer;
+
+use crate::{
+    cursor::{
+        bidi_append, connect,
+        proto::{agent::v1 as agent, aiserver::v1 as ai},
+        run_sse,
+    },
+    run::RunRegistry,
+    Result,
+};
+
+pub fn router(registry: RunRegistry) -> Router {
+    Router::new()
+        .route("/agent.v1.AgentService/RunSSE", post(run_sse_handler))
+        .route(
+            "/aiserver.v1.BidiService/BidiAppend",
+            post(bidi_append_handler),
+        )
+        .layer(DefaultBodyLimit::disable())
+        .layer(RequestDecompressionLayer::new())
+        .with_state(registry)
+}
+
+async fn run_sse_handler(
+    State(registry): State<RunRegistry>,
+    body: Bytes,
+) -> Result<Response<axum::body::Body>> {
+    let request: agent::BidiRequestId = connect::decode_unary(&body)?;
+    run_sse::stream(&registry, &request.request_id).await
+}
+
+async fn bidi_append_handler(
+    State(registry): State<RunRegistry>,
+    body: Bytes,
+) -> Result<Response<axum::body::Body>> {
+    let request: ai::BidiAppendRequest = connect::decode_unary(&body)?;
+    bidi_append::append(&registry, request).await?;
+    let mut response = Response::new(axum::body::Body::empty());
+    *response.status_mut() = StatusCode::OK;
+    response.headers_mut().insert(
+        header::CONTENT_TYPE,
+        HeaderValue::from_static("application/proto"),
+    );
+    Ok(response)
+}

cursor-server/src/cursor/interaction.rs

@@ -0,0 +1,547 @@
+use crate::{
+    cursor::{
+        proto::agent::v1 as pb,
+        tool_result::{self, ToolCompletion},
+    },
+    model::{ToolCall, Usage},
+    provider::ResponseEvent,
+};
+use serde_json::Value;
+use std::time::Duration;
+
+use crate::{Error, Result};
+
+pub fn response_event(
+    event: &ResponseEvent,
+    model_call_id: &str,
+) -> Result<Option<pb::AgentServerMessage>> {
+    use pb::interaction_update::Message;
+    let message = match event {
+        ResponseEvent::TextDelta(text) => Message::TextDelta(pb::TextDeltaUpdate {
+            text: text.clone(),
+            is_server_notice: false,
+        }),
+        ResponseEvent::ThinkingDelta(text) => Message::ThinkingDelta(pb::ThinkingDeltaUpdate {
+            text: text.clone(),
+            thinking_style: Some(pb::ThinkingStyle::Default as i32),
+        }),
+        ResponseEvent::ToolCallStart { call_id, name, .. } => {
+            Message::PartialToolCall(pb::PartialToolCallUpdate {
+                call_id: call_id.clone(),
+                tool_call: Some(tool_placeholder(name, call_id)?),
+                args_text_delta: String::new(),
+                model_call_id: model_call_id.into(),
+            })
+        }
+        ResponseEvent::ToolCallArgumentsDelta { .. } => return Ok(None),
+        ResponseEvent::ToolCallEnd { .. }
+        | ResponseEvent::Start { .. }
+        | ResponseEvent::TextStart
+        | ResponseEvent::TextEnd
+        | ResponseEvent::ThinkingStart
+        | ResponseEvent::ThinkingEnd
+        | ResponseEvent::Usage(_)
+        | ResponseEvent::Done(_) => return Ok(None),
+    };
+    Ok(Some(server_interaction(message)))
+}
+
+pub fn thinking_completed(elapsed: Duration) -> pb::AgentServerMessage {
+    let milliseconds = elapsed.as_millis().clamp(1, i32::MAX as u128) as i32;
+    server_interaction(pb::interaction_update::Message::ThinkingCompleted(
+        pb::ThinkingCompletedUpdate {
+            thinking_duration_ms: milliseconds,
+        },
+    ))
+}
+
+pub fn arguments_delta(call: &ToolCall, delta: &str) -> Result<pb::AgentServerMessage> {
+    Ok(server_interaction(
+        pb::interaction_update::Message::PartialToolCall(pb::PartialToolCallUpdate {
+            call_id: call.call_id.clone(),
+            tool_call: Some(tool_placeholder(&call.name, &call.call_id)?),
+            args_text_delta: delta.into(),
+            model_call_id: call.model_call_id.clone(),
+        }),
+    ))
+}
+
+pub fn tool_started(call: &ToolCall) -> Result<pb::AgentServerMessage> {
+    Ok(server_interaction(
+        pb::interaction_update::Message::ToolCallStarted(pb::ToolCallStartedUpdate {
+            call_id: call.call_id.clone(),
+            tool_call: Some(render_tool_call(call, false)?),
+            model_call_id: call.model_call_id.clone(),
+        }),
+    ))
+}
+
+pub fn tool_completed(call: &ToolCall, completion: &ToolCompletion) -> pb::AgentServerMessage {
+    server_interaction(pb::interaction_update::Message::ToolCallCompleted(
+        pb::ToolCallCompletedUpdate {
+            call_id: call.call_id.clone(),
+            tool_call: Some(completion.tool_call().clone()),
+            model_call_id: call.model_call_id.clone(),
+        },
+    ))
+}
+
+pub fn turn_ended(usage: Usage) -> pb::AgentServerMessage {
+    server_interaction(pb::interaction_update::Message::TurnEnded(
+        pb::TurnEndedUpdate {
+            input_tokens: Some(usage.input_tokens as i64),
+            output_tokens: Some(usage.output_tokens as i64),
+            cache_read_tokens: Some(usage.cache_read_tokens as i64),
+            cache_write_tokens: Some(usage.cache_write_tokens as i64),
+            reasoning_tokens: Some(usage.reasoning_tokens as i64),
+        },
+    ))
+}
+
+pub fn tool_query(id: u32, call: &ToolCall) -> Result<pb::AgentServerMessage> {
+    use pb::interaction_query::Query;
+    let string = |name: &str| {
+        call.arguments
+            .get(name)
+            .and_then(Value::as_str)
+            .map(str::to_string)
+            .ok_or_else(|| Error::Protocol(format!("{} is missing {name}", call.name)))
+    };
+    let optional_string = |name: &str| {
+        call.arguments
+            .get(name)
+            .and_then(Value::as_str)
+            .map(str::to_string)
+    };
+    let query = match normalized(&call.name).as_str() {
+        "askquestion" => {
+            let questions = call
+                .arguments
+                .get("questions")
+                .and_then(Value::as_array)
+                .into_iter()
+                .flatten()
+                .map(|question| -> Result<_> {
+                    let required = |name: &str| {
+                        question
+                            .get(name)
+                            .and_then(Value::as_str)
+                            .map(str::to_string)
+                            .ok_or_else(|| Error::Protocol(format!("question is missing {name}")))
+                    };
+                    let options = question
+                        .get("options")
+                        .and_then(Value::as_array)
+                        .into_iter()
+                        .flatten()
+                        .map(|option| -> Result<_> {
+                            let value = |name: &str| {
+                                option
+                                    .get(name)
+                                    .and_then(Value::as_str)
+                                    .map(str::to_string)
+                                    .ok_or_else(|| {
+                                        Error::Protocol(format!(
+                                            "question option is missing {name}"
+                                        ))
+                                    })
+                            };
+                            Ok(pb::ask_question_args::Option {
+                                id: value("id")?,
+                                label: value("label")?,
+                            })
+                        })
+                        .collect::<Result<Vec<_>>>()?;
+                    Ok(pb::ask_question_args::Question {
+                        id: required("id")?,
+                        prompt: required("prompt")?,
+                        options,
+                        allow_multiple: question
+                            .get("allow_multiple")
+                            .and_then(Value::as_bool)
+                            .unwrap_or(false),
+                    })
+                })
+                .collect::<Result<Vec<_>>>()?;
+            Query::AskQuestionInteractionQuery(pb::AskQuestionInteractionQuery {
+                args: Some(pb::AskQuestionArgs {
+                    title: optional_string("title").unwrap_or_default(),
+                    questions,
+                    run_async: false,
+                    async_original_tool_call_id: String::new(),
+                }),
+                tool_call_id: call.call_id.clone(),
+            })
+        }
+        "websearch" => Query::WebSearchRequestQuery(pb::WebSearchRequestQuery {
+            args: Some(pb::WebSearchArgs {
+                search_term: string("search_term")?,
+                tool_call_id: call.call_id.clone(),
+            }),
+        }),
+        "webfetch" => Query::WebFetchRequestQuery(pb::WebFetchRequestQuery {
+            args: Some(pb::WebFetchArgs {
+                url: string("url")?,
+                tool_call_id: call.call_id.clone(),
+            }),
+            skip_approval: false,
+            smart_mode_approval: None,
+        }),
+        "switchmode" => Query::SwitchModeRequestQuery(pb::SwitchModeRequestQuery {
+            args: Some(pb::SwitchModeArgs {
+                target_mode_id: string("target_mode_id")?,
+                explanation: optional_string("explanation"),
+                tool_call_id: call.call_id.clone(),
+            }),
+        }),
+        "createplan" => {
+            let todos = call
+                .arguments
+                .get("todos")
+                .and_then(Value::as_array)
+                .into_iter()
+                .flatten()
+                .map(|todo| pb::TodoItem {
+                    id: todo
+                        .get("id")
+                        .and_then(Value::as_str)
+                        .unwrap_or_default()
+                        .into(),
+                    content: todo
+                        .get("content")
+                        .and_then(Value::as_str)
+                        .unwrap_or_default()
+                        .into(),
+                    status: pb::TodoStatus::Pending as i32,
+                    created_at: 0,
+                    updated_at: 0,
+                    dependencies: Vec::new(),
+                })
+                .collect();
+            Query::CreatePlanRequestQuery(pb::CreatePlanRequestQuery {
+                args: Some(pb::CreatePlanArgs {
+                    plan: string("plan")?,
+                    todos,
+                    overview: string("overview")?,
+                    name: string("name")?,
+                    is_project: false,
+                    phases: Vec::new(),
+                }),
+                tool_call_id: call.call_id.clone(),
+            })
+        }
+        "generateimage" => Query::GenerateImageRequestQuery(pb::GenerateImageRequestQuery {
+            args: Some(pb::GenerateImageArgs {
+                description: optional_string("description").unwrap_or_default(),
+                file_path: optional_string("file_path"),
+                reference_image_paths: Vec::new(),
+                aspect_ratio: optional_string("aspect_ratio"),
+            }),
+            tool_call_id: call.call_id.clone(),
+        }),
+        other => {
+            return Err(Error::Protocol(format!(
+                "tool {other} is not an InteractionQuery"
+            )))
+        }
+    };
+    Ok(pb::AgentServerMessage {
+        ttft_breakdown: None,
+        message: Some(pb::agent_server_message::Message::InteractionQuery(
+            pb::InteractionQuery {
+                id,
+                query: Some(query),
+            },
+        )),
+    })
+}
+
+pub fn server_interaction(message: pb::interaction_update::Message) -> pb::AgentServerMessage {
+    pb::AgentServerMessage {
+        ttft_breakdown: None,
+        message: Some(pb::agent_server_message::Message::InteractionUpdate(
+            pb::InteractionUpdate {
+                message: Some(message),
+            },
+        )),
+    }
+}
+
+pub fn tool_placeholder(name: &str, call_id: &str) -> Result<pb::ToolCall> {
+    use pb::tool_call::Tool;
+    let tool = match normalized(name).as_str() {
+        "shell" | "forcebackgroundshell" => Tool::ShellToolCall(pb::ShellToolCall::default()),
+        "delete" => Tool::DeleteToolCall(pb::DeleteToolCall::default()),
+        "glob" => Tool::GlobToolCall(pb::GlobToolCall::default()),
+        "grep" => Tool::GrepToolCall(pb::GrepToolCall::default()),
+        "read" => Tool::ReadToolCall(pb::ReadToolCall::default()),
+        "todowrite" => Tool::UpdateTodosToolCall(pb::UpdateTodosToolCall::default()),
+        "patchedit" | "write" => Tool::EditToolCall(pb::EditToolCall::default()),
+        "ls" => Tool::LsToolCall(pb::LsToolCall::default()),
+        "readlints" => Tool::ReadLintsToolCall(pb::ReadLintsToolCall::default()),
+        "callmcptool" => Tool::McpToolCall(pb::McpToolCall::default()),
+        "createplan" => Tool::CreatePlanToolCall(pb::CreatePlanToolCall::default()),
+        "websearch" => Tool::WebSearchToolCall(pb::WebSearchToolCall::default()),
+        "task" => Tool::TaskToolCall(pb::TaskToolCall::default()),
+        "fetchmcpresource" => Tool::ReadMcpResourceToolCall(pb::ReadMcpResourceToolCall::default()),
+        "askquestion" => Tool::AskQuestionToolCall(pb::AskQuestionToolCall::default()),
+        "webfetch" => Tool::WebFetchToolCall(pb::WebFetchToolCall::default()),
+        "switchmode" => Tool::SwitchModeToolCall(pb::SwitchModeToolCall::default()),
+        "generateimage" => Tool::GenerateImageToolCall(pb::GenerateImageToolCall::default()),
+        "communicateupdate" => {
+            Tool::CommunicateUpdateToolCall(pb::CommunicateUpdateToolCall::default())
+        }
+        "writeshellstdin" => Tool::WriteShellStdinToolCall(pb::WriteShellStdinToolCall::default()),
+        _ => return Err(Error::Protocol(format!("unsupported tool: {name}"))),
+    };
+    Ok(pb::ToolCall {
+        hook_additional_contexts: Vec::new(),
+        tool_call_id: Some(call_id.into()),
+        started_at_ms: None,
+        completed_at_ms: None,
+        tool: Some(tool),
+    })
+}
+
+pub fn render_tool_call(call: &ToolCall, completed: bool) -> Result<pb::ToolCall> {
+    let mut output = tool_placeholder(&call.name, &call.call_id)?;
+    let timestamp = now_ms();
+    output.started_at_ms = Some(timestamp);
+    if completed {
+        output.completed_at_ms = Some(timestamp);
+    }
+    let string = |name: &str| {
+        call.arguments
+            .get(name)
+            .and_then(Value::as_str)
+            .unwrap_or_default()
+            .to_string()
+    };
+    let optional = |name: &str| {
+        call.arguments
+            .get(name)
+            .and_then(Value::as_str)
+            .map(str::to_string)
+    };
+    match output.tool.as_mut() {
+        Some(pb::tool_call::Tool::ShellToolCall(tool)) => {
+            tool.args = Some(pb::ShellArgs {
+                command: string("command"),
+                working_directory: optional("working_directory").unwrap_or_default(),
+                tool_call_id: call.call_id.clone(),
+                ..Default::default()
+            })
+        }
+        Some(pb::tool_call::Tool::DeleteToolCall(tool)) => {
+            tool.args = Some(pb::DeleteArgs {
+                path: string("path"),
+                tool_call_id: call.call_id.clone(),
+            })
+        }
+        Some(pb::tool_call::Tool::GlobToolCall(tool)) => {
+            tool.args = Some(pb::GlobToolArgs {
+                target_directory: optional("target_directory"),
+                glob_pattern: string("glob_pattern"),
+            })
+        }
+        Some(pb::tool_call::Tool::GrepToolCall(tool)) => {
+            tool.args = Some(pb::GrepArgs {
+                pattern: string("pattern"),
+                path: optional("path"),
+                glob: optional("glob"),
+                output_mode: optional("output_mode"),
+                tool_call_id: call.call_id.clone(),
+                ..Default::default()
+            })
+        }
+        Some(pb::tool_call::Tool::ReadToolCall(tool)) => {
+            tool.args = Some(pb::ReadToolArgs {
+                path: string("path"),
+                offset: call
+                    .arguments
+                    .get("offset")
+                    .and_then(Value::as_i64)
+                    .map(|value| value as i32),
+                limit: call
+                    .arguments
+                    .get("limit")
+                    .and_then(Value::as_i64)
+                    .map(|value| value as i32),
+                include_line_numbers: call
+                    .arguments
+                    .get("include_line_numbers")
+                    .and_then(Value::as_bool),
+            })
+        }
+        Some(pb::tool_call::Tool::UpdateTodosToolCall(tool)) => {
+            tool.args = Some(pb::UpdateTodosArgs {
+                todos: tool_result::todo_items(&call.arguments),
+                merge: call
+                    .arguments
+                    .get("merge")
+                    .and_then(Value::as_bool)
+                    .unwrap_or(false),
+            })
+        }
+        Some(pb::tool_call::Tool::EditToolCall(tool)) => {
+            let stream_content = if normalized(&call.name) == "write" {
+                optional("contents").unwrap_or_default()
+            } else {
+                format!("{}\n---\n{}", string("old_string"), string("new_string"))
+            };
+            tool.args = Some(pb::EditArgs {
+                path: string("path"),
+                stream_content: Some(stream_content),
+            })
+        }
+        Some(pb::tool_call::Tool::LsToolCall(tool)) => {
+            tool.args = Some(pb::LsArgs {
+                path: string("path"),
+                tool_call_id: call.call_id.clone(),
+                ..Default::default()
+            })
+        }
+        Some(pb::tool_call::Tool::ReadLintsToolCall(tool)) => {
+            tool.args = Some(pb::ReadLintsToolArgs {
+                paths: call
+                    .arguments
+                    .get("paths")
+                    .and_then(Value::as_array)
+                    .into_iter()
+                    .flatten()
+                    .filter_map(Value::as_str)
+                    .map(str::to_string)
+                    .collect(),
+            })
+        }
+        Some(pb::tool_call::Tool::McpToolCall(tool)) => {
+            tool.args = Some(pb::McpArgs {
+                name: optional("toolName").unwrap_or_default(),
+                args: call
+                    .arguments
+                    .get("arguments")
+                    .and_then(Value::as_object)
+                    .map(super::exec::json_object_to_prost)
+                    .unwrap_or_default(),
+                tool_call_id: call.call_id.clone(),
+                tool_name: optional("toolName").unwrap_or_default(),
+                server_identifier: string("server"),
+                ..Default::default()
+            })
+        }
+        Some(pb::tool_call::Tool::CreatePlanToolCall(tool)) => {
+            tool.args = Some(pb::CreatePlanArgs {
+                plan: string("plan"),
+                todos: tool_result::todo_items(&call.arguments),
+                overview: string("overview"),
+                name: string("name"),
+                is_project: false,
+                phases: Vec::new(),
+            })
+        }
+        Some(pb::tool_call::Tool::WebSearchToolCall(tool)) => {
+            tool.args = Some(pb::WebSearchArgs {
+                search_term: string("search_term"),
+                tool_call_id: call.call_id.clone(),
+            })
+        }
+        Some(pb::tool_call::Tool::TaskToolCall(tool)) => {
+            tool.args = Some(pb::TaskArgs {
+                description: string("description"),
+                prompt: string("prompt"),
+                subagent_type: Some(subagent_type(&string("subagent_type"))),
+                model: optional("model"),
+                resume: optional("resume"),
+                agent_id: None,
+                attachments: Vec::new(),
+                mode: 0,
+                responding_to_message_ids: Vec::new(),
+                environment: 0,
+                machine: None,
+            })
+        }
+        Some(pb::tool_call::Tool::ReadMcpResourceToolCall(tool)) => {
+            tool.args = Some(pb::ReadMcpResourceExecArgs {
+                server: string("server"),
+                uri: string("uri"),
+                download_path: optional("downloadPath"),
+                tool_call_id: call.call_id.clone(),
+                smart_mode_approval: None,
+            })
+        }
+        Some(pb::tool_call::Tool::WebFetchToolCall(tool)) => {
+            tool.args = Some(pb::WebFetchArgs {
+                url: string("url"),
+                tool_call_id: call.call_id.clone(),
+            })
+        }
+        Some(pb::tool_call::Tool::SwitchModeToolCall(tool)) => {
+            tool.args = Some(pb::SwitchModeArgs {
+                target_mode_id: string("target_mode_id"),
+                explanation: optional("explanation"),
+                tool_call_id: call.call_id.clone(),
+            })
+        }
+        Some(pb::tool_call::Tool::GenerateImageToolCall(tool)) => {
+            tool.args = Some(pb::GenerateImageArgs {
+                description: string("description"),
+                file_path: optional("file_path"),
+                reference_image_paths: Vec::new(),
+                aspect_ratio: optional("aspect_ratio"),
+            })
+        }
+        Some(pb::tool_call::Tool::CommunicateUpdateToolCall(tool)) => {
+            tool.args = Some(pb::CommunicateUpdateArgs {
+                current_step: optional("current_step"),
+                final_summary: optional("final_summary"),
+                completed_subtitle: optional("completed_subtitle"),
+            })
+        }
+        Some(pb::tool_call::Tool::WriteShellStdinToolCall(tool)) => {
+            tool.args = Some(pb::WriteShellStdinArgs {
+                shell_id: call
+                    .arguments
+                    .get("shell_id")
+                    .and_then(Value::as_u64)
+                    .unwrap_or_default() as u32,
+                chars: string("chars"),
+            })
+        }
+        _ => {}
+    }
+    Ok(output)
+}
+
+fn subagent_type(name: &str) -> pb::SubagentType {
+    use pb::subagent_type::Type;
+    let r#type = match name.to_ascii_lowercase().as_str() {
+        "explore" => Type::Explore(pb::SubagentTypeExplore {}),
+        "browser-use" | "browseruse" => Type::BrowserUse(pb::SubagentTypeBrowserUse {}),
+        "shell" => Type::Shell(pb::SubagentTypeShell {}),
+        "bash" => Type::Bash(pb::SubagentTypeBash {}),
+        "debug" => Type::Debug(pb::SubagentTypeDebug {}),
+        "computer-use" | "computeruse" => Type::ComputerUse(pb::SubagentTypeComputerUse {}),
+        "" => Type::Unspecified(pb::SubagentTypeUnspecified {}),
+        custom => Type::Custom(pb::SubagentTypeCustom {
+            name: custom.into(),
+        }),
+    };
+    pb::SubagentType {
+        r#type: Some(r#type),
+    }
+}
+
+fn normalized(value: &str) -> String {
+    value
+        .chars()
+        .filter(|character| character.is_ascii_alphanumeric())
+        .flat_map(char::to_lowercase)
+        .collect()
+}
+
+fn now_ms() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as u64
+}

cursor-server/src/cursor/mod.rs

@@ -0,0 +1,12 @@
+pub mod bidi_append;
+pub mod blob_sync;
+pub mod checkpoint;
+pub mod connect;
+pub mod exec;
+pub mod handlers;
+pub mod interaction;
+pub mod pending;
+pub mod proto;
+pub mod run_sse;
+pub mod tool_result;
+pub mod tools;

cursor-server/src/cursor/pending.rs

@@ -0,0 +1,139 @@
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicU32, Ordering},
+        Arc,
+    },
+};
+
+use tokio::sync::Mutex;
+
+use crate::{model::ToolCall, Error, Result};
+
+#[derive(Clone, Default)]
+pub struct PendingExecRegistry {
+    next_id: Arc<AtomicU32>,
+    entries: Arc<Mutex<HashMap<u32, PendingExec>>>,
+}
+
+pub(crate) struct PendingExec {
+    pub call: ToolCall,
+    pub context: ExecContext,
+    pub started_at_ms: u64,
+    pub stdout: String,
+    pub stderr: String,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct ExecContext {
+    pub conversation_id: String,
+    pub terminals_folder: String,
+    pub admin_command_denylist: Vec<String>,
+}
+
+#[derive(Clone, Default)]
+pub struct PendingClientTools {
+    next_id: Arc<AtomicU32>,
+    calls: Arc<Mutex<HashMap<u32, PendingClientTool>>>,
+}
+
+pub(crate) struct PendingClientTool {
+    pub call: ToolCall,
+    pub context: ExecContext,
+    pub started_at_ms: u64,
+}
+
+impl PendingExecRegistry {
+    pub async fn reserve(&self, call: &ToolCall, context: &ExecContext) -> Result<u32> {
+        let id = next_id(&self.next_id)?;
+        self.entries.lock().await.insert(
+            id,
+            PendingExec {
+                call: call.clone(),
+                context: context.clone(),
+                started_at_ms: now_ms(),
+                stdout: String::new(),
+                stderr: String::new(),
+            },
+        );
+        Ok(id)
+    }
+
+    pub async fn call(&self, id: u32) -> Option<ToolCall> {
+        self.entries
+            .lock()
+            .await
+            .get(&id)
+            .map(|entry| entry.call.clone())
+    }
+
+    pub async fn append_stdout(&self, id: u32, data: &str) -> bool {
+        let mut entries = self.entries.lock().await;
+        let Some(entry) = entries.get_mut(&id) else {
+            return false;
+        };
+        entry.stdout.push_str(data);
+        true
+    }
+
+    pub async fn append_stderr(&self, id: u32, data: &str) -> bool {
+        let mut entries = self.entries.lock().await;
+        let Some(entry) = entries.get_mut(&id) else {
+            return false;
+        };
+        entry.stderr.push_str(data);
+        true
+    }
+
+    pub(crate) async fn take(&self, id: u32) -> Option<PendingExec> {
+        self.entries.lock().await.remove(&id)
+    }
+
+    pub async fn discard(&self, id: u32) {
+        self.entries.lock().await.remove(&id);
+    }
+
+    pub async fn drain_running(&self) -> Vec<u32> {
+        let mut entries = self.entries.lock().await;
+        let mut ids = entries.drain().map(|(id, _)| id).collect::<Vec<_>>();
+        ids.sort_unstable();
+        ids
+    }
+}
+
+impl PendingClientTools {
+    pub async fn reserve(&self, call: &ToolCall, context: &ExecContext) -> Result<u32> {
+        let id = next_id(&self.next_id)?;
+        self.calls.lock().await.insert(
+            id,
+            PendingClientTool {
+                call: call.clone(),
+                context: context.clone(),
+                started_at_ms: now_ms(),
+            },
+        );
+        Ok(id)
+    }
+
+    pub(crate) async fn take(&self, id: u32) -> Option<PendingClientTool> {
+        self.calls.lock().await.remove(&id)
+    }
+
+    pub async fn discard(&self, id: u32) {
+        self.calls.lock().await.remove(&id);
+    }
+}
+
+fn next_id(counter: &AtomicU32) -> Result<u32> {
+    counter
+        .fetch_add(1, Ordering::Relaxed)
+        .checked_add(1)
+        .ok_or_else(|| Error::Protocol("Cursor message id space exhausted".into()))
+}
+
+pub(crate) fn now_ms() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as u64
+}

cursor-server/src/cursor/proto.rs

@@ -0,0 +1,71 @@
+pub mod agent {
+    #[allow(clippy::large_enum_variant)]
+    pub mod v1 {
+        include!(concat!(env!("OUT_DIR"), "/agent.v1.rs"));
+    }
+}
+
+pub mod aiserver {
+    pub mod v1 {
+        #[derive(Clone, PartialEq, ::prost::Message)]
+        pub struct BidiRequestId {
+            #[prost(string, tag = "1")]
+            pub request_id: String,
+        }
+
+        #[derive(Clone, PartialEq, ::prost::Message)]
+        pub struct BidiAppendRequest {
+            #[prost(string, tag = "1")]
+            pub data: String,
+            #[prost(message, optional, tag = "2")]
+            pub request_id: Option<BidiRequestId>,
+            #[prost(int64, tag = "3")]
+            pub append_seqno: i64,
+            #[prost(bytes = "vec", tag = "4")]
+            pub data_binary: Vec<u8>,
+        }
+
+        #[derive(Clone, Copy, PartialEq, ::prost::Message)]
+        pub struct BidiAppendResponse {}
+
+        #[derive(Clone, PartialEq, ::prost::Message)]
+        pub struct CustomErrorDetails {
+            #[prost(string, tag = "1")]
+            pub title: String,
+            #[prost(string, tag = "2")]
+            pub detail: String,
+            #[prost(bool, optional, tag = "3")]
+            pub allow_command_links_potentially_unsafe_please_only_use_for_handwritten_trusted_markdown:
+                Option<bool>,
+            #[prost(bool, optional, tag = "4")]
+            pub is_retryable: Option<bool>,
+            #[prost(bool, optional, tag = "5")]
+            pub show_request_id: Option<bool>,
+            #[prost(bool, optional, tag = "6")]
+            pub should_show_immediate_error: Option<bool>,
+        }
+
+        #[derive(Clone, PartialEq, ::prost::Message)]
+        pub struct ErrorDetails {
+            #[prost(enumeration = "error_details::Error", tag = "1")]
+            pub error: i32,
+            #[prost(message, optional, tag = "2")]
+            pub details: Option<CustomErrorDetails>,
+            #[prost(bool, optional, tag = "3")]
+            pub is_expected: Option<bool>,
+        }
+
+        pub mod error_details {
+            #[derive(
+                Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration,
+            )]
+            #[repr(i32)]
+            pub enum Error {
+                Unspecified = 0,
+                CustomMessage = 29,
+                ProviderError = 57,
+                Internal = 59,
+            }
+        }
+    }
+}

cursor-server/src/cursor/run_sse.rs

@@ -0,0 +1,28 @@
+use axum::{
+    body::Body,
+    http::{header, HeaderValue, Response, StatusCode},
+};
+use bytes::Bytes;
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_stream::StreamExt;
+
+use crate::{run::RunRegistry, Result};
+
+pub async fn stream(registry: &RunRegistry, request_id: &str) -> Result<Response<Body>> {
+    let receiver = registry.get_or_create(request_id).await?.subscribe();
+    let body_stream =
+        UnboundedReceiverStream::new(receiver).map(Ok::<Bytes, std::convert::Infallible>);
+    let mut response = Response::new(Body::from_stream(body_stream));
+    *response.status_mut() = StatusCode::OK;
+    response.headers_mut().insert(
+        header::CONTENT_TYPE,
+        HeaderValue::from_static("text/event-stream"),
+    );
+    response
+        .headers_mut()
+        .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-cache"));
+    response
+        .headers_mut()
+        .insert("connect-protocol-version", HeaderValue::from_static("1"));
+    Ok(response)
+}

cursor-server/src/cursor/tools.rs

@@ -0,0 +1,318 @@
+use std::collections::{BTreeMap, HashSet};
+
+use crate::{
+    model::{CanonicalMessage, MessageContent, Origin, Role, ToolCall},
+    Error, Result,
+};
+
+use super::{
+    exec, interaction,
+    pending::{ExecContext, PendingClientTools, PendingExecRegistry},
+    proto::agent::v1 as pb,
+    tool_result::{self, ToolCompletion},
+};
+
+#[derive(Clone)]
+pub struct ToolDispatcher {
+    pending_execs: PendingExecRegistry,
+    pending_interactions: PendingClientTools,
+}
+
+pub struct DispatchedTool {
+    pub messages: Vec<pb::AgentServerMessage>,
+    pub completion: Option<ToolCompletion>,
+}
+
+pub enum ClientToolEvent {
+    Message(Box<pb::AgentServerMessage>),
+    Completed(Box<ToolCompletion>),
+}
+
+impl ToolDispatcher {
+    pub fn new(
+        pending_execs: PendingExecRegistry,
+        pending_interactions: PendingClientTools,
+    ) -> Self {
+        Self {
+            pending_execs,
+            pending_interactions,
+        }
+    }
+
+    pub async fn start_batch(
+        &self,
+        calls: &[ToolCall],
+        completed: &HashSet<String>,
+        messages: &[CanonicalMessage],
+        response_text: &str,
+        response_thinking: &str,
+        dynamic_mcp: &BTreeMap<String, pb::McpToolDefinition>,
+        context: &ExecContext,
+    ) -> Result<Vec<DispatchedTool>> {
+        let first_tool_index = current_turn_step_count(messages)
+            + usize::from(!response_thinking.is_empty())
+            + usize::from(!response_text.is_empty())
+            + 1;
+        let mut dispatched = Vec::with_capacity(calls.len() - completed.len().min(calls.len()));
+        for (position, call) in calls.iter().enumerate() {
+            if completed.contains(&call.call_id) {
+                continue;
+            }
+            dispatched.push(
+                self.start(call, first_tool_index + position, dynamic_mcp, context)
+                    .await?,
+            );
+        }
+        Ok(dispatched)
+    }
+
+    async fn start(
+        &self,
+        call: &ToolCall,
+        message_index: usize,
+        dynamic_mcp: &BTreeMap<String, pb::McpToolDefinition>,
+        context: &ExecContext,
+    ) -> Result<DispatchedTool> {
+        let mut messages = vec![interaction::tool_started(call)?];
+        let completion = if let Some(definition) = dynamic_mcp.get(&call.name) {
+            let id = self.pending_execs.reserve(call, context).await?;
+            messages.push(exec::mcp_request(id, call, definition)?);
+            None
+        } else {
+            match normalized(&call.name).as_str() {
+                "shell"
+                | "forcebackgroundshell"
+                | "read"
+                | "write"
+                | "delete"
+                | "grep"
+                | "glob"
+                | "ls"
+                | "readlints"
+                | "patchedit"
+                | "writeshellstdin"
+                | "task"
+                | "callmcptool"
+                | "fetchmcpresource" => {
+                    let id = self.pending_execs.reserve(call, context).await?;
+                    messages.push(exec::request(id, call, context)?);
+                    None
+                }
+                "askquestion" | "websearch" | "webfetch" | "switchmode" | "createplan"
+                | "generateimage" => {
+                    let id = self.pending_interactions.reserve(call, context).await?;
+                    messages.push(interaction::tool_query(id, call)?);
+                    None
+                }
+                "todowrite" | "communicateupdate" => Some(tool_result::local(call, message_index)?),
+                _ => return Err(Error::Protocol(format!("unsupported tool: {}", call.name))),
+            }
+        };
+        Ok(DispatchedTool {
+            messages,
+            completion,
+        })
+    }
+
+    pub async fn interaction_response(
+        &self,
+        response: &pb::InteractionResponse,
+    ) -> Result<ClientToolEvent> {
+        let pending = self
+            .pending_interactions
+            .take(response.id)
+            .await
+            .ok_or_else(|| {
+                Error::Protocol(format!("unknown InteractionResponse id: {}", response.id))
+            })?;
+        if normalized(&pending.call.name) == "webfetch"
+            && matches!(
+                response.result.as_ref(),
+                Some(pb::interaction_response::Result::WebFetchRequestResponse(
+                    pb::WebFetchRequestResponse {
+                        result: Some(pb::web_fetch_request_response::Result::Approved(_)),
+                    }
+                ))
+            )
+        {
+            let id = self
+                .pending_execs
+                .reserve(&pending.call, &pending.context)
+                .await?;
+            return Ok(ClientToolEvent::Message(Box::new(exec::request(
+                id,
+                &pending.call,
+                &pending.context,
+            )?)));
+        }
+        Ok(ClientToolEvent::Completed(Box::new(
+            tool_result::from_interaction(pending, response)?,
+        )))
+    }
+}
+
+fn current_turn_step_count(messages: &[CanonicalMessage]) -> usize {
+    let turn_start = messages
+        .iter()
+        .rposition(|message| message.role == Role::User && message.origin == Origin::User)
+        .map_or(0, |position| position + 1);
+    messages[turn_start..]
+        .iter()
+        .map(|message| match &message.content {
+            MessageContent::Assistant {
+                text,
+                thinking,
+                tool_calls,
+                ..
+            } => {
+                usize::from(!thinking.is_empty()) + usize::from(!text.is_empty()) + tool_calls.len()
+            }
+            _ => 0,
+        })
+        .sum()
+}
+
+fn normalized(name: &str) -> String {
+    name.chars()
+        .filter(|character| character.is_ascii_alphanumeric())
+        .flat_map(char::to_lowercase)
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::*;
+    use crate::model::{CanonicalMessage, Origin, Role};
+
+    fn call(name: &str) -> ToolCall {
+        ToolCall {
+            index: 0,
+            call_id: "call-1".into(),
+            model_call_id: "model-1".into(),
+            name: name.into(),
+            arguments_text: "{}".into(),
+            arguments: json!({}),
+        }
+    }
+
+    #[tokio::test]
+    async fn communicate_update_completes_locally_at_the_cursor_step_index() {
+        let dispatcher = ToolDispatcher::new(
+            PendingExecRegistry::default(),
+            PendingClientTools::default(),
+        );
+        let calls = [ToolCall {
+            arguments: json!({"current_step": "Reading"}),
+            ..call("CommunicateUpdate")
+        }];
+        let user = CanonicalMessage::text("user", Role::User, Origin::User, "go");
+        let dispatched = dispatcher
+            .start_batch(
+                &calls,
+                &HashSet::new(),
+                &[user],
+                "I will inspect it.",
+                "Need to read.",
+                &BTreeMap::new(),
+                &ExecContext::default(),
+            )
+            .await
+            .unwrap();
+        let completion = dispatched[0].completion.as_ref().unwrap();
+        let Some(pb::tool_call::Tool::CommunicateUpdateToolCall(tool)) =
+            completion.tool_call().tool.as_ref()
+        else {
+            panic!("expected CommunicateUpdateToolCall")
+        };
+        let Some(pb::communicate_update_result::Result::Success(success)) = tool
+            .result
+            .as_ref()
+            .and_then(|result| result.result.as_ref())
+        else {
+            panic!("expected communicate update success")
+        };
+        assert_eq!(success.message_index, 3);
+    }
+
+    #[tokio::test]
+    async fn approved_web_fetch_moves_from_interaction_to_exec() {
+        let dispatcher = ToolDispatcher::new(
+            PendingExecRegistry::default(),
+            PendingClientTools::default(),
+        );
+        let calls = [ToolCall {
+            arguments: json!({"url": "https://example.com"}),
+            ..call("WebFetch")
+        }];
+        let dispatched = dispatcher
+            .start_batch(
+                &calls,
+                &HashSet::new(),
+                &[],
+                "",
+                "",
+                &BTreeMap::new(),
+                &ExecContext::default(),
+            )
+            .await
+            .unwrap();
+        let Some(pb::agent_server_message::Message::InteractionQuery(query)) =
+            dispatched[0].messages[1].message.as_ref()
+        else {
+            panic!("expected WebFetch InteractionQuery")
+        };
+        let event = dispatcher
+            .interaction_response(&pb::InteractionResponse {
+                id: query.id,
+                result: Some(pb::interaction_response::Result::WebFetchRequestResponse(
+                    pb::WebFetchRequestResponse {
+                        result: Some(pb::web_fetch_request_response::Result::Approved(
+                            pb::web_fetch_request_response::Approved {},
+                        )),
+                    },
+                )),
+            })
+            .await
+            .unwrap();
+        let ClientToolEvent::Message(message) = event else {
+            panic!("approval must open the Exec phase")
+        };
+        let Some(pb::agent_server_message::Message::ExecServerMessage(exec)) = message.message
+        else {
+            panic!("expected FetchArgs")
+        };
+        assert!(matches!(
+            exec.message,
+            Some(pb::exec_server_message::Message::FetchArgs(_))
+        ));
+        let event = exec::client_event(
+            &pb::ExecClientMessage {
+                id: exec.id,
+                message: Some(pb::exec_client_message::Message::FetchResult(
+                    pb::FetchResult {
+                        result: Some(pb::fetch_result::Result::Success(pb::FetchSuccess {
+                            url: "https://example.com".into(),
+                            content: "hello".into(),
+                            status_code: 200,
+                            content_type: "text/html".into(),
+                        })),
+                    },
+                )),
+                ..Default::default()
+            },
+            &dispatcher.pending_execs,
+        )
+        .await
+        .unwrap();
+        let exec::ClientExecEvent::Completed(completion) = event else {
+            panic!("FetchResult must complete WebFetch")
+        };
+        assert_eq!(completion.result().output, json!("hello"));
+        assert!(matches!(
+            completion.tool_call().tool,
+            Some(pb::tool_call::Tool::WebFetchToolCall(_))
+        ));
+    }
+}