fix(acp): spawn prompt turn to prevent permission-gate deadlock

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
 

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<Notify>` —
@@ -19,19 +27,36 @@ use crate::agent::ZephAcpAgentState;
 pub(crate) async fn handle_prompt(
     req: acp::schema::v1::PromptRequest,
     responder: acp::Responder<acp::schema::v1::PromptResponse>,
-    #[cfg_attr(not(feature = "unstable-cancel-request"), allow(unused_variables))]
     cx: acp::ConnectionTo<acp::Client>,
     state: Arc<ZephAcpAgentState>,
 ) -> 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

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<bool>) -> 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::<bool>();
+            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::<bool>();
+            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;
+}