fix(external_websocket_sync): force-reset claude-agent-acp on wedged Query

crates/acp_thread/src/connection.rs

@@ -134,6 +134,33 @@ pub trait AgentConnection {
         Task::ready(Err(anyhow::Error::msg("Closing sessions is not supported")))
     }
 
+    /// Force-close a session at the wire level, bypassing any ref counting.
+    /// Sends `CloseSessionRequest` to the agent immediately and drops the
+    /// session from internal bookkeeping so a follow-up `load_session`
+    /// recreates it cleanly.
+    ///
+    /// Necessary for recovery from a wedged downstream session (e.g. the
+    /// Anthropic Claude Code SDK `Query` gets stuck after a mid-stream
+    /// cancellation; only the wrapper's `teardownSession` unwedges it,
+    /// and we cannot reach it through `close_session` because the
+    /// `AcpThread` entity still holds a reference). Callers must ensure
+    /// the consuming entity is dropped or replaced before re-using the
+    /// session id; otherwise events for the closed session will be
+    /// orphaned.
+    fn force_close_session(
+        self: Rc<Self>,
+        _session_id: &acp::SessionId,
+        _cx: &mut App,
+    ) -> Task<Result<()>> {
+        // Default implementation is intentionally an error — only ACP-style
+        // wrapper connections (e.g. AcpConnection) need force-close for
+        // wedged-Query recovery. Implementations that route via other agent
+        // backends should explicitly override if recovery applies.
+        Task::ready(Err(anyhow::Error::msg(
+            "Force-closing sessions is not supported by this agent backend",
+        )))
+    }
+
     /// Whether this agent supports resuming existing sessions without loading history.
     fn supports_resume_session(&self) -> bool {
         false

crates/agent_servers/src/acp.rs

@@ -1919,6 +1919,32 @@ impl AgentConnection for AcpConnection {
         })
     }
 
+    fn force_close_session(
+        self: Rc<Self>,
+        session_id: &acp::SessionId,
+        cx: &mut App,
+    ) -> Task<Result<()>> {
+        if !self.supports_close_session() {
+            return Task::ready(Err(anyhow!(LoadError::Other(
+                "Force-closing sessions is not supported by this agent.".into()
+            ))));
+        }
+
+        // Unconditionally drop our session bookkeeping so a follow-up
+        // `load_session` cannot accidentally return a stale handle while
+        // the wrapper finishes its teardown. Ignored if the entry is gone.
+        self.pending_sessions.borrow_mut().remove(session_id);
+        self.sessions.borrow_mut().remove(session_id);
+
+        let conn = self.connection.clone();
+        let session_id = session_id.clone();
+        cx.foreground_executor().spawn(async move {
+            into_foreground_future(conn.send_request(acp::CloseSessionRequest::new(session_id)))
+                .await?;
+            Ok(())
+        })
+    }
+
     fn auth_methods(&self) -> &[acp::AuthMethod] {
         &self.auth_methods
     }
@@ -2292,6 +2318,14 @@ pub mod test_support {
             self.inner.clone().close_session(session_id, cx)
         }
 
+        fn force_close_session(
+            self: Rc<Self>,
+            session_id: &acp::SessionId,
+            cx: &mut App,
+        ) -> Task<Result<()>> {
+            self.inner.clone().force_close_session(session_id, cx)
+        }
+
         fn supports_resume_session(&self) -> bool {
             self.inner.supports_resume_session()
         }

crates/agent_servers/src/claude.rs

@@ -256,3 +256,4 @@ impl AgentServer for ClaudeCode {
         self
     }
 }
+// force recheck 1781532536