diff --git a/CHANGELOG.md b/CHANGELOG.md index c1aafe0c9..96602f651 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,18 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). bar; a second `Ctrl+C` within that window quits, a later press re-arms instead. `q` and `/quit` are unaffected. +### Fixed + +- `zeph-acp`: fixed a deadlock on every permission-gated tool call (issue #6656). `handle_prompt` + awaited the entire agent turn inline inside the ACP SDK's `on_receive_request` callback, holding + the SDK's strictly serial dispatch loop for the whole turn; since that same loop demultiplexes + inbound RPC responses, the IDE's reply to a permission-gated tool call's + `session/request_permission` request could never route back while the loop was blocked on the + still-running turn, hanging the tool call indefinitely (fail-closed, not a permission bypass). + The turn is now spawned via `cx.spawn` instead of awaited inline, per the SDK's documented + ordering contract, freeing the dispatch loop to keep routing inbound messages — including the + permission reply — while the turn runs. + ## [0.22.3] - 2026-07-22 ### Fixed diff --git a/crates/zeph-acp/src/agent/handlers/prompt.rs b/crates/zeph-acp/src/agent/handlers/prompt.rs index a2f375cde..233dd9a0e 100644 --- a/crates/zeph-acp/src/agent/handlers/prompt.rs +++ b/crates/zeph-acp/src/agent/handlers/prompt.rs @@ -11,6 +11,14 @@ use crate::agent::ZephAcpAgentState; /// Handle an ACP `prompt` request. /// +/// `do_prompt` runs a whole agent turn, which for permission-gated tool calls awaits the IDE's +/// reply to a `session/request_permission` request routed back through this same connection's +/// dispatch loop. Per the SDK's ordering contract (`agent_client_protocol::concepts::ordering`), +/// `on_receive_request` callbacks hold that loop until they return — awaiting `do_prompt` inline +/// here would deadlock the turn against its own permission response. `cx.spawn` escapes the loop +/// so it stays free to route the permission reply (and any other inbound traffic) while the turn +/// runs; the response is sent from inside the spawned task once the turn completes (#6656). +/// /// When the `unstable-cancel-request` feature is enabled, this bridges the real ACP /// `$/cancel_request` protocol notification (scoped to this specific JSON-RPC request via /// [`acp::Responder::cancellation`]) onto the session's existing `cancel_signal: Arc` — @@ -19,19 +27,36 @@ use crate::agent::ZephAcpAgentState; pub(crate) async fn handle_prompt( req: acp::schema::v1::PromptRequest, responder: acp::Responder, - #[cfg_attr(not(feature = "unstable-cancel-request"), allow(unused_variables))] cx: acp::ConnectionTo, state: Arc, ) -> acp::Result<()> { #[cfg(feature = "unstable-cancel-request")] - let cancel_request_bridge = spawn_cancel_request_bridge(&req, &responder, &cx, &state); + let bridge_cx = cx.clone(); - let resp = state.do_prompt(req).await?; + cx.spawn(async move { + #[cfg(feature = "unstable-cancel-request")] + let cancel_request_bridge = + spawn_cancel_request_bridge(&req, &responder, &bridge_cx, &state); - #[cfg(feature = "unstable-cancel-request")] - drop(cancel_request_bridge); + let result = state.do_prompt(req).await; - responder.respond(resp) + #[cfg(feature = "unstable-cancel-request")] + drop(cancel_request_bridge); + + // Infallible by construction: a `respond`/`respond_with_error` failure means the + // connection is already going away (e.g. the client disconnected mid-turn), not a + // problem with this prompt — log and swallow instead of returning `Err`, since per + // `ConnectionTo::spawn`'s contract, an `Err` returned from this future would tear down + // the *entire* connection over what is otherwise a benign, unrelated disconnect. + let send_result = match result { + Ok(resp) => responder.respond(resp), + Err(e) => responder.respond_with_error(e), + }; + if let Err(e) = send_result { + tracing::debug!(error = %e, "failed to send session/prompt response"); + } + Ok(()) + }) } /// Spawn a watcher that notifies `entry.cancel_signal` if the IDE sends `$/cancel_request` for diff --git a/crates/zeph-acp/tests/integration.rs b/crates/zeph-acp/tests/integration.rs index 4b24ec849..2327c0e7c 100644 --- a/crates/zeph-acp/tests/integration.rs +++ b/crates/zeph-acp/tests/integration.rs @@ -67,6 +67,37 @@ fn text_chunks_spawner(chunks: Vec<&'static str>) -> AgentSpawner { }) } +/// Spawner that requests tool-call permission via `AcpContext::permission_gate` before +/// completing the turn — reproduces the `session/request_permission` round-trip that +/// deadlocked before #6656 was fixed: the server's request-dispatch loop was blocked +/// awaiting `do_prompt` inline, so it could never route the client's permission reply +/// back to the pending `check_permission` future. +/// +/// The gate's decision is forwarded on `decision_tx` so tests can assert on it directly +/// without depending on `PromptResponse` carrying assembled chunk text. +fn permission_gated_spawner(decision_tx: tokio::sync::mpsc::UnboundedSender) -> AgentSpawner { + Arc::new(move |mut channel, ctx, session| { + let decision_tx = decision_tx.clone(); + Box::pin(async move { + let _ = channel.recv().await; + let gate = ctx + .expect("AcpContext must be present") + .permission_gate + .expect("permission gate must be present"); + let tool_call = acp::schema::v1::ToolCallUpdate::new( + "tc-perm-1".to_owned(), + acp::schema::v1::ToolCallUpdateFields::new().title("shell_execute".to_owned()), + ); + let allowed = gate + .check_permission(session.session_id, tool_call) + .await + .unwrap_or(false); + let _ = decision_tx.send(allowed); + let _ = channel.flush_chunks().await; + }) + }) +} + /// Minimal server config for tests. fn test_config(name: &str) -> AcpServerConfig { AcpServerConfig { @@ -2333,3 +2364,187 @@ async fn fork_session_cross_owner_fails() { }) .await; } + +/// #6656 regression: a permission-gated tool call must not deadlock the `session/prompt` +/// round-trip. Before the fix, `handle_prompt` awaited `do_prompt` inline inside the ACP SDK's +/// serial request-dispatch loop; the same loop demultiplexes the client's reply to the +/// `session/request_permission` request sent by `AcpPermissionGate::check_permission`, so the +/// loop deadlocked on itself. Wrapped in a timeout so a regression fails this test fast instead +/// of hanging the suite. +#[tokio::test(flavor = "current_thread")] +#[allow(clippy::large_futures)] +async fn permission_gated_prompt_round_trip_does_not_deadlock() { + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + let workdir = temp_workdir(); + let (sw, sr, cw, cr) = duplex_pair(); + let (decision_tx, mut decision_rx) = tokio::sync::mpsc::unbounded_channel::(); + let server_fut = serve_connection( + permission_gated_spawner(decision_tx), + test_config("test-agent"), + sw, + sr, + "acp-local".to_owned(), + ); + let client_fut = acp::Client + .builder() + .on_receive_request( + async |_req: acp::schema::v1::RequestPermissionRequest, + responder: acp::Responder< + acp::schema::v1::RequestPermissionResponse, + >, + _cx| { + responder.respond(acp::schema::v1::RequestPermissionResponse::new( + acp::schema::v1::RequestPermissionOutcome::Selected( + acp::schema::v1::SelectedPermissionOutcome::new("allow_once"), + ), + )) + }, + acp::on_receive_request!(), + ) + .connect_with(acp::ByteStreams::new(cw, cr), async |cx| { + cx.send_request(acp::schema::v1::InitializeRequest::new( + acp::schema::ProtocolVersion::LATEST, + )) + .block_task() + .await?; + + let session_id = cx + .send_request(acp::schema::v1::NewSessionRequest::new(workdir.path())) + .block_task() + .await? + .session_id; + + let content = vec![acp::schema::v1::ContentBlock::Text( + acp::schema::v1::TextContent::new("run a gated tool"), + )]; + let resp = cx + .send_request(acp::schema::v1::PromptRequest::new(session_id, content)) + .block_task() + .await?; + + assert_eq!( + resp.stop_reason, + acp::schema::v1::StopReason::EndTurn, + "expected EndTurn, got {:?}", + resp.stop_reason, + ); + Ok(()) + }); + + let outcome = tokio::time::timeout(std::time::Duration::from_secs(10), async { + tokio::select! { + res = server_fut => panic!("server exited before client: {res:?}"), + result = client_fut => result, + } + }) + .await; + + let result = outcome.expect( + "permission-gated prompt round-trip timed out — likely a #6656 deadlock regression", + ); + assert!(result.is_ok(), "prompt round-trip failed: {result:?}"); + + let decision = decision_rx + .recv() + .await + .expect("spawner must report a permission decision"); + assert!( + decision, + "IDE selected allow_once, expected the gate to allow" + ); + }) + .await; +} + +/// #6656 regression, denial path: same round-trip as +/// `permission_gated_prompt_round_trip_does_not_deadlock`, but the IDE rejects the tool call. +/// Confirms fail-closed behavior still completes correctly through the new `cx.spawn`-based +/// dispatch path instead of also deadlocking. +#[tokio::test(flavor = "current_thread")] +#[allow(clippy::large_futures)] +async fn permission_gated_prompt_denial_does_not_deadlock() { + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + let workdir = temp_workdir(); + let (sw, sr, cw, cr) = duplex_pair(); + let (decision_tx, mut decision_rx) = tokio::sync::mpsc::unbounded_channel::(); + let server_fut = serve_connection( + permission_gated_spawner(decision_tx), + test_config("test-agent"), + sw, + sr, + "acp-local".to_owned(), + ); + let client_fut = acp::Client + .builder() + .on_receive_request( + async |_req: acp::schema::v1::RequestPermissionRequest, + responder: acp::Responder< + acp::schema::v1::RequestPermissionResponse, + >, + _cx| { + responder.respond(acp::schema::v1::RequestPermissionResponse::new( + acp::schema::v1::RequestPermissionOutcome::Selected( + acp::schema::v1::SelectedPermissionOutcome::new("reject_once"), + ), + )) + }, + acp::on_receive_request!(), + ) + .connect_with(acp::ByteStreams::new(cw, cr), async |cx| { + cx.send_request(acp::schema::v1::InitializeRequest::new( + acp::schema::ProtocolVersion::LATEST, + )) + .block_task() + .await?; + + let session_id = cx + .send_request(acp::schema::v1::NewSessionRequest::new(workdir.path())) + .block_task() + .await? + .session_id; + + let content = vec![acp::schema::v1::ContentBlock::Text( + acp::schema::v1::TextContent::new("run a gated tool"), + )]; + let resp = cx + .send_request(acp::schema::v1::PromptRequest::new(session_id, content)) + .block_task() + .await?; + + assert_eq!( + resp.stop_reason, + acp::schema::v1::StopReason::EndTurn, + "expected EndTurn, got {:?}", + resp.stop_reason, + ); + Ok(()) + }); + + let outcome = tokio::time::timeout(std::time::Duration::from_secs(10), async { + tokio::select! { + res = server_fut => panic!("server exited before client: {res:?}"), + result = client_fut => result, + } + }) + .await; + + let result = outcome.expect( + "permission-gated prompt round-trip timed out — likely a #6656 deadlock regression", + ); + assert!(result.is_ok(), "prompt round-trip failed: {result:?}"); + + let decision = decision_rx + .recv() + .await + .expect("spawner must report a permission decision"); + assert!( + !decision, + "IDE selected reject_once, expected the gate to deny" + ); + }) + .await; +}