Enhance task handling and error management in models

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) {}
 }

src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java

@@ -350,7 +350,10 @@ private void sendMessage(ByteString message) {
   @Override
   public void onOpen(WebSocket webSocket, Response response) {
     isOpen.set(true);
-    connectLatch.get().countDown();
+    CountDownLatch latch = connectLatch.get();
+    if (latch != null && latch.getCount() > 0) {
+      latch.countDown();
+    }
     log.debug("WebSocket opened");
     callback.onOpen();
   }
@@ -410,15 +413,28 @@ public void onMessage(WebSocket webSocket, String text) {
   @Override
   public void onClosed(WebSocket webSocket, int code, String reason) {
     isOpen.set(false);
-    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);
   }
 
   @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

src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java

@@ -239,7 +239,7 @@ public Flowable<SpeechSynthesisResult> streamingCallAsFlowable(Flowable<String>
     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<SpeechSynthesisResult> 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<DashScopeResult>() {
             //                        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<TextStreamItem> 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<TextStreamItem> 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<Object> getStreamingData() {
       return textStream

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
@@ -413,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.

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;
+  }
 }

src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java

@@ -396,22 +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);
+    // 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");
   }
 
   /**
@@ -420,29 +431,37 @@ public void requestToRespond(
    * <p>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);
+    // 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");
   }
 
   /**

src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java

@@ -25,6 +25,13 @@ public class MultiModalRequestParam extends FullDuplexServiceParam {
   private CustomInput customInput;
   private List<Object> 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<String, Object> 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<Object> images;
     BizParams bizParams;
     ClientInfo clientInfo;
+    /** Custom key/value pairs merged directly into payload.parameters. */
+    Map<String, Object> parameters;
   }
 
   @Override
@@ -263,6 +273,9 @@ public Map<String, Object> getParameters() {
     if (this.parameters != null) {
       params.putAll(this.parameters);
     }
+    if (this.extraParameters != null) {
+      params.putAll(this.extraParameters);
+    }
     return params;
   }
 

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);