From 569efafa8803f6320559d50f585383ba640f1ecf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Wed, 8 Jul 2026 15:17:22 +0800 Subject: [PATCH 1/4] feat(model/cosyvoice):support cancel in finish-task --- .../audio/ttsv2/SpeechSynthesizer.java | 26 ++++++++++++++----- .../audio/ttsv2/SpeechSynthesizerV2.java | 6 ++++- .../base/FullDuplexServiceParam.java | 12 +++++++++ .../dashscope/protocol/FullDuplexRequest.java | 5 +++- 4 files changed, 40 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java index 9a856ce7..bcd53688 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java @@ -239,7 +239,7 @@ public Flowable streamingCallAsFlowable(Flowable return duplexApi .duplexCall( StreamInputTtsParamWithStream.fromStreamInputTtsParam( - this.parameters, inputStream, preRequestId, false)) + this.parameters, inputStream, preRequestId, false, this.canceled)) .filter(item -> item.getEvent() != WebSocketEventType.TASK_STARTED.getValue()) .map(SpeechSynthesisResult::fromDashScopeResult) .filter(item -> !canceled.get()) @@ -292,7 +292,8 @@ public Flowable callAsFlowable(String text) }, BackpressureStrategy.BUFFER), preRequestId, - true)) + true, + this.canceled)) .filter(item -> item.getEvent() != WebSocketEventType.TASK_STARTED.getValue()) .map(SpeechSynthesisResult::fromDashScopeResult) .doOnNext( @@ -368,7 +369,7 @@ private void startStream(boolean enableSsml) { try { duplexApi.duplexCall( SpeechSynthesizer.StreamInputTtsParamWithStream.fromStreamInputTtsParam( - this.parameters, textFrames, preRequestId, enableSsml), + this.parameters, textFrames, preRequestId, enableSsml, this.canceled), new ResultCallback() { // private Sentence lastSentence = null; @@ -441,9 +442,6 @@ public void onEvent(DashScopeResult message) { @Override public void onComplete() { log.debug("[TtsV2] onComplete"); - if (canceled.get()) { - return; - } synchronized (SpeechSynthesizer.this) { state = SpeechSynthesisState.IDLE; } @@ -802,11 +800,19 @@ private static class StreamInputTtsParamWithStream extends SpeechSynthesisParam @NonNull private Flowable textStream; + /** + * Shared reference to the outer {@link SpeechSynthesizer}'s canceled flag. When set to true + * before the finish-task message is sent, the finish-task message will carry + * payload.input.directive="cancel" to notify the server to discard remaining audio. + */ + private AtomicBoolean canceled; + public static StreamInputTtsParamWithStream fromStreamInputTtsParam( SpeechSynthesisParam param, Flowable textStream, String preRequestId, - boolean enableSsml) { + boolean enableSsml, + AtomicBoolean canceled) { return StreamInputTtsParamWithStream.builder() .headers(param.getHeaders()) .parameters(param.getParameters()) @@ -817,9 +823,15 @@ public static StreamInputTtsParamWithStream fromStreamInputTtsParam( .model(param.getModel()) .voice(param.getVoice()) .apiKey(param.getApiKey()) + .canceled(canceled) .build(); } + @Override + public String getDirective() { + return (canceled != null && canceled.get()) ? "cancel" : null; + } + @Override public Flowable getStreamingData() { return textStream diff --git a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java index 94acb336..0b8c6862 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java @@ -212,7 +212,11 @@ public void sendText(String text) { } public void stopSynthesizer() { - sendTaskMessage("finish-task", new JsonObject()); + JsonObject input = new JsonObject(); + if (canceled.get()) { + input.addProperty("directive", "cancel"); + } + sendTaskMessage("finish-task", input); } @Override diff --git a/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java b/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java index 345a9cdf..5442be8b 100644 --- a/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java +++ b/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java @@ -45,4 +45,16 @@ public void putHeader(String key, String value) { headers.put(key, value); } } + + /** + * The directive to be carried in payload.input.directive when sending the finish-task message. + * Subclasses can override this method to instruct the server with special semantics (e.g. + * "cancel") when the finish-task message is sent. Returns null by default, meaning no directive + * will be added. + * + * @return the directive string, or null if none. + */ + public String getDirective() { + return null; + } } diff --git a/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java b/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java index c01dce08..11dae0c2 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java +++ b/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java @@ -172,7 +172,10 @@ public JsonObject getFinishedTaskMessage(String taskId) { wsMessage.add(ApiKeywords.HEADER, header); JsonObject payload = new JsonObject(); JsonObject input = new JsonObject(); - if (serviceOption.getTask().equals("multimodal-generation")) { + String directive = param.getDirective(); + if (directive != null) { + input.addProperty("directive", directive); + } else if (serviceOption.getTask().equals("multimodal-generation")) { input.addProperty("directive", "Stop"); } payload.add("input", input); From 9cc541cdcd12bbde3af482b4257ef1dab882c33e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Wed, 8 Jul 2026 16:10:34 +0800 Subject: [PATCH 2/4] fix(model/omni): add onError() in callback --- .../dashscope/audio/omni/OmniRealtimeCallback.java | 9 +++++++++ .../audio/omni/OmniRealtimeConversation.java | 14 ++++++++++++-- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java index f63950ac..9b375c97 100644 --- a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java +++ b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java @@ -17,4 +17,13 @@ public void onOpen() {} /** Will be called when the connection is closed. */ public abstract void onClose(int code, String reason); + + /** + * Will be called when the websocket connection fails, e.g. network disconnected, connect timeout, + * or other io exceptions. Default implementation does nothing, override it to be notified of such + * errors. + * + * @param throwable the exception that caused the failure. + */ + public void onError(Throwable throwable) {} } diff --git a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java index 2b2dcfd6..2a099a19 100644 --- a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java +++ b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java @@ -417,8 +417,18 @@ public void onClosed(WebSocket webSocket, int code, String reason) { @Override public void onFailure(WebSocket webSocket, Throwable t, Response response) { - connectLatch.get().countDown(); - log.error("WebSocket failed: " + t.getMessage()); + isOpen.set(false); + isClosed.set(true); + CountDownLatch cLatch = connectLatch.get(); + if (cLatch != null) { + cLatch.countDown(); + } + CountDownLatch dLatch = disconnectLatch.get(); + if (dLatch != null) { + dLatch.countDown(); + } + log.error("WebSocket failed: " + t.getMessage(), t); + callback.onError(t); } @Override From ad6b3c331f9071c936e1bb098337df45a826a0f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Wed, 8 Jul 2026 16:30:13 +0800 Subject: [PATCH 3/4] feat(agent/multimodal-dialog): support payload.parameters when requestToRespond --- .../audio/omni/OmniRealtimeConversation.java | 8 ++++++-- .../dashscope/audio/ttsv2/SpeechSynthesizerV2.java | 1 + .../dashscope/multimodal/MultiModalDialog.java | 6 ++++++ .../multimodal/MultiModalRequestParam.java | 13 +++++++++++++ 4 files changed, 26 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java index 2a099a19..d9258813 100644 --- a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java +++ b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java @@ -350,7 +350,9 @@ private void sendMessage(ByteString message) { @Override public void onOpen(WebSocket webSocket, Response response) { isOpen.set(true); - connectLatch.get().countDown(); + if (connectLatch != null && connectLatch.get().getCount() > 0) { + connectLatch.get().countDown(); + } log.debug("WebSocket opened"); callback.onOpen(); } @@ -410,7 +412,9 @@ public void onMessage(WebSocket webSocket, String text) { @Override public void onClosed(WebSocket webSocket, int code, String reason) { isOpen.set(false); - connectLatch.get().countDown(); + if (connectLatch != null && connectLatch.get().getCount() > 0) { + connectLatch.get().countDown(); + } log.debug("WebSocket closed: " + code + ", " + reason); callback.onClose(code, reason); } diff --git a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java index 0b8c6862..1837a9a2 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java @@ -417,6 +417,7 @@ private void startStream(boolean enableSsml) throws NoApiKeyException, Interrupt connect(); } else { startStreamTimeStamp = System.currentTimeMillis(); + canceled.set(false); } checkConnectStatus(); // check websocket connection， if socket is closed. diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java index 4672c26c..4043d824 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java @@ -411,6 +411,9 @@ public void requestToRespond( if (updateParams != null && updateParams.bizParams != null) { requestParamWithStream.setBizParams(updateParams.bizParams); } + if (updateParams != null && updateParams.parameters != null) { + requestParamWithStream.setExtraParameters(updateParams.parameters); + } sendTextFrame("RequestToRespond"); } @@ -442,6 +445,9 @@ public void updateInfo(MultiModalRequestParam.UpdateParams updateParams) { if (updateParams != null && updateParams.downStream != null) { requestParamWithStream.setDownStream(updateParams.downStream); } + if (updateParams != null && updateParams.parameters != null) { + requestParamWithStream.setExtraParameters(updateParams.parameters); + } sendTextFrame("UpdateInfo"); } diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java index fa75ebf0..1909b2e3 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java @@ -25,6 +25,13 @@ public class MultiModalRequestParam extends FullDuplexServiceParam { private CustomInput customInput; private List images; private String taskId; + /** + * Extra custom payload.parameters entries, mutable at runtime (unlike the immutable {@code + * parameters} field inherited from {@link FullDuplexServiceParam}). Used by {@code + * requestToRespond}/{@code updateInfo} to let callers pass arbitrary key/value pairs into + * payload.parameters for a specific request. + */ + private Map extraParameters; @Builder public static class CustomInput { @@ -142,6 +149,7 @@ public void clearParameters() { customInput = null; images = null; dialogAttributes = null; + extraParameters = null; } @Builder @@ -151,6 +159,8 @@ public static class UpdateParams { List images; BizParams bizParams; ClientInfo clientInfo; + /** Custom key/value pairs merged directly into payload.parameters. */ + Map parameters; } @Override @@ -263,6 +273,9 @@ public Map getParameters() { if (this.parameters != null) { params.putAll(this.parameters); } + if (this.extraParameters != null) { + params.putAll(this.extraParameters); + } return params; } From 085deee78efaf3657584f7142cebe05247f1359c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Wed, 15 Jul 2026 11:33:15 +0800 Subject: [PATCH 4/4] fix: potential nullpointer exception and synchronization race conditions --- .../audio/omni/OmniRealtimeConversation.java | 10 +- .../multimodal/MultiModalDialog.java | 99 +++++++++++-------- 2 files changed, 62 insertions(+), 47 deletions(-) diff --git a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java index d9258813..53ab7ba3 100644 --- a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java +++ b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java @@ -350,8 +350,9 @@ private void sendMessage(ByteString message) { @Override public void onOpen(WebSocket webSocket, Response response) { isOpen.set(true); - if (connectLatch != null && connectLatch.get().getCount() > 0) { - connectLatch.get().countDown(); + CountDownLatch latch = connectLatch.get(); + if (latch != null && latch.getCount() > 0) { + latch.countDown(); } log.debug("WebSocket opened"); callback.onOpen(); @@ -412,8 +413,9 @@ public void onMessage(WebSocket webSocket, String text) { @Override public void onClosed(WebSocket webSocket, int code, String reason) { isOpen.set(false); - if (connectLatch != null && connectLatch.get().getCount() > 0) { - connectLatch.get().countDown(); + CountDownLatch latch = connectLatch.get(); + if (latch != null && latch.getCount() > 0) { + latch.countDown(); } log.debug("WebSocket closed: " + code + ", " + reason); callback.onClose(code, reason); diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java index 4043d824..2e119db8 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java @@ -396,25 +396,33 @@ public void sendHeartBeat() { */ public void requestToRespond( String type, String text, MultiModalRequestParam.UpdateParams updateParams) { - requestParamWithStream.clearParameters(); - MultiModalRequestParam.CustomInput customInput = - MultiModalRequestParam.CustomInput.builder() - .directive("RequestToRespond") - .dialogId(currentDialogId) - .type(type) - .text(text) - .build(); - requestParamWithStream.setCustomInput(customInput); - if (updateParams != null && updateParams.images != null) { - requestParamWithStream.setImages(updateParams.images); - } - if (updateParams != null && updateParams.bizParams != null) { - requestParamWithStream.setBizParams(updateParams.bizParams); - } - if (updateParams != null && updateParams.parameters != null) { - requestParamWithStream.setExtraParameters(updateParams.parameters); + // Synchronize the whole "mutate requestParamWithStream + send" sequence on the same + // monitor used by sendTextFrame/sendAudioData/sendFinishTaskMessage. Without this, + // concurrent calls to requestToRespond/updateInfo could interleave their + // clearParameters()/setXxx() mutations before either one reaches sendTextFrame, + // causing one caller's payload to be sent with another caller's data (race condition). + // synchronized is reentrant, so the nested lock acquired inside sendTextFrame is safe. + synchronized (MultiModalDialog.this) { + requestParamWithStream.clearParameters(); + MultiModalRequestParam.CustomInput customInput = + MultiModalRequestParam.CustomInput.builder() + .directive("RequestToRespond") + .dialogId(currentDialogId) + .type(type) + .text(text) + .build(); + requestParamWithStream.setCustomInput(customInput); + if (updateParams != null && updateParams.images != null) { + requestParamWithStream.setImages(updateParams.images); + } + if (updateParams != null && updateParams.bizParams != null) { + requestParamWithStream.setBizParams(updateParams.bizParams); + } + if (updateParams != null && updateParams.parameters != null) { + requestParamWithStream.setExtraParameters(updateParams.parameters); + } + sendTextFrame("RequestToRespond"); } - sendTextFrame("RequestToRespond"); } /** @@ -423,32 +431,37 @@ public void requestToRespond( *

param: updateParams Update parameters */ public void updateInfo(MultiModalRequestParam.UpdateParams updateParams) { - requestParamWithStream.clearParameters(); - MultiModalRequestParam.CustomInput customInput = - MultiModalRequestParam.CustomInput.builder() - .directive("UpdateInfo") - .dialogId(currentDialogId) - .build(); - requestParamWithStream.setCustomInput(customInput); - if (updateParams != null && updateParams.clientInfo != null) { - requestParamWithStream.setClientInfo(updateParams.clientInfo); - } - if (updateParams != null && updateParams.bizParams != null) { - requestParamWithStream.setBizParams(updateParams.bizParams); - } - if (updateParams != null && updateParams.images != null) { - requestParamWithStream.setImages(updateParams.images); - } - if (updateParams != null && updateParams.upStream != null) { - requestParamWithStream.setUpStream(updateParams.upStream); - } - if (updateParams != null && updateParams.downStream != null) { - requestParamWithStream.setDownStream(updateParams.downStream); - } - if (updateParams != null && updateParams.parameters != null) { - requestParamWithStream.setExtraParameters(updateParams.parameters); + // See requestToRespond() for why this whole block must be synchronized on the same + // monitor as sendTextFrame: mutation of requestParamWithStream and the subsequent send + // must be atomic with respect to concurrent requestToRespond/updateInfo callers. + synchronized (MultiModalDialog.this) { + requestParamWithStream.clearParameters(); + MultiModalRequestParam.CustomInput customInput = + MultiModalRequestParam.CustomInput.builder() + .directive("UpdateInfo") + .dialogId(currentDialogId) + .build(); + requestParamWithStream.setCustomInput(customInput); + if (updateParams != null && updateParams.clientInfo != null) { + requestParamWithStream.setClientInfo(updateParams.clientInfo); + } + if (updateParams != null && updateParams.bizParams != null) { + requestParamWithStream.setBizParams(updateParams.bizParams); + } + if (updateParams != null && updateParams.images != null) { + requestParamWithStream.setImages(updateParams.images); + } + if (updateParams != null && updateParams.upStream != null) { + requestParamWithStream.setUpStream(updateParams.upStream); + } + if (updateParams != null && updateParams.downStream != null) { + requestParamWithStream.setDownStream(updateParams.downStream); + } + if (updateParams != null && updateParams.parameters != null) { + requestParamWithStream.setExtraParameters(updateParams.parameters); + } + sendTextFrame("UpdateInfo"); } - sendTextFrame("UpdateInfo"); } /**