diff --git a/server/src/cursor/conversation/output.rs b/server/src/cursor/conversation/output.rs index 349fccc..55dd9b6 100644 --- a/server/src/cursor/conversation/output.rs +++ b/server/src/cursor/conversation/output.rs @@ -158,6 +158,7 @@ impl ConversationOutput { let mut streams = BTreeMap::::new(); let mut completions = HashMap::::new(); let mut completed = HashSet::::new(); + let mut completed_round = None::; let mut response_text = String::new(); let mut response_thinking = String::new(); let mut active_round = None::; @@ -425,6 +426,16 @@ impl ConversationOutput { round_id, calls: round_calls, } => { + // `completed` exists so that replaying ExecuteToolRound for a round + // does not dispatch a call this round already committed. Tool call ids + // are only unique *within* a round -- the schema says as much with + // `UNIQUE (round_id, call_id)` -- so an id retained from an earlier + // round would make start_batch skip a fresh call, and tool_round wait + // forever for a result nothing will ever produce. + if completed_round.as_ref() != Some(&round_id) { + completed.clear(); + completed_round = Some(round_id.clone()); + } active_round = Some(round_id.clone()); active_tool_calls = round_calls .iter() diff --git a/server/tests/tool_round.rs b/server/tests/tool_round.rs index 309ed1d..bfea8f8 100644 --- a/server/tests/tool_round.rs +++ b/server/tests/tool_round.rs @@ -1253,6 +1253,151 @@ fn client_run_for_model( } } +/// A provider that reuses a tool call id across two rounds of the same run +/// must not wedge the run. `ToolDispatcher::start_batch` skips any call whose +/// id is already in `ToolBatchState::completed`, and that set is never cleared +/// for the lifetime of the run, so the skipped call produced no completion and +/// `tool_round::execute` waited forever for a result that could never arrive. +#[tokio::test] +async fn duplicate_tool_call_id_across_rounds_does_not_wedge_the_run() { + let (_directory, store) = fixtures::temp_store().await; + let provider = fake_provider::FakeProvider::default(); + for _ in 0..2 { + provider.push(vec![ + ModelEvent::Start { + model_call_id: "ignored".into(), + }, + ModelEvent::ToolCallStart { + index: 0, + call_id: "call-1".into(), + name: "Read".into(), + }, + ModelEvent::ToolCallArgumentsDelta { + index: 0, + delta: "{\"path\":\"/tmp/a\"}".into(), + }, + ModelEvent::ToolCallEnd { index: 0 }, + ModelEvent::Done(FinishReason::ToolUse), + ]); + } + provider.push(vec![ + ModelEvent::Start { + model_call_id: "ignored".into(), + }, + ModelEvent::TextStart, + ModelEvent::TextDelta("done".into()), + ModelEvent::TextEnd, + ModelEvent::Done(FinishReason::Stop), + ]); + let assets = PromptAssets::load( + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("prompt/cursor") + .as_path(), + ) + .unwrap(); + let registry = TransportRegistry::new( + store.clone(), + Arc::new(provider.clone()), + PromptCompiler::new(assets), + ); + let handle = registry.get_or_create("tool-request").await.unwrap(); + let mut output = handle.subscribe(); + handle + .command(TransportCommand::Append { + seqno: 0, + message: Box::new(client_run()), + }) + .await + .unwrap(); + let mut seqno = 1; + let mut execs = 0; + loop { + let frame = tokio::time::timeout(std::time::Duration::from_secs(10), output.recv()) + .await + .expect("run must not hang on a reused tool call id") + .unwrap(); + let (flags, payload) = connect::decode_frames(&frame).unwrap().pop().unwrap(); + if flags & connect::END_STREAM_FLAG != 0 { + break; + } + let server = pb::AgentServerMessage::decode(payload).unwrap(); + match server.message { + Some(pb::agent_server_message::Message::KvServerMessage(kv)) => { + handle + .command(TransportCommand::Append { + seqno, + message: Box::new(kv_ack(kv.id)), + }) + .await + .unwrap(); + seqno += 1; + } + Some(pb::agent_server_message::Message::ExecServerMessage(exec)) => { + execs += 1; + let exec_id = exec.id; + handle + .command(TransportCommand::Append { + seqno, + message: Box::new(pb::AgentClientMessage { + message: Some(pb::agent_client_message::Message::ExecClientMessage( + pb::ExecClientMessage { + id: exec_id, + exec_id: String::new(), + message: Some(pb::exec_client_message::Message::ReadResult( + pb::ReadResult { + result: Some(pb::read_result::Result::Success( + pb::ReadSuccess { + path: "/tmp/a".into(), + total_lines: 1, + file_size: 1, + output: Some( + pb::read_success::Output::Content( + "x".into(), + ), + ), + ..Default::default() + }, + )), + }, + )), + ..Default::default() + }, + )), + }), + }) + .await + .unwrap(); + seqno += 1; + handle + .command(TransportCommand::Append { + seqno, + message: Box::new(pb::AgentClientMessage { + message: Some( + pb::agent_client_message::Message::ExecClientControlMessage( + pb::ExecClientControlMessage { + message: Some( + pb::exec_client_control_message::Message::StreamClose( + pb::ExecClientStreamClose { id: exec_id }, + ), + ), + }, + ), + ), + }), + }) + .await + .unwrap(); + seqno += 1; + } + _ => {} + } + } + // Both rounds have to reach the client, and the run has to get far enough + // to ask the provider a third time and finish. + assert_eq!(execs, 2); + assert_eq!(provider.requests().len(), 3); +} + fn client_run() -> pb::AgentClientMessage { let user = pb::UserMessage { text: "read it".into(),