From a42f84cfe78b03b94fa07e9caa04a8588c67f4b3 Mon Sep 17 00:00:00 2001 From: kevin9327 Date: Mon, 31 Aug 2026 20:07:52 +0900 Subject: [PATCH] fix(conversation): scope completed tool call ids to their round A provider that reuses a tool call id across two rounds of one run wedges the run permanently. `ToolDispatcher::start_batch` skips any call whose id is in `ToolBatchState::completed`, so the second call is never dispatched and never produces a `ToolCompletion`, while `tool_round::execute` blocks waiting for `calls.len()` results with no timeout on that path. The client sees the tool call appear and then nothing: no completion, no further output, no end-stream frame. The `completed` set is built once per run and never cleared, so it is run-scoped. A tool call id is only unique within a round, which the schema already states as `UNIQUE (round_id, call_id)`; the sibling runtime completed-map is likewise already cleared per round at output.rs:615. Clear `completed` when `ExecuteToolRound` begins a new round, tracked independently of `active_round` so it does not depend on the order in which ToolRoundStarted and ExecuteToolRound are observed. Replaying a round still skips the calls that round already committed. The practical trigger is openai_chat.rs:196, which synthesizes `call-{index}` from a per-stream index when a provider omits tool call ids, so `call-0` recurs on every model call. That file is left alone here. Co-Authored-By: Claude Opus 4.8 --- server/src/cursor/conversation/output.rs | 11 ++ server/tests/tool_round.rs | 145 +++++++++++++++++++++++ 2 files changed, 156 insertions(+) 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(),