From 20a7bb4db20314f48d5fb352baa1e54374945dfb Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 2 Sep 2026 12:53:41 +0100 Subject: [PATCH 1/4] fix(acp): preserve interrupted responses --- Cargo.lock | 5 +-- Cargo.toml | 4 +- src/protocols/acp/v2.rs | 90 +++++++++++++++-------------------------- 3 files changed, 37 insertions(+), 62 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 79840d79..5dabe761 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,7 +134,7 @@ dependencies = [ [[package]] name = "agentkit-acp" version = "0.10.11" -source = "git+https://github.com/danielkov/agentkit.git?rev=64244da92cfe983d7da8bf5ecac20b54bce4677a#64244da92cfe983d7da8bf5ecac20b54bce4677a" +source = "git+https://github.com/danielkov/agentkit.git?rev=0199d04c4d92bf858f4f4895b9f835c755201ea5#0199d04c4d92bf858f4f4895b9f835c755201ea5" dependencies = [ "agent-client-protocol", "agentkit-core", @@ -241,8 +241,7 @@ dependencies = [ [[package]] name = "agentkit-loop" version = "0.10.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f9deddd6431bb2368aebd5eb51f3b43256c8347bfde9769ff726804dbc86355" +source = "git+https://github.com/danielkov/agentkit.git?rev=0199d04c4d92bf858f4f4895b9f835c755201ea5#0199d04c4d92bf858f4f4895b9f835c755201ea5" dependencies = [ "agentkit-core", "agentkit-task-manager", diff --git a/Cargo.toml b/Cargo.toml index debc0215..24b7491a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -88,13 +88,13 @@ tempfile = "=3.27.0" tokio = { version = "=1.53.1", features = ["test-util"] } [patch.crates-io] -agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "64244da92cfe983d7da8bf5ecac20b54bce4677a" } +agentkit-loop = { git = "https://github.com/danielkov/agentkit.git", rev = "0199d04c4d92bf858f4f4895b9f835c755201ea5" } +agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "0199d04c4d92bf858f4f4895b9f835c755201ea5" } agent-client-protocol = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" } agent-client-protocol-http = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" } [patch."https://github.com/danielkov/agentkit.git"] agentkit-capabilities = "=0.10.5" agentkit-core = "=0.10.5" -agentkit-loop = "=0.10.11" agentkit-task-manager = "=0.10.7" agentkit-tools-core = "=0.10.5" diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index aba923ff..ee2e944f 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -373,17 +373,12 @@ where } return; } + // A cancelled turn leaves its last streamed attempt visible. Only an explicit + // supersession event proves that the current attempt is stale and must be cleared. if matches!(&event.event, AgentEvent::ResponseAttemptSuperseded) { self.clear_current(); return; } - if matches!( - &event.event, - AgentEvent::TurnFinished(result) - if result.finish_reason == FinishReason::Cancelled - ) { - self.clear_current(); - } if matches!( &event.event, AgentEvent::TurnStarted { .. } @@ -2256,7 +2251,7 @@ mod tests { } #[test] - fn response_replacement_clears_streamed_messages_when_turn_is_cancelled() { + fn response_replacement_preserves_streamed_messages_when_turn_is_cancelled() { let integration = AcpIntegration::default(); let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); @@ -2305,26 +2300,16 @@ mod tests { emit(finish("turn-1", FinishReason::Cancelled)); let updates = recording.updates.lock().unwrap(); - assert_eq!(updates.len(), 4); - let agent_id = match &updates[0].update { - wire::SessionUpdate::AgentMessageChunk(chunk) => chunk.message_id.clone(), - update => panic!("expected agent message chunk, got {update:?}"), - }; - let thought_id = match &updates[1].update { - wire::SessionUpdate::AgentThoughtChunk(chunk) => chunk.message_id.clone(), - update => panic!("expected agent thought chunk, got {update:?}"), - }; + assert_eq!(updates.len(), 2); assert!(matches!( - &updates[2].update, - wire::SessionUpdate::AgentMessage(message) - if message.message_id == agent_id - && message.content.value().is_some_and(Vec::is_empty) + &updates[0].update, + wire::SessionUpdate::AgentMessageChunk(chunk) + if chunk.content == wire::ContentBlock::Text(wire::TextContent::new("answer")) )); assert!(matches!( - &updates[3].update, - wire::SessionUpdate::AgentThought(message) - if message.message_id == thought_id - && message.content.value().is_some_and(Vec::is_empty) + &updates[1].update, + wire::SessionUpdate::AgentThoughtChunk(chunk) + if chunk.content == wire::ContentBlock::Text(wire::TextContent::new("thinking")) )); drop(updates); @@ -2334,7 +2319,7 @@ mod tests { }); emit_part("message-2", agentkit_core::PartKind::Text, "completed"); emit(finish("turn-2", FinishReason::Completed)); - assert_eq!(recording.updates.lock().unwrap().len(), 5); + assert_eq!(recording.updates.lock().unwrap().len(), 3); emit(AgentEvent::TurnStarted { session_id: loop_session_id.clone(), @@ -2349,16 +2334,12 @@ mod tests { emit(finish("turn-3", FinishReason::Cancelled)); let updates = recording.updates.lock().unwrap(); - assert_eq!(updates.len(), 7); - let replacement_id = match &updates[5].update { - wire::SessionUpdate::AgentMessageChunk(chunk) => chunk.message_id.clone(), - update => panic!("expected replacement message chunk, got {update:?}"), - }; + assert_eq!(updates.len(), 4); assert!(matches!( - &updates[6].update, - wire::SessionUpdate::AgentMessage(message) - if message.message_id == replacement_id - && message.content.value().is_some_and(Vec::is_empty) + &updates[3].update, + wire::SessionUpdate::AgentMessageChunk(chunk) + if chunk.content + == wire::ContentBlock::Text(wire::TextContent::new("partial replacement")) )); } @@ -2385,40 +2366,40 @@ mod tests { events: VecDeque, } - struct MarkerCancellationAdapter { + struct StreamingCancellationAdapter { interrupt: AcpSessionHandle, } - struct MarkerCancellationSession { + struct StreamingCancellationSession { interrupt: AcpSessionHandle, } - struct MarkerCancellationTurn { + struct StreamingCancellationTurn { interrupt: AcpSessionHandle, next: u8, } #[async_trait] - impl ModelAdapter for MarkerCancellationAdapter { - type Session = MarkerCancellationSession; + impl ModelAdapter for StreamingCancellationAdapter { + type Session = StreamingCancellationSession; async fn start_session(&self, _config: SessionConfig) -> Result { - Ok(MarkerCancellationSession { + Ok(StreamingCancellationSession { interrupt: self.interrupt.clone(), }) } } #[async_trait] - impl ModelSession for MarkerCancellationSession { - type Turn = MarkerCancellationTurn; + impl ModelSession for StreamingCancellationSession { + type Turn = StreamingCancellationTurn; async fn begin_turn( &mut self, _request: TurnRequest, _cancellation: Option, ) -> Result { - Ok(MarkerCancellationTurn { + Ok(StreamingCancellationTurn { interrupt: self.interrupt.clone(), next: 0, }) @@ -2426,7 +2407,7 @@ mod tests { } #[async_trait] - impl ModelTurn for MarkerCancellationTurn { + impl ModelTurn for StreamingCancellationTurn { async fn next_event( &mut self, _cancellation: Option, @@ -2442,7 +2423,7 @@ mod tests { }), 2 => { self.interrupt.interrupt(); - ModelTurnEvent::ResponseAttemptSuperseded + return Err(LoopError::Cancelled); } _ => return Ok(None), }; @@ -2594,7 +2575,7 @@ mod tests { } #[tokio::test] - async fn loop_driver_cancellation_clears_streamed_message_when_marker_is_suppressed() { + async fn loop_driver_cancellation_preserves_streamed_message() { let integration = AcpIntegration::default(); let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); @@ -2609,7 +2590,7 @@ mod tests { .unwrap(); let observer = ResponseReplacementObserver::new(integration, sink, session_id); let mut driver = Agent::builder() - .model(MarkerCancellationAdapter { + .model(StreamingCancellationAdapter { interrupt: handle.clone(), }) .observer(observer) @@ -2629,16 +2610,11 @@ mod tests { assert_eq!(result.finish_reason, FinishReason::Cancelled); let updates = recording.updates.lock().unwrap(); - assert_eq!(updates.len(), 2); - let message_id = match &updates[0].update { - wire::SessionUpdate::AgentMessageChunk(chunk) => chunk.message_id.clone(), - update => panic!("expected partial message chunk, got {update:?}"), - }; + assert_eq!(updates.len(), 1); assert!(matches!( - &updates[1].update, - wire::SessionUpdate::AgentMessage(message) - if message.message_id == message_id - && message.content.value().is_some_and(Vec::is_empty) + &updates[0].update, + wire::SessionUpdate::AgentMessageChunk(chunk) + if chunk.content == wire::ContentBlock::Text(wire::TextContent::new("partial answer")) )); } From d072e7a82d4e76ba648face59398e68804f0339d Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 2 Sep 2026 13:49:19 +0100 Subject: [PATCH 2/4] chore: trigger review webhook From c3cb2c1714702ef4cd2ae5ccd2439e374307cd27 Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 2 Sep 2026 14:15:20 +0100 Subject: [PATCH 3/4] chore(deps): refresh tracked AgentKit refs --- Cargo.lock | 4 ++-- Cargo.toml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5dabe761..021231a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,7 +134,7 @@ dependencies = [ [[package]] name = "agentkit-acp" version = "0.10.11" -source = "git+https://github.com/danielkov/agentkit.git?rev=0199d04c4d92bf858f4f4895b9f835c755201ea5#0199d04c4d92bf858f4f4895b9f835c755201ea5" +source = "git+https://github.com/danielkov/agentkit.git?rev=6480d909800f25a5ddf042cf3f6791bf6726fe41#6480d909800f25a5ddf042cf3f6791bf6726fe41" dependencies = [ "agent-client-protocol", "agentkit-core", @@ -241,7 +241,7 @@ dependencies = [ [[package]] name = "agentkit-loop" version = "0.10.11" -source = "git+https://github.com/danielkov/agentkit.git?rev=0199d04c4d92bf858f4f4895b9f835c755201ea5#0199d04c4d92bf858f4f4895b9f835c755201ea5" +source = "git+https://github.com/danielkov/agentkit.git?rev=6480d909800f25a5ddf042cf3f6791bf6726fe41#6480d909800f25a5ddf042cf3f6791bf6726fe41" dependencies = [ "agentkit-core", "agentkit-task-manager", diff --git a/Cargo.toml b/Cargo.toml index 24b7491a..266aa526 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -88,8 +88,8 @@ tempfile = "=3.27.0" tokio = { version = "=1.53.1", features = ["test-util"] } [patch.crates-io] -agentkit-loop = { git = "https://github.com/danielkov/agentkit.git", rev = "0199d04c4d92bf858f4f4895b9f835c755201ea5" } -agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "0199d04c4d92bf858f4f4895b9f835c755201ea5" } +agentkit-loop = { git = "https://github.com/danielkov/agentkit.git", rev = "6480d909800f25a5ddf042cf3f6791bf6726fe41" } +agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "6480d909800f25a5ddf042cf3f6791bf6726fe41" } agent-client-protocol = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" } agent-client-protocol-http = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" } From 7ebc91b85298813f08112517f680504e9e8a558c Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 2 Sep 2026 14:38:36 +0100 Subject: [PATCH 4/4] chore(deps): bound interrupted response buffering --- Cargo.lock | 4 ++-- Cargo.toml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 021231a5..8a609331 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,7 +134,7 @@ dependencies = [ [[package]] name = "agentkit-acp" version = "0.10.11" -source = "git+https://github.com/danielkov/agentkit.git?rev=6480d909800f25a5ddf042cf3f6791bf6726fe41#6480d909800f25a5ddf042cf3f6791bf6726fe41" +source = "git+https://github.com/danielkov/agentkit.git?rev=7604f6c2ba1eed1de83fe3d9a0058192b5623bb4#7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" dependencies = [ "agent-client-protocol", "agentkit-core", @@ -241,7 +241,7 @@ dependencies = [ [[package]] name = "agentkit-loop" version = "0.10.11" -source = "git+https://github.com/danielkov/agentkit.git?rev=6480d909800f25a5ddf042cf3f6791bf6726fe41#6480d909800f25a5ddf042cf3f6791bf6726fe41" +source = "git+https://github.com/danielkov/agentkit.git?rev=7604f6c2ba1eed1de83fe3d9a0058192b5623bb4#7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" dependencies = [ "agentkit-core", "agentkit-task-manager", diff --git a/Cargo.toml b/Cargo.toml index 266aa526..56dd6eeb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -88,8 +88,8 @@ tempfile = "=3.27.0" tokio = { version = "=1.53.1", features = ["test-util"] } [patch.crates-io] -agentkit-loop = { git = "https://github.com/danielkov/agentkit.git", rev = "6480d909800f25a5ddf042cf3f6791bf6726fe41" } -agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "6480d909800f25a5ddf042cf3f6791bf6726fe41" } +agentkit-loop = { git = "https://github.com/danielkov/agentkit.git", rev = "7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" } +agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" } agent-client-protocol = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" } agent-client-protocol-http = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" }