refactor: update cursor request handling and improve parent request management

- Changed `run_id` to `request_id` in `CursorParent` struct for clarity.
- Enhanced the `prepare` function to handle parent requests asynchronously, ensuring proper error handling for active runs.
- Updated database interactions to include `cursor_request_id` for better tracking of requests.
- Added tests to verify the behavior of reused cursor request IDs and their mapping to distinct executions.
This commit is contained in:
leookun
2026-08-26 01:46:30 +08:00
parent ae31757635
commit 847e92c7ea
20 changed files with 271 additions and 57 deletions
+35 -17
View File
@@ -89,24 +89,42 @@ impl CursorActor {
.map(|parent| parent.tool_call_id.clone()),
request.conversation_state.clone(),
);
let parent = handle.parent().map(|parent| {
(
crate::model::RunId::new(&parent.run_id),
parent.tool_call_id.clone(),
let prepared = async {
let parent = match handle.parent() {
Some(parent) => {
let parent_run_id = dependencies
.store
.active_run_for_cursor_request(
&parent.request_id,
)
.await?
.ok_or_else(|| {
crate::Error::Protocol(format!(
"Cursor parent request {} has no active local Run",
parent.request_id
))
})?;
Some((
parent_run_id,
parent.tool_call_id.clone(),
))
}
None => None,
};
request::prepare(
handle.request_id(),
&request,
parent,
request::PrepareDependencies {
compiler: &dependencies.compiler,
store: &dependencies.store,
checkpoint: &checkpoint,
blob_sync: &blob_sync,
context_sync: &context_sync,
},
)
});
let prepared = request::prepare(
handle.request_id(),
&request,
parent,
request::PrepareDependencies {
compiler: &dependencies.compiler,
store: &dependencies.store,
checkpoint: &checkpoint,
blob_sync: &blob_sync,
context_sync: &context_sync,
},
)
.await
}
.await;
let (prepared, context) = match prepared {
Ok(prepared) => prepared,
+5 -5
View File
@@ -203,12 +203,12 @@ async fn buffered(request: Request<Body>) -> Result<(axum::http::request::Parts,
}
fn parent_headers(headers: &HeaderMap) -> Result<Option<CursorParent>> {
let run_id = header_text(headers, "x-parent-request-id")?;
let request_id = header_text(headers, "x-parent-request-id")?;
let tool_call_id = header_text(headers, "x-parent-agent-tool-call-id")?;
match (run_id, tool_call_id) {
match (request_id, tool_call_id) {
(None, None) => Ok(None),
(Some(run_id), Some(tool_call_id)) => Ok(Some(CursorParent {
run_id: run_id.into(),
(Some(request_id), Some(tool_call_id)) => Ok(Some(CursorParent {
request_id: request_id.into(),
tool_call_id: tool_call_id.into(),
})),
_ => Err(crate::Error::Protocol(
@@ -270,7 +270,7 @@ mod tests {
assert_eq!(
parent_headers(&headers).unwrap(),
Some(CursorParent {
run_id: "parent-run".into(),
request_id: "parent-run".into(),
tool_call_id: "parent-call".into(),
})
);
+98 -16
View File
@@ -1,5 +1,7 @@
use std::collections::BTreeMap;
use uuid::Uuid;
use crate::{
cursor::prompting::{Mode, PromptCompiler},
cursor::{
@@ -71,9 +73,7 @@ pub(crate) async fn prepare(
.clone()
.unwrap_or_else(|| request_id.into()),
);
// RunSSE/Bidi request_id identifies this concrete execution attempt. Cursor may
// reuse AgentRunRequest.run_id when a queued or subagent-driven attempt resumes.
let run_id = RunId::new(request_id);
let run_id = execution_run_id(request_id);
let mut base_messages = if request.conversation_state.is_some() {
Some(
checkpoint
@@ -128,7 +128,7 @@ pub(crate) async fn prepare(
starts_turn,
compacting,
background_completion,
} = action(request_id, request)?;
} = action(request)?;
let checkpoint_mode = if request.subagent_type_name.is_some() {
Mode::Subagent
} else {
@@ -286,6 +286,7 @@ pub(crate) async fn prepare(
Ok((
PreparedRun {
run_id,
cursor_request_id: Some(request_id.into()),
conversation_id,
kind,
model,
@@ -348,12 +349,17 @@ fn validate_prompt_root(messages: &[CanonicalMessage]) -> Result<()> {
Ok(())
}
fn action(request_id: &str, request: &pb::AgentRunRequest) -> Result<ActionProjection> {
let mode = request
fn execution_run_id(request_id: &str) -> RunId {
let execution_id = Uuid::new_v4().simple().to_string();
RunId::new(format!("{request_id}:{}", &execution_id[..8]))
}
fn action(request: &pb::AgentRunRequest) -> Result<ActionProjection> {
let conversation_mode = request
.conversation_state
.as_ref()
.and_then(|state| state.mode)
.unwrap_or(pb::AgentMode::Agent as i32);
.and_then(|state| state.mode);
let mode = conversation_mode.unwrap_or(pb::AgentMode::Agent as i32);
let Some(action) = request
.action
.as_ref()
@@ -375,6 +381,11 @@ fn action(request_id: &str, request: &pb::AgentRunRequest) -> Result<ActionProje
let user = action.user_message.as_ref().ok_or_else(|| {
Error::Protocol("Cursor user message action has no UserMessage".into())
})?;
let mode = if user.mode == pb::AgentMode::Unspecified as i32 {
conversation_mode.unwrap_or(user.mode)
} else {
user.mode
};
if user.message_id.is_empty() {
return Err(Error::Protocol(
"Cursor user message action has no message_id".into(),
@@ -382,7 +393,7 @@ fn action(request_id: &str, request: &pb::AgentRunRequest) -> Result<ActionProje
}
if user.text.trim() == "/summarize" {
return Ok(ActionProjection {
mode: user.mode,
mode,
turn_user: Some(user.clone()),
action_context: String::new(),
event_id: None,
@@ -405,12 +416,13 @@ fn action(request_id: &str, request: &pb::AgentRunRequest) -> Result<ActionProje
.filter(|text| !text.is_empty())
.cloned(),
);
let event_id = format!("cursor:user:{}", user.message_id);
Ok(ActionProjection {
mode: user.mode,
mode,
turn_user: Some(user.clone()),
action_context: context.join("\n\n"),
event_id: Some(format!("run-request:{request_id}")),
input_id: Some(format!("cursor:user:{}", user.message_id)),
event_id: Some(event_id.clone()),
input_id: Some(event_id),
starts_turn: true,
compacting: false,
background_completion: false,
@@ -418,10 +430,11 @@ fn action(request_id: &str, request: &pb::AgentRunRequest) -> Result<ActionProje
}
pb::conversation_action::Action::BackgroundTaskCompletionAction(action) => {
let projection = background::project(action, mode)?;
let event_id = projection.turn_user.message_id.clone();
Ok(ActionProjection {
mode,
action_context: projection.context,
event_id: Some(format!("run-request:{request_id}")),
event_id: Some(event_id),
input_id: None,
turn_user: Some(projection.turn_user),
starts_turn: true,
@@ -598,6 +611,18 @@ mod tests {
));
}
#[test]
fn execution_run_id_keeps_the_request_id_and_adds_eight_uuid_hex_digits() {
let run_id = execution_run_id("01bba7c5-9c00-4922-b1df-1f58146b5d90");
let suffix = run_id
.as_str()
.strip_prefix("01bba7c5-9c00-4922-b1df-1f58146b5d90:")
.unwrap();
assert_eq!(suffix.len(), 8);
assert!(suffix.bytes().all(|byte| byte.is_ascii_hexdigit()));
}
#[test]
fn current_user_message_consumes_the_mode_instead_of_history_mode() {
let request = pb::AgentRunRequest {
@@ -621,7 +646,7 @@ mod tests {
}),
..Default::default()
};
let projection = action("request", &request).unwrap();
let projection = action(&request).unwrap();
assert_eq!(projection.mode, pb::AgentMode::Ask as i32);
assert_eq!(
projection.input_id.as_deref(),
@@ -630,6 +655,63 @@ mod tests {
assert_eq!(mode_from_proto(projection.mode).unwrap(), Mode::Ask);
}
#[test]
fn queued_user_message_without_mode_inherits_conversation_mode() {
let request = pb::AgentRunRequest {
conversation_state: Some(pb::ConversationStateStructure {
mode: Some(pb::AgentMode::Agent as i32),
..Default::default()
}),
action: Some(pb::ConversationAction {
action: Some(pb::conversation_action::Action::UserMessageAction(
pb::UserMessageAction {
user_message: Some(pb::UserMessage {
text: "queued follow-up".into(),
message_id: "queued-user-message".into(),
..Default::default()
}),
..Default::default()
},
)),
..Default::default()
}),
..Default::default()
};
let projection = action(&request).unwrap();
assert_eq!(projection.mode, pb::AgentMode::Agent as i32);
assert_eq!(mode_from_proto(projection.mode).unwrap(), Mode::Agent);
}
#[test]
fn queued_messages_reusing_a_request_id_keep_distinct_runtime_identities() {
let request = |message_id: &str| pb::AgentRunRequest {
action: Some(pb::ConversationAction {
action: Some(pb::conversation_action::Action::UserMessageAction(
pb::UserMessageAction {
user_message: Some(pb::UserMessage {
text: "queued follow-up".into(),
message_id: message_id.into(),
mode: pb::AgentMode::Agent as i32,
..Default::default()
}),
..Default::default()
},
)),
..Default::default()
}),
..Default::default()
};
let first = action(&request("message-one")).unwrap();
let second = action(&request("message-two")).unwrap();
assert_eq!(first.event_id.as_deref(), Some("cursor:user:message-one"));
assert_eq!(second.event_id.as_deref(), Some("cursor:user:message-two"));
assert_ne!(first.event_id, second.event_id);
}
#[test]
fn execute_plan_appends_the_approved_plan_as_a_stable_runtime_event() {
let execute = pb::ExecutePlanAction {
@@ -648,8 +730,8 @@ mod tests {
..Default::default()
};
let first = action("request-one", &request).unwrap();
let second = action("request-two", &request).unwrap();
let first = action(&request).unwrap();
let second = action(&request).unwrap();
assert_eq!(first.mode, pb::AgentMode::Agent as i32);
assert!(first.starts_turn);
assert_eq!(first.event_id, second.event_id);
+3 -3
View File
@@ -35,7 +35,7 @@ pub struct CursorSessionHandle {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CursorParent {
pub run_id: String,
pub request_id: String,
pub tool_call_id: String,
}
@@ -69,9 +69,9 @@ impl CursorSessionHandle {
self.cancellation.clone()
}
pub fn set_parent(&self, parent: CursorParent) -> Result<()> {
if parent.run_id.is_empty() || parent.tool_call_id.is_empty() {
if parent.request_id.is_empty() || parent.tool_call_id.is_empty() {
return Err(crate::Error::Protocol(
"Cursor parent run and tool call ids are required".into(),
"Cursor parent request and tool call ids are required".into(),
));
}
if self.parent.get().is_some_and(|current| current != &parent) {
+1
View File
@@ -48,6 +48,7 @@ pub struct RecoveredToolRound {
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct PreparedRun {
pub run_id: RunId,
pub cursor_request_id: Option<String>,
pub conversation_id: ConversationId,
pub kind: RunKind,
pub model: ModelSpec,
+19 -2
View File
@@ -88,12 +88,13 @@ impl Store {
run_kind_columns(&prepared.kind);
sqlx::query(
"INSERT INTO runs
(run_id, conversation_id, base_revision_id, head_revision_id,
(run_id, cursor_request_id, conversation_id, base_revision_id, head_revision_id,
parent_run_id, parent_tool_call_id, run_kind, subagent_kind,
status, created_at_ms, updated_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'running', ?, ?)",
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'running', ?, ?)",
)
.bind(prepared.run_id.as_str())
.bind(prepared.cursor_request_id.as_deref())
.bind(prepared.conversation_id.as_str())
.bind(prepared.base_revision_id.0)
.bind(prepared.base_revision_id.0)
@@ -128,6 +129,22 @@ impl Store {
})
}
pub async fn active_run_for_cursor_request(
&self,
cursor_request_id: &str,
) -> Result<Option<RunId>> {
let run_id: Option<String> = sqlx::query_scalar(
"SELECT run_id FROM runs
WHERE cursor_request_id = ? AND status = 'running'
ORDER BY created_at_ms DESC
LIMIT 1",
)
.bind(cursor_request_id)
.fetch_optional(&self.pool)
.await?;
Ok(run_id.map(RunId))
}
pub async fn begin_provider_call(&self, run_id: &RunId) -> Result<u64> {
let index: Option<i64> = sqlx::query_scalar(
"UPDATE runs SET provider_call_index = provider_call_index + 1, updated_at_ms = ?